Netty服务端 异步实现修改

byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/NettyPluginServer.java

-package com.byit.server;
-
-import com.byit.factory.PluginServerFactory;
-
-/**
- * Netty服务启动
- * @author huangfu
- */
-public class NettyPluginServer extends PluginServer {
-    private Thread serverThread;
-    @Override
-    public void start(PluginServerFactory pluginServerFactory) {
-        serverThread = new Thread(() ->{
-            System.out.println("-----com.byit.server.NettyPluginServer.start----");
-        });
-        serverThread.setName("【com.byit.server.NettyPluginServer#start thread run】"+serverThread.hashCode());
-        serverThread.setDaemon(true);
-        serverThread.start();
-        super.onStart();
-    }
-
-    @Override
-    public void stop() {
-        System.out.println("-----com.byit.server.NettyPluginServer.stop-----");
-    }
-}

byit-plugin-core/myth-plugin-rpc-server/src/main/java/com/byit/server/netty/handler/NettyPluginServerHandler.java

+package com.byit.server.netty.handler;
+
+import com.byit.dto.web.ReturnResult;
+import com.byit.enums.ResponseTyEnum;
+import com.byit.executor.handler.interfaces.IJobHandler;
+import com.byit.factory.PluginServerFactory;
+import com.byit.packet.request.PluginRpcRequestPacket;
+import com.byit.packet.response.PluginRpcResponsePacket;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+import java.util.Map;
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * netty rpc服务器的业务处理
+ * @author huangfu
+ */
+public class NettyPluginServerHandler extends SimpleChannelInboundHandler<PluginRpcRequestPacket> {
+
+    private PluginServerFactory pluginServerFactory;
+    private ThreadPoolExecutor threadPoolExecutor;
+
+    public NettyPluginServerHandler(PluginServerFactory pluginServerFactory, ThreadPoolExecutor threadPoolExecutor) {
+        this.pluginServerFactory = pluginServerFactory;
+        this.threadPoolExecutor = threadPoolExecutor;
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, PluginRpcRequestPacket msg) throws Exception {
+        PluginRpcResponsePacket transferPluginRpcResponse = new PluginRpcResponsePacket();
+        threadPoolExecutor.execute(()->{
+            PluginRpcResponsePacket rpcResponsePacket = new PluginRpcResponsePacket();
+            try {
+                Map<String, Object> serverPoll = pluginServerFactory.getServerPoll();
+                String jobName = msg.getJobName();
+                Object bean = serverPoll.get(jobName);
+                IJobHandler iJobHandler = (IJobHandler)bean;
+                ReturnResult<String> execute = iJobHandler.execute(msg.getParam());
+                rpcResponsePacket.setResult(execute);
+                rpcResponsePacket.setCode("000000");
+                rpcResponsePacket.setMsg("SUCCESS");
+                rpcResponsePacket.setStatus(true);
+                rpcResponsePacket.setExtension(msg.getExtension());
+            }catch (Exception e){
+                rpcResponsePacket.setCode("500000");
+                rpcResponsePacket.setMsg(e.getMessage());
+                rpcResponsePacket.setStatus(false);
+                rpcResponsePacket.setExtension(msg.getExtension());
+            }
+            rpcResponsePacket.setType(ResponseTyEnum.RESPONSE.getType());
+            ctx.channel().writeAndFlush(rpcResponsePacket);
+        });
+        transferPluginRpcResponse.setStatus(true);
+        transferPluginRpcResponse.setMsg("调用成功");
+        transferPluginRpcResponse.setExtension(msg.getExtension());
+        transferPluginRpcResponse.setType(ResponseTyEnum.TRANSFER.getType());
+        ctx.channel().writeAndFlush(transferPluginRpcResponse);
+    }
+
+    /**
+     * 异常处理
+     * @param ctx 上下文对象
+     * @param cause 异常对象
+     * @throws Exception 异常信息
+     */
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        cause.printStackTrace();
+        ctx.close();
+    }
+
+    //TODO 心跳检测功能暂时不添加
+}