third_codes/PowerJob
1
0
Fork 1
mirror of https://github.com/PowerJob/PowerJob.git synced 2025-07-17 00:00:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: use gatling to have a presure test for remote framework

Browse Source
This commit is contained in:
tjq 2023-01-08 21:22:56 +08:00
parent 0d29b6369a
commit 2afb20df0b
4 changed files with 14 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
powerjob-remote/powerjob-remote-benchmark/src/main/java/tech/powerjob/remote/benchmark/PressureTestController.java
View File

@ -36,12 +36,15 @@ public class PressureTestController {
@GetMapping("/tell") @GetMapping("/tell")
public void httpTell(String protocol, Integer blockMs, Integer responseSize, String content) { public void httpTell(String protocol, Integer blockMs, Integer responseSize, String content) {
URL url = new URL().setLocation(HL).setAddress(new Address().setPort(SERVER_HTTP_PORT).setHost(HOST)); Address address = new Address().setHost(HOST);
URL url = new URL().setLocation(HL).setAddress(address);
final BenchmarkActor.BenchmarkRequest request = new BenchmarkActor.BenchmarkRequest().setContent(content).setBlockingMills(blockMs).setResponseSize(responseSize); final BenchmarkActor.BenchmarkRequest request = new BenchmarkActor.BenchmarkRequest().setContent(content).setBlockingMills(blockMs).setResponseSize(responseSize);
try { try {
if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) { if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) {
address.setPort(SERVER_HTTP_PORT);
engineService.getHttpTransporter().tell(url, request); engineService.getHttpTransporter().tell(url, request);
} else { } else {
address.setPort(SERVER_AKKA_PORT);
engineService.getAkkaTransporter().tell(url, request); engineService.getAkkaTransporter().tell(url, request);
} }
} catch (Exception e) { } catch (Exception e) {
@ -53,14 +56,17 @@ public class PressureTestController {
@GetMapping("/ask") @GetMapping("/ask")
public void httpAsk(String protocol, Integer blockMs, Integer responseSize, String content, Boolean debug) { public void httpAsk(String protocol, Integer blockMs, Integer responseSize, String content, Boolean debug) {
URL url = new URL().setLocation(HL).setAddress(new Address().setPort(SERVER_HTTP_PORT).setHost(HOST)); Address address = new Address().setHost(HOST);
URL url = new URL().setLocation(HL).setAddress(address);
final BenchmarkActor.BenchmarkRequest request = new BenchmarkActor.BenchmarkRequest().setContent(content).setBlockingMills(blockMs).setResponseSize(responseSize); final BenchmarkActor.BenchmarkRequest request = new BenchmarkActor.BenchmarkRequest().setContent(content).setBlockingMills(blockMs).setResponseSize(responseSize);
try { try {
CompletionStage<BenchmarkActor.BenchmarkResponse> responseOpt = null; CompletionStage<BenchmarkActor.BenchmarkResponse> responseOpt = null;
if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) { if (Protocol.HTTP.name().equalsIgnoreCase(protocol)) {
address.setPort(SERVER_HTTP_PORT);
responseOpt = engineService.getHttpTransporter().ask(url, request, BenchmarkActor.BenchmarkResponse.class); responseOpt = engineService.getHttpTransporter().ask(url, request, BenchmarkActor.BenchmarkResponse.class);
} else { } else {
address.setPort(SERVER_AKKA_PORT);
responseOpt = engineService.getAkkaTransporter().ask(url, request, BenchmarkActor.BenchmarkResponse.class); responseOpt = engineService.getAkkaTransporter().ask(url, request, BenchmarkActor.BenchmarkResponse.class);
} }
final BenchmarkActor.BenchmarkResponse response = responseOpt.toCompletableFuture().get(); final BenchmarkActor.BenchmarkResponse response = responseOpt.toCompletableFuture().get();

2
powerjob-remote/powerjob-remote-framework/src/main/java/tech/powerjob/remote/framework/engine/EngineOutput.java
View File

@ -1,11 +1,9 @@
package tech.powerjob.remote.framework.engine; package tech.powerjob.remote.framework.engine;
import com.google.common.collect.Maps;
import lombok.Getter; import lombok.Getter;
import lombok.Setter; import lombok.Setter;
import tech.powerjob.remote.framework.transporter.Transporter; import tech.powerjob.remote.framework.transporter.Transporter;
import java.util.Map;
/** /**
* 引擎输出 * 引擎输出

2
powerjob-remote/powerjob-remote-impl-akka/src/main/java/tech/powerjob/remote/akka/AkkaCSInitializer.java
View File

@ -64,6 +64,8 @@ public class AkkaCSInitializer implements CSInitializer {
// 处理系统中产生的异常情况 // 处理系统中产生的异常情况
ActorRef troubleshootingActor = actorSystem.actorOf(Props.create(AkkaTroubleshootingActor.class), "troubleshooting"); ActorRef troubleshootingActor = actorSystem.actorOf(Props.create(AkkaTroubleshootingActor.class), "troubleshooting");
actorSystem.eventStream().subscribe(troubleshootingActor, DeadLetter.class); actorSystem.eventStream().subscribe(troubleshootingActor, DeadLetter.class);
log.info("[PowerJob-AKKA] initialize actorSystem[{}] successfully!", actorSystem.name());
} }
@Override @Override

23
powerjob-remote/powerjob-remote-impl-akka/src/main/java/tech/powerjob/remote/akka/AkkaTransporter.java
View File

@ -27,7 +27,6 @@ import java.util.concurrent.ExecutorService;
*/ */
public class AkkaTransporter implements Transporter { public class AkkaTransporter implements Transporter {
private final ServerType serverType;
private final ActorSystem actorSystem; private final ActorSystem actorSystem;
private final String targetActorSystemName; private final String targetActorSystemName;
@ -37,22 +36,8 @@ public class AkkaTransporter implements Transporter {
*/ */
private static final String AKKA_NODE_PATH = "akka://%s@%s/user/%s"; private static final String AKKA_NODE_PATH = "akka://%s@%s/user/%s";
private static final Map<String, String> SERVER_PATH_MAP = Maps.newHashMap();
private static final Map<String, String> WORKER_PATH_MAP = Maps.newHashMap();
/*
Akka 使用 ActorName + 入参类型 寻址，因此只需要 rootPath
HandlerLocation#rootPathName -> actorName
*/
static {
SERVER_PATH_MAP.put("benchmark", "benchmark");
WORKER_PATH_MAP.put("benchmark", "benchmark");
}
public AkkaTransporter(ServerType serverType, ActorSystem actorSystem) { public AkkaTransporter(ServerType serverType, ActorSystem actorSystem) {
this.actorSystem = actorSystem; this.actorSystem = actorSystem;
this.serverType = serverType;
this.targetActorSystemName = AkkaConstant.fetchActorSystemName(serverType, false); this.targetActorSystemName = AkkaConstant.fetchActorSystemName(serverType, false);
} }
@ -76,12 +61,12 @@ public class AkkaTransporter implements Transporter {
private ActorSelection fetchActorSelection(URL url) { private ActorSelection fetchActorSelection(URL url) {
Map<String, String> rootPath2ActorNameMap = serverType == ServerType.SERVER ? SERVER_PATH_MAP : WORKER_PATH_MAP; String targetActorName = AkkaMappingService.parseActorName(url.getLocation().getRootPath()).getActorName();
final String actorName = rootPath2ActorNameMap.get(url.getLocation().getRootPath());
CommonUtils.requireNonNull(actorName, "can't find actor by URL: " + url.getLocation()); CommonUtils.requireNonNull(targetActorName, "can't find actor by URL: " + url.getLocation());
String address = url.getAddress().toFullAddress(); String address = url.getAddress().toFullAddress();
return actorSystem.actorSelection(String.format(AKKA_NODE_PATH, targetActorSystemName, address, actorName)); return actorSystem.actorSelection(String.format(AKKA_NODE_PATH, targetActorSystemName, address, targetActorName));
} }
} }