From d55068a5bf04310558ef093e89257dc1964cd886 Mon Sep 17 00:00:00 2001 From: cszxyang Date: Wed, 20 Apr 2022 00:14:01 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BC=82=E5=B8=B8=E5=A4=84=E7=90=86=E3=80=81?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E5=A4=84=E7=90=86=E5=8F=8A=E9=83=A8=E5=88=86?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E7=AE=80=E5=8D=95=E7=BE=8E=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../xxl/job/core/executor/XxlJobExecutor.java | 10 +-- .../com/xxl/job/core/server/EmbedServer.java | 64 ++++++++----------- 2 files changed, 33 insertions(+), 41 deletions(-) diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java b/xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java index 70a2dad9..4719b7b7 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java @@ -83,6 +83,7 @@ public class XxlJobExecutor { // init executor-server initEmbedServer(address, ip, port, appname, accessToken); } + public void destroy(){ // destroy executor-server stopEmbedServer(); @@ -131,6 +132,7 @@ public class XxlJobExecutor { } } } + public static List getAdminBizList(){ return adminBizList; } @@ -251,6 +253,7 @@ public class XxlJobExecutor { return newJobThread; } + public static JobThread removeJobThread(int jobId, String removeOldReason){ JobThread oldJobThread = jobThreadRepository.remove(jobId); if (oldJobThread != null) { @@ -261,9 +264,8 @@ public class XxlJobExecutor { } return null; } - public static JobThread loadJobThread(int jobId){ - JobThread jobThread = jobThreadRepository.get(jobId); - return jobThread; - } + public static JobThread loadJobThread(int jobId){ + return jobThreadRepository.get(jobId); + } } diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/server/EmbedServer.java b/xxl-job-core/src/main/java/com/xxl/job/core/server/EmbedServer.java index 242d6b63..540e0ea2 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/server/EmbedServer.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/server/EmbedServer.java @@ -36,10 +36,8 @@ public class EmbedServer { public void start(final String address, final int port, final String appname, final String accessToken) { executorBiz = new ExecutorBizImpl(); thread = new Thread(new Runnable() { - @Override public void run() { - // param EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); @@ -61,8 +59,6 @@ public class EmbedServer { throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!"); } }); - - try { // start server ServerBootstrap bootstrap = new ServerBootstrap(); @@ -92,11 +88,9 @@ public class EmbedServer { future.channel().closeFuture().sync(); } catch (InterruptedException e) { - if (e instanceof InterruptedException) { - logger.info(">>>>>>>>>>> xxl-job remoting server stop."); - } else { - logger.error(">>>>>>>>>>> xxl-job remoting server error.", e); - } + logger.info(">>>>>>>>>>> xxl-job remoting server stop."); + } catch (Exception e) { + logger.error(">>>>>>>>>>> xxl-job remoting server error.", e); } finally { // stop try { @@ -106,17 +100,15 @@ public class EmbedServer { logger.error(e.getMessage(), e); } } - } - }); - thread.setDaemon(true); // daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave + thread.setDaemon(true); // daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave thread.start(); } public void stop() throws Exception { // destroy server thread - if (thread!=null && thread.isAlive()) { + if (thread != null && thread.isAlive()) { thread.interrupt(); } @@ -130,7 +122,7 @@ public class EmbedServer { /** * netty_http - * + *

* Copy from : https://github.com/xuxueli/xxl-rpc * * @author xuxueli 2015-11-24 22:25:15 @@ -141,6 +133,7 @@ public class EmbedServer { private ExecutorBiz executorBiz; private String accessToken; private ThreadPoolExecutor bizThreadPool; + public EmbedHttpServerHandler(ExecutorBiz executorBiz, String accessToken, ThreadPoolExecutor bizThreadPool) { this.executorBiz = executorBiz; this.accessToken = accessToken; @@ -149,7 +142,6 @@ public class EmbedServer { @Override protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception { - // request parse //final byte[] requestBytes = ByteBufUtil.getBytes(msg.content()); // byteBuf.toString(io.netty.util.CharsetUtil.UTF_8); String requestData = msg.content().toString(CharsetUtil.UTF_8); @@ -175,38 +167,38 @@ public class EmbedServer { } private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) { - // valid if (HttpMethod.POST != httpMethod) { return new ReturnT(ReturnT.FAIL_CODE, "invalid request, HttpMethod not support."); } - if (uri==null || uri.trim().length()==0) { + if (uri == null || uri.trim().length() == 0) { return new ReturnT(ReturnT.FAIL_CODE, "invalid request, uri-mapping empty."); } - if (accessToken!=null - && accessToken.trim().length()>0 + if (accessToken != null + && accessToken.trim().length() > 0 && !accessToken.equals(accessTokenReq)) { return new ReturnT(ReturnT.FAIL_CODE, "The access token is wrong."); } // services mapping try { - if ("/beat".equals(uri)) { - return executorBiz.beat(); - } else if ("/idleBeat".equals(uri)) { - IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class); - return executorBiz.idleBeat(idleBeatParam); - } else if ("/run".equals(uri)) { - TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class); - return executorBiz.run(triggerParam); - } else if ("/kill".equals(uri)) { - KillParam killParam = GsonTool.fromJson(requestData, KillParam.class); - return executorBiz.kill(killParam); - } else if ("/log".equals(uri)) { - LogParam logParam = GsonTool.fromJson(requestData, LogParam.class); - return executorBiz.log(logParam); - } else { - return new ReturnT(ReturnT.FAIL_CODE, "invalid request, uri-mapping("+ uri +") not found."); + switch (uri) { + case "/beat": + return executorBiz.beat(); + case "/idleBeat": + IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class); + return executorBiz.idleBeat(idleBeatParam); + case "/run": + TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class); + return executorBiz.run(triggerParam); + case "/kill": + KillParam killParam = GsonTool.fromJson(requestData, KillParam.class); + return executorBiz.kill(killParam); + case "/log": + LogParam logParam = GsonTool.fromJson(requestData, LogParam.class); + return executorBiz.log(logParam); + default: + return new ReturnT(ReturnT.FAIL_CODE, "invalid request, uri-mapping(" + uri + ") not found."); } } catch (Exception e) { logger.error(e.getMessage(), e); @@ -261,6 +253,4 @@ public class EmbedServer { // stop registry ExecutorRegistryThread.getInstance().toStop(); } - - }