Merge remote-tracking branch 'origin/developer' into developer

byit-plugin-core/byit-plugin-rpc-common/pom.xml

@@ -46,6 +46,12 @@
         </dependency>
 
         <dependency>
+            <groupId>myth-job</groupId>
+            <artifactId>myth-executor-api</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+
+        <dependency>
             <groupId>org.dom4j</groupId>
             <artifactId>dom4j</artifactId>
             <version>2.1.1</version>

byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/enums/ResponseTyEnum.java

+package com.byit.enums;
+
+/**
+ * 响应的类型
+ * @author huangfu
+ */
+
+public enum ResponseTyEnum {
+    TRANSFER("transfer"),
+    RESPONSE("response")
+    ;
+    private String type;
+
+    public String getType() {
+        return type;
+    }
+
+    ResponseTyEnum(String type) {
+        this.type = type;
+    }
+}

byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/model/ServiceConfigModel.java

@@ -17,7 +17,6 @@ import java.util.Map;
 @AllArgsConstructor
 @ToString
 public class ServiceConfigModel {
-    private Boolean autoScan;
     private ServerConfigurationModel serverConfigurationModel;
     private Map<String,String> classNames;
 }

byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/packet/request/PluginRpcRequestPacket.java

@@ -6,8 +6,6 @@ import com.byit.packet.BasePacketModel;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 
-import java.util.Map;
-
 /**
  * 插件服务端的请求数据包
  * @author huangfu
@@ -20,9 +18,11 @@ public class PluginRpcRequestPacket extends BasePacketModel {
      */
     private String jobName;
     /**
-     * 任务参数
+     * 日志ID 扩展字段也可传其他参数
      */
-    private Map<String,Object> param;
+    private String extension;
+
+    private String param;
 
     @Override
     public Command getCommand() {

byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/packet/response/PluginRpcResponsePacket.java

@@ -14,9 +14,14 @@ import lombok.EqualsAndHashCode;
 @EqualsAndHashCode(callSuper = true)
 public class PluginRpcResponsePacket extends BasePacketModel {
 
+    /**
+     * 日志ID 扩展字段也可传其他参数
+     */
+    private String extension;
     private String code;
     private String msg;
     private Object result;
+    private String type;
     private boolean status = false;
     @Override
     public Command getCommand() {

byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/utils/ThreadPoolUtil.java

+package com.byit.utils;
+
+import com.byit.rpc.util.RpcException;
+
+import java.util.concurrent.*;
+
+
+/**
+ * 线程池创建工具
+ * @author huangfu
+ */
+public class ThreadPoolUtil {
+
+    /**
+     * 线程池创建
+     * @param serverType 服务名称
+     * @return 返回一个线程池
+     */
+    public static ThreadPoolExecutor makeServerThreadPool(final String serverType, int corePoolSize, int maxPoolSize){
+        return  new ThreadPoolExecutor(
+                corePoolSize,
+                maxPoolSize,
+                60L,
+                TimeUnit.SECONDS,
+                new LinkedBlockingQueue<Runnable>(1000),
+                r -> new Thread(r, "myth-plugin-rpc, "+serverType+"-serverHandlerPool-" + r.hashCode()),
+                (r, executor) -> {
+                    throw new RpcException("myth-plugin-rpc "+serverType+" Thread pool is EXHAUSTED!");
+                });
+
+    }
+
+}

byit-plugin-core/byit-plugin-rpc-common/src/main/java/com/byit/utils/XmlParseUtil.java

@@ -11,7 +11,6 @@ import org.dom4j.Element;
 import org.dom4j.io.SAXReader;
 
 import java.net.URL;
-import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -42,14 +41,15 @@ public class XmlParseUtil {
             throw new LauncherException(LauncherEnum.XML_DATA_ERROR.getMsg());
         }
         //开始解析内部数据 serverConfiguration
-        Element autoScanElement = rootElement.element("autoScan");
-        String value = autoScanElement.attributeValue("value");
-        serviceConfigModel.setAutoScan(Boolean.parseBoolean(value));
-        //开始解析内部数据 serverConfiguration
         Element serverConfigurationElement = rootElement.element("serverConfiguration");
         Element classNamesElement = rootElement.element("classNames");
-        parseServerConfiguration(serverConfigurationElement,serviceConfigModel);
-        packageScanValueParse(classNamesElement,serviceConfigModel);
+        if (serverConfigurationElement!=null) {
+            parseServerConfiguration(serverConfigurationElement,serviceConfigModel);
+        }
+
+        if(classNamesElement != null){
+            packageScanValueParse(classNamesElement,serviceConfigModel);
+        }
         return serviceConfigModel;
     }
 
@@ -68,15 +68,19 @@ public class XmlParseUtil {
         Element pluginServiceRegistryElement = serverConfigurationElement.element("pluginServiceRegistry");
         Element pluginServerClassElement = serverConfigurationElement.element("pluginServerClass");
         portParse(portElement,serverConfigurationModel);
-        serverStringParse(registryUrlElement,value -> {
-            serverConfigurationModel.setRegistryUrl(value);
-        });
-        serverStringParse(serverBizElement, value -> {
-            serverConfigurationModel.setServerBiz(value);
-        });
-        serverStringParse(serverEnvElement, value -> {
-            serverConfigurationModel.setServerEnv(value);
-        });
+
+        if (registryUrlElement != null) {
+            serverStringParse(registryUrlElement, serverConfigurationModel::setRegistryUrl);
+        }
+
+        if (serverBizElement != null) {
+            serverStringParse(serverBizElement, serverConfigurationModel::setServerBiz);
+        }
+
+        if (serverEnvElement != null) {
+            serverStringParse(serverEnvElement, serverConfigurationModel::setServerEnv);
+        }
+
         serverThreadParse(serverThreadElement,serverConfigurationModel);
 
         if (pluginServiceRegistryElement != null) {
@@ -157,12 +161,15 @@ public class XmlParseUtil {
         Iterator<Element> elementIterator = packageRootElement.elementIterator();
         while (elementIterator.hasNext()) {
             Element packageElement = elementIterator.next();
-            String packageElementText = packageElement.attributeValue("class");
-            String jobNameText = packageElement.attributeValue("jobName");
-            if(serviceConfigModel.getClassNames() == null){
-                serviceConfigModel.setClassNames(new ConcurrentHashMap<>(8));
+            if (packageElement!=null) {
+                String packageElementText = packageElement.attributeValue("class");
+                String jobNameText = packageElement.attributeValue("jobName");
+                if(serviceConfigModel.getClassNames() == null){
+                    serviceConfigModel.setClassNames(new ConcurrentHashMap<>(8));
+                }
+                serviceConfigModel.getClassNames().put(jobNameText,packageElementText);
             }
-            serviceConfigModel.getClassNames().put(jobNameText,packageElementText);
+
         }
     }
 

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

@@ -2,21 +2,16 @@ package com.byit.factory;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.net.NetUtil;
-import com.byit.model.ServerConfigurationModel;
-import com.byit.model.ServiceConfigModel;
 import com.byit.registry.DataSourceServiceRegistry;
 import com.byit.registry.PluginServiceRegistry;
 import com.byit.registry.client.model.RegistryDataParamVO;
-import com.byit.server.NettyPluginServer;
+import com.byit.server.netty.NettyPluginServer;
 import com.byit.server.PluginServer;
 import com.byit.utils.IpUtil;
-import com.byit.utils.XmlParseUtil;
 import org.apache.commons.lang3.StringUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * 插件服务端的工厂对象
@@ -27,6 +22,10 @@ public abstract class PluginServerFactory {
     private final static String DEFAULT_ENV_NAME="plugin-netty";
     private final static Class<? extends PluginServiceRegistry> DEFAULT_SERVICE_REGISTRY_CLASS = DataSourceServiceRegistry.class;
     private final static Class<? extends PluginServer> DEFAULT_NETTY_PLUGIN_SERVER_CLASS = NettyPluginServer.class;
+    /**
+     * 任务缓存处理器
+     */
+    private Map<String, Object> serverPoll = new ConcurrentHashMap<>(8);
     private int corePoolSize;
     private int maxPoolSize;
     private String registryUrl;
@@ -44,38 +43,16 @@ public abstract class PluginServerFactory {
     private Class<? extends PluginServer> pluginServerClass;
     private PluginServer pluginServer;
     private PluginServiceRegistry pluginServiceRegistry;
-    private Set<String> serverKeys;
+    private Set<String> serverKeys = new HashSet<>(8);
     private List<RegistryDataParamVO> registryDataParamVOs = new ArrayList<>(2);
 
 
     public PluginServerFactory() {
     }
 
-    public PluginServerFactory(String configLocation) {
-        ServiceConfigModel serviceConfigModel = XmlParseUtil.parse(configLocation);
-        ServerConfigurationModel serverConfigurationModel = serviceConfigModel.getServerConfigurationModel();
-        String pluginServiceRegistryStr = serverConfigurationModel.getPluginServiceRegistry();
-        String pluginServerClassStr = serverConfigurationModel.getPluginServerClass();
-        Class<? extends PluginServiceRegistry> pluginServiceRegistryClass = null;
-        Class<? extends PluginServer> pluginServerClass = null;
-        try {
-            pluginServiceRegistryClass = (Class<? extends PluginServiceRegistry>) Class.forName(pluginServiceRegistryStr);
-            pluginServerClass = (Class<? extends PluginServer>) Class.forName(pluginServerClassStr);
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-
-        Map<String, String> classNames = serviceConfigModel.getClassNames();
-        Set<String> keySet = classNames.keySet();
-
-        init(serverConfigurationModel.getCoreSize(),serverConfigurationModel.getMaxSize(),
-                serverConfigurationModel.getRegistryUrl(),serverConfigurationModel.getServerEnv(),
-                serverConfigurationModel.getServerBiz(),pluginServiceRegistryClass,pluginServerClass,keySet);
-    }
-
-    public void init(int corePoolSize, int maxPoolSize, String registryUrl, String env, String biz,
+    public void init(int corePoolSize, int maxPoolSize,Integer port, String registryUrl, String env, String biz,
                      Class<? extends PluginServiceRegistry> serviceRegistryClass,
-                     Class<? extends PluginServer> pluginServerClass, Set<String> serverKeys){
+                     Class<? extends PluginServer> pluginServerClass){
 
         if(!(corePoolSize>0 && maxPoolSize>0 && maxPoolSize>=corePoolSize)){
             this.corePoolSize = 60;
@@ -129,10 +106,17 @@ public abstract class PluginServerFactory {
         if(this.pluginServerClass == null){
             throw new RuntimeException("插件端使用的服务类型不能为空!");
         }
+    }
 
-        this.serverKeys = serverKeys;
+    public void addService(String key,Object serverBean){
+        serverKeys.add(key);
+        serverPoll.put(key,serverBean);
     }
 
+    /**
+     * 启动方法
+     * @throws Exception
+     */
     public void start() throws Exception {
         pluginServer = pluginServerClass.newInstance();
         pluginServiceRegistry = serviceRegistryClass.newInstance();
@@ -255,4 +239,20 @@ public abstract class PluginServerFactory {
     public void setServerKeys(Set<String> serverKeys) {
         this.serverKeys = serverKeys;
     }
+
+    public Map<String, Object> getServerPoll() {
+        return serverPoll;
+    }
+
+    public void setServerPoll(Map<String, Object> serverPoll) {
+        this.serverPoll = serverPoll;
+    }
+
+    public List<RegistryDataParamVO> getRegistryDataParamVOs() {
+        return registryDataParamVOs;
+    }
+
+    public void setRegistryDataParamVOs(List<RegistryDataParamVO> registryDataParamVOs) {
+        this.registryDataParamVOs = registryDataParamVOs;
+    }
 }

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

 package com.byit.factory;
 
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.model.ServerConfigurationModel;
+import com.byit.model.ServiceConfigModel;
 import com.byit.registry.PluginServiceRegistry;
 import com.byit.server.PluginServer;
 import com.byit.utils.XmlParseUtil;
 
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -12,15 +16,55 @@ import java.util.Set;
  */
 public class RpcMainPluginServerFactory extends PluginServerFactory {
 
-    public RpcMainPluginServerFactory(int corePoolSize, int maxPoolSize, String registryUrl, String env, String biz, Set<String> serverKeys, Class<? extends PluginServiceRegistry> serviceRegistryClass, Class<? extends PluginServer> pluginServerClass) {
-        init(corePoolSize,maxPoolSize,registryUrl,env,biz,serviceRegistryClass,pluginServerClass,serverKeys);
+
+    /**
+     * 全属性传递注入
+     * @param corePoolSize 核心线程池大小
+     * @param maxPoolSize  最大线程池大小
+     * @param port 端口号
+     * @param registryUrl 注册中心地址
+     * @param env 环境标识
+     * @param biz 工程名称
+     * @param serverKeys 服务key
+     * @param serviceRegistryClass 注册中心类
+     * @param pluginServerClass 插件服务类
+     */
+    public RpcMainPluginServerFactory(int corePoolSize, int maxPoolSize,Integer port, String registryUrl, String env, String biz, Set<String> serverKeys, Class<? extends PluginServiceRegistry> serviceRegistryClass, Class<? extends PluginServer> pluginServerClass) {
+        init(corePoolSize,maxPoolSize,port,registryUrl,env,biz,serviceRegistryClass,pluginServerClass);
     }
 
+    /**
+     * 配置文件注入
+     * @param configLocation
+     */
     public RpcMainPluginServerFactory(String configLocation) {
-        super(configLocation);
-    }
+        try {
+            ServiceConfigModel serviceConfigModel = XmlParseUtil.parse(configLocation);
+            ServerConfigurationModel serverConfigurationModel = serviceConfigModel.getServerConfigurationModel();
+            String pluginServiceRegistryStr = serverConfigurationModel.getPluginServiceRegistry();
+            String pluginServerClassStr = serverConfigurationModel.getPluginServerClass();
+            Class<? extends PluginServiceRegistry> pluginServiceRegistryClass = (Class<? extends PluginServiceRegistry>) Class.forName(pluginServiceRegistryStr);
+            Class<? extends PluginServer> pluginServerClass = (Class<? extends PluginServer>) Class.forName(pluginServerClassStr);
+            Map<String, String> classNames = serviceConfigModel.getClassNames();
+            super.init(serverConfigurationModel.getCoreSize(),serverConfigurationModel.getMaxSize(),
+                    serverConfigurationModel.getPort(), serverConfigurationModel.getRegistryUrl(),
+                    serverConfigurationModel.getServerEnv(), serverConfigurationModel.getServerBiz(),
+                    pluginServiceRegistryClass,pluginServerClass);
+
+            if (CollectionUtil.isNotEmpty(classNames)) {
+                classNames.forEach((key,value) ->{
+                    try {
+                        Object o = Class.forName(value).newInstance();
+                        super.addService(key,o);
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                    }
+                });
+            }
 
-    public static void main(String[] args) {
-        System.out.println(XmlParseUtil.parse("/plugin.xml"));
+            super.start();
+        }catch (Exception e){
+            e.printStackTrace();
+        }
     }
 }

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 {
-    @Override
-    public void start(PluginServerFactory pluginServerFactory) {
-        System.out.println("-----com.byit.server.NettyPluginServer.start----");
-    }
-
-    @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/MainNettyPluginServer.java

+package com.byit.server.netty;
+
+import com.byit.factory.PluginServerFactory;
+import com.byit.handler.PackerSpliterHandler;
+import com.byit.handler.PacketDecodeHandler;
+import com.byit.handler.PacketEncodeHandler;
+import com.byit.server.PluginServer;
+import com.byit.server.netty.handler.NettyPluginServerHandler;
+import com.byit.utils.ThreadPoolUtil;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * main方法启动
+ * @author huangfu
+ */
+public class MainNettyPluginServer extends PluginServer {
+    @Override
+    public void start(PluginServerFactory pluginServerFactory) {
+        ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.makeServerThreadPool(NettyPluginServer.class.getName(),
+                pluginServerFactory.getCorePoolSize(), pluginServerFactory.getMaxPoolSize());
+
+        EventLoopGroup bossGroup = new NioEventLoopGroup();
+        EventLoopGroup workGroup = new NioEventLoopGroup();
+
+        try {
+            ServerBootstrap serverBootstrap = new ServerBootstrap();
+            serverBootstrap.group(bossGroup,workGroup)
+                    .channel(NioServerSocketChannel.class)
+                    .childOption(ChannelOption.TCP_NODELAY, true)
+                    .childOption(ChannelOption.SO_KEEPALIVE, true)
+                    .childHandler(new ChannelInitializer<SocketChannel>() {
+                        @Override
+                        protected void initChannel(SocketChannel ch) {
+                            ChannelPipeline pipeline = ch.pipeline();
+                            pipeline.addLast("packerSpliterHandler",new PackerSpliterHandler());
+                            pipeline.addLast("packetDecodeHandler",new PacketDecodeHandler());
+                            pipeline.addLast("nettyPluginServerHandler",new NettyPluginServerHandler(pluginServerFactory,threadPoolExecutor));
+                            pipeline.addLast("packetEncodeHandler",new PacketEncodeHandler());
+                        }
+                    });
+            ChannelFuture channelFuture = serverBootstrap.bind(pluginServerFactory.getPort()).sync();
+            super.onStart();
+            ChannelFuture closeFuture = channelFuture.channel().closeFuture().sync();
+            closeFuture.addListener(future ->{
+                if (future.isSuccess()) {
+                    System.out.println("-----------------");
+                }
+            });
+        }catch (Exception e){
+            e.printStackTrace();
+        }finally {
+            try {
+                threadPoolExecutor.shutdown();
+            }catch (Exception e){
+                e.printStackTrace();
+            }
+
+            try {
+                bossGroup.shutdownGracefully();
+                workGroup.shutdownGracefully();
+
+            }catch (Exception e){
+                e.printStackTrace();
+            }
+
+        }
+    }
+
+    @Override
+    public void stop() {
+        System.out.println("-------------");
+        super.onStop();
+    }
+}

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

+package com.byit.server.netty;
+
+import com.byit.factory.PluginServerFactory;
+import com.byit.handler.PackerSpliterHandler;
+import com.byit.handler.PacketDecodeHandler;
+import com.byit.handler.PacketEncodeHandler;
+import com.byit.server.PluginServer;
+import com.byit.server.netty.handler.NettyPluginServerHandler;
+import com.byit.utils.ThreadPoolUtil;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+import java.util.concurrent.ThreadPoolExecutor;
+
+/**
+ * Netty服务启动
+ * @author huangfu
+ */
+public class NettyPluginServer extends PluginServer {
+    private Thread serverThread;
+    @Override
+    public void start(PluginServerFactory pluginServerFactory) {
+        serverThread = new Thread(() ->{
+            final ThreadPoolExecutor threadPoolExecutor = ThreadPoolUtil.makeServerThreadPool(NettyPluginServer.class.getName(),
+                    pluginServerFactory.getCorePoolSize(), pluginServerFactory.getMaxPoolSize());
+
+            EventLoopGroup bossGroup = new NioEventLoopGroup();
+            EventLoopGroup workGroup = new NioEventLoopGroup();
+
+            try {
+                ServerBootstrap serverBootstrap = new ServerBootstrap();
+                serverBootstrap.group(bossGroup,workGroup)
+                        .channel(NioServerSocketChannel.class)
+                        .childOption(ChannelOption.TCP_NODELAY, true)
+                        .childOption(ChannelOption.SO_KEEPALIVE, true)
+                        .childHandler(new ChannelInitializer<SocketChannel>() {
+                            @Override
+                            protected void initChannel(SocketChannel ch) {
+                                ChannelPipeline pipeline = ch.pipeline();
+                                pipeline.addLast("packerSpliterHandler",new PackerSpliterHandler());
+                                pipeline.addLast("packetDecodeHandler",new PacketDecodeHandler());
+                                pipeline.addLast("nettyPluginServerHandler",new NettyPluginServerHandler(pluginServerFactory,threadPoolExecutor));
+                                pipeline.addLast("packetEncodeHandler",new PacketEncodeHandler());
+                            }
+                        });
+                ChannelFuture channelFuture = serverBootstrap.bind(pluginServerFactory.getPort()).sync();
+                super.onStart();
+                channelFuture.channel().closeFuture().sync();
+            }catch (Exception e){
+                e.printStackTrace();
+            }finally {
+                try {
+                    threadPoolExecutor.shutdown();
+                }catch (Exception e){
+                    e.printStackTrace();
+                }
+
+                try {
+                    bossGroup.shutdownGracefully();
+                    workGroup.shutdownGracefully();
+                }catch (Exception e){
+                    e.printStackTrace();
+                }
+
+            }
+
+        });
+        serverThread.setName("【com.byit.server.netty.NettyPluginServer#start thread run】"+serverThread.hashCode());
+        serverThread.setDaemon(true);
+        serverThread.start();
+    }
+
+
+    @Override
+    public void stop() {
+        if (serverThread != null && serverThread.isAlive()) {
+            serverThread.interrupt();
+        }
+        super.onStop();
+    }
+}

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 心跳检测功能暂时不添加
+}

byit-plugin-core/myth-plugin-rpc-server/src/main/resources/plugin.xml

 <?xml version="1.0" encoding="UTF-8"?>
 <executor-plugin>
-    <autoScan value="false"/>
-
     <serverConfiguration>
-        <port value="8989"/>
-        <registryUrl value="http://slocalhost:8080"/>
+        <port value="8980"/>
+        <registryUrl value="http://localhost:8080/myth-register"/>
         <server-biz value="myth-job"/>
         <server-env value="dev"/>
-        <server-thread core-size="" max-size=""/>
-        <pluginServiceRegistry value="com.byit.Test"/>
-        <pluginServerClass value="com.byit.Test"/>
+        <server-thread core-size="30" max-size="100"/>
+        <pluginServiceRegistry value="com.byit.registry.DataSourceServiceRegistry"/>
+        <pluginServerClass value="com.byit.server.netty.MainNettyPluginServer"/>
     </serverConfiguration>
 
     <classNames>
-        <className jobName="demo" class="com.byit.job.DemoJob"/>
-        <className jobName="demo1" class="com.byit.job.EndNode"/>
-        <className jobName="demo2" class="com.byit.job.StartNode"/>
+        <className jobName="sendEmailTest" class="com.byit.server.SendEmailTest"/>
     </classNames>
 </executor-plugin>
\ No newline at end of file

byit-plugin-core/myth-plugin-rpc-server/src/test/java/com/byit/server/SendEmailTest.java

+package com.byit.server;
+
+import com.byit.dto.web.ReturnResult;
+import com.byit.executor.handler.interfaces.IJobHandler;
+
+public class SendEmailTest implements IJobHandler {
+    @Override
+    public ReturnResult<String> execute(String param) throws Exception {
+        return null;
+    }
+
+    @Override
+    public void init(String param) {
+        System.out.println("---");
+    }
+
+    @Override
+    public void destroy(String param) {
+        System.out.println("----");
+    }
+}

byit-plugin-core/myth-plugin-rpc-server/src/test/java/com/byit/server/TestServerStart.java

+package com.byit.server;
+
+import com.byit.factory.PluginServerFactory;
+import com.byit.factory.RpcMainPluginServerFactory;
+
+public class TestServerStart {
+
+    public static void main(String[] args) throws Exception {
+        new RpcMainPluginServerFactory("/plugin.xml");
+        System.out.println("----");
+    }
+}