改为异步调用，增加插件开发文档

byit-myth-admin/src/main/java/com/byit/controller/JobController.java

@@ -2,6 +2,7 @@ package com.byit.controller;
 
 import com.byit.job.WorkRoulette;
 import com.byit.job.model.JavaBeanJobInfo;
+import com.byit.job.model.ReturnResult;
 import org.springframework.web.bind.annotation.*;
 
 /**
@@ -19,4 +20,9 @@ public class JobController {
         WorkRoulette.addJob(javaBeanJobInfo );
         return "SUCCESS";
     }
+
+    @PostMapping(value = "callbackRes")
+    public void callbackRes(@RequestBody ReturnResult<String> result){
+        System.out.println(result.getCode()+"-----"+result.getMsg());
+    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/job/WorkRoulette.java

@@ -14,6 +14,7 @@ import java.util.concurrent.TimeUnit;
  * @date: 2019/11/15 14:59
  **/
 public class WorkRoulette {
+
     /**
      * hashedWheelTimer:工作轮盘
      * ThreadFactory：创建work线程
@@ -24,7 +25,7 @@ public class WorkRoulette {
     private static final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(new ThreadFactory( ) {
         @Override
         public Thread newThread(Runnable r) {
-            return new Thread(r,"hashedWheelTimer"+r.hashCode());
+            return new Thread(r,"hashedWheelTimer皇甫"+r.hashCode());
         }
     },1, TimeUnit.SECONDS,8,true,0);
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/JavaBeanJobTask.java

@@ -37,6 +37,7 @@ public class JavaBeanJobTask implements TimerTask {
 
             JSONObject jsonObject = new JSONObject();
             jsonObject.put("jobHandelName",jobHandelName);
+            jsonObject.put("callbackMethod","http://127.0.0.1:8080/callbackRes");
             jsonObject.put("param",param);
             String result = HttpUtil.post(url, JSON.toJSONString(jsonObject),10*1000);
             log.info("---------------{}------------",result);

byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/rpc/RunJobServerHandler.java

@@ -11,9 +11,12 @@ import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import io.netty.handler.codec.http.*;
 import io.netty.util.CharsetUtil;
-import io.netty.util.internal.StringUtil;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
 /**
  * @program: byit-myth-job->RunJobServer
  * @description: 处理器最终处理类
@@ -22,14 +25,24 @@ import lombok.extern.slf4j.Slf4j;
  **/
 @Slf4j
 public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
+    /**
+     * LinkedBlockingQueue 不指定容量就变成了无界队列
+     */
+    private static final ThreadPoolExecutor jobTriggerPool = new ThreadPoolExecutor(
+            50,
+            200,
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<Runnable>(1000),
+            r ->new Thread(r, "Netty RunJobServerHandler serverThread-" + r.hashCode()));
+
     private static final String PENG = "PENG";
     private static final String PONG = "PONG";
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
         log.info("----------------------有请求过来了------------------------");
-        String responseBody = "";
+        String heartbeatResponseBody = "FAILURE";
         if(req instanceof HttpRequest){
-
             JSONObject jsonObject = analysisParam(req.content( ));
             if(null == jsonObject){
                 throw new Exception("核心参数 为 null");
@@ -37,22 +50,14 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
             //检测是否有心跳参数，有心跳参数则为测试参数，且为PENG的话，服务端回复  PONG
             String heartbeat = (String)(jsonObject.get("heartbeat"));
             if(null == heartbeat){
-                String jobHandelName = ((String)(jsonObject.get("jobHandelName")));
-                String param = ((String)(jsonObject.get("param")));
-                if(StringUtil.isNullOrEmpty(jobHandelName)){
-                    throw new Exception("jobHandelName 为 null");
-                }
-                ReturnResult<String> stringReturnResult = runJob(jobHandelName, param);
-                assert stringReturnResult != null;
-                responseBody = stringReturnResult.getMsg();
-                log.info("---------服务器端:{},{}-------------",stringReturnResult.getCode(),stringReturnResult.getMsg());
+                jobTriggerPool.execute(new RunJobThread(jsonObject));
+                heartbeatResponseBody = "SUCCESS";
             }else if(PENG.equals(heartbeat)){
                 log.info("----------调度平台心跳检测-------------");
-                responseBody = PONG;
+                heartbeatResponseBody = PONG;
             }
-
             //----------------------------------消息发送-----------------------------------
-            ByteBuf byteBuf = Unpooled.copiedBuffer(responseBody, CharsetUtil.UTF_8);
+            ByteBuf byteBuf = Unpooled.copiedBuffer(heartbeatResponseBody, CharsetUtil.UTF_8);
             //HTTP响应
             FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,byteBuf);
             //设置头信息
@@ -60,24 +65,11 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
             //响应给客户端
             response.headers().set(HttpHeaderNames.CONTENT_LENGTH,byteBuf.readableBytes());
             ctx.writeAndFlush(response);
-
         }
         ctx.close();
 
     }
 
-    private ReturnResult<String> runJob(String jobHandlerName,String param){
-        Class<? extends IJobHandler> jobClass = JobUtils.jobCache.get(jobHandlerName);
-        try {
-            IJobHandler iJobHandler = jobClass.newInstance( );
-            return iJobHandler.execute(param);
-        } catch (Exception e) {
-            e.printStackTrace( );
-        }
-        return null;
-
-    }
-
     /**
      * 格式化参数
      * @param byteBuf
@@ -99,6 +91,36 @@ public class RunJobServerHandler extends SimpleChannelInboundHandler<FullHttpReq
         ctx.close();
     }
 }
+@Slf4j
+class RunJobThread implements Runnable{
+    private JSONObject jsonObject;
+
+    public RunJobThread(JSONObject jsonObject) {
+        this.jsonObject = jsonObject;
+    }
+
+    @Override
+    public void run() {
+        String jobHandelName = ((String)(jsonObject.get("jobHandelName")));
+        String param = ((String)(jsonObject.get("param")));
+        ReturnResult<String> stringReturnResult = runJob(jobHandelName, param);
+        String callbackMethod = (String)jsonObject.get("callbackMethod");
+        cn.hutool.http.HttpUtil.post(callbackMethod,JSON.toJSONString(stringReturnResult));
+        log.info("---------服务器端:{},{}-------------",stringReturnResult.getCode(),stringReturnResult.getMsg());
+    }
+
+    private ReturnResult<String> runJob(String jobHandlerName,String param){
+        Class<? extends IJobHandler> jobClass = JobUtils.jobCache.get(jobHandlerName);
+        try {
+            IJobHandler iJobHandler = jobClass.newInstance( );
+            return iJobHandler.execute(param);
+        } catch (Exception e) {
+            e.printStackTrace( );
+        }
+        return null;
+
+    }
+}
 
 
 

demo-client/byit-demo-client/src/main/java/com/byit/job/DemoJob1.java

+package com.byit.job;
+
+import com.byit.annotations.JobHandler;
+import com.byit.job.handler.BaseJobHandler;
+import com.byit.job.model.ReturnResult;
+
+/**
+ * @program: byit-myth-job->DemoJob
+ * @description: TODO
+ * @author: huangfu
+ * @date: 2019/11/20 12:40
+ **/
+@JobHandler("addJob1")
+public class DemoJob1 extends BaseJobHandler {
+    @Override
+    public ReturnResult<String> execute(String s) throws Exception {
+        Thread.sleep(200000);
+        System.out.println("--------------DemoJob1----------------"+s );
+        return ReturnResult.SUCCESS;
+    }
+}

demo-client/byit-demo-client/src/main/java/com/byit/job/Mains.java

@@ -20,6 +20,7 @@ import java.util.concurrent.TimeUnit;
  **/
 public class Mains {
     public static void main(String[] args) throws IOException {
+        new JobRunServerLauncher(8888);
         String plServerUrl = "http://10.0.55.237:8888";
         String mythCron = "时间";
         String routingStrategy = LoadBalance.ROUND.name();
@@ -31,8 +32,10 @@ public class Mains {
         String param="sadsadsa";
         String name="addJob";
         JavaBeanJobInfo javaBeanJobInfo = new JavaBeanJobInfo(name,plServerUrl,mythCron,routingStrategy,blockingStrategy,callbackToken,gatewayToken,requestIP,requestPort,param);
+        //JavaBeanJobInfo javaBeanJobInfo1 = new JavaBeanJobInfo(name,plServerUrl,mythCron,routingStrategy,blockingStrategy,callbackToken,gatewayToken,requestIP,requestPort,param);
+        //javaBeanJobInfo1.setJobHandelName("addJob1");
+        //JobUtils.addJob(javaBeanJobInfo1);
         JobUtils.addJob(javaBeanJobInfo);
-        new JobRunServerLauncher(8888);
     }
 
 }