[ISSUE #549] Unify request handler code style of tcp and http protocol (

#4333)

* refactor: enhance http and tcp server style

* fix: create trace-plugin

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByIpPortHandler.java

@@ -119,7 +119,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
                         if (session.getClient().getHost().equals(ip) && String.valueOf(
                             session.getClient().getPort()).equals(port)) {
                             redirectResult.append("|");
-                            redirectResult.append(EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer,
+                            redirectResult.append(EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer.getTcpThreadPoolGroup(),
                                     destEventMeshIp, Integer.parseInt(destEventMeshPort),
                                     session, clientSessionGroupMapping));
                         }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientByPathHandler.java

@@ -116,7 +116,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
                         // to the new EventMesh node specified by given EventMesh IP and port.
                         if (session.getClient().getPath().contains(path)) {
                             redirectResult.append("|");
-                            redirectResult.append(EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer,
+                            redirectResult.append(EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer.getTcpThreadPoolGroup(),
                                 destEventMeshIp, Integer.parseInt(destEventMeshPort),
                                 session, clientSessionGroupMapping));
                         }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RedirectClientBySubSystemHandler.java

@@ -120,7 +120,7 @@ public void handle(final HttpExchange httpExchange) throws IOException {
                         // to the new EventMesh node specified by given EventMesh IP and port.
                         if (session.getClient().getSubsystem().equals(subSystem)) {
                             redirectResult.append('|')
-                                .append(EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer,
+                                .append(EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer.getTcpThreadPoolGroup(),
                                     destEventMeshIp, Integer.parseInt(destEventMeshPort),
                                     session, clientSessionGroupMapping));
                         }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectAllClientHandler.java

@@ -90,7 +90,7 @@ public void handle(final HttpExchange httpExchange) throws IOException {
                     // Iterate through the sessionMap and close each client connection
                     for (final Map.Entry<InetSocketAddress, Session> entry : sessionMap.entrySet()) {
                         final InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(
-                            eventMeshTCPServer, entry.getValue(), clientSessionGroupMapping);
+                            eventMeshTCPServer.getTcpThreadPoolGroup(), entry.getValue(), clientSessionGroupMapping);
                         // Add the remote client address to a list of successfully rejected addresses
                         if (addr != null) {
                             successRemoteAddrs.add(addr);

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientByIpPortHandler.java

@@ -113,7 +113,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
                     for (Map.Entry<InetSocketAddress, Session> entry : sessionMap.entrySet()) {
                         // Reject client connection for each matching session found
                         if (entry.getKey().getHostString().equals(ip) && String.valueOf(entry.getKey().getPort()).equals(port)) {
-                            InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer,
+                            InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer.getTcpThreadPoolGroup(),
                                 entry.getValue(), clientSessionGroupMapping);
                             // Add the remote client address to a list of successfully rejected addresses
                             if (addr != null) {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/RejectClientBySubSystemHandler.java

@@ -121,7 +121,7 @@ public void handle(HttpExchange httpExchange) throws IOException {
                     for (Session session : sessionMap.values()) {
                         // Reject client connection for each matching session found
                         if (session.getClient().getSubsystem().equals(subSystem)) {
-                            InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer, session,
+                            InetSocketAddress addr = EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer.getTcpThreadPoolGroup(), session,
                                 clientSessionGroupMapping);
                             // Add the remote client address to a list of successfully rejected addresses
                             if (addr != null) {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/admin/handler/TCPClientHandler.java

@@ -126,7 +126,7 @@ void delete(HttpExchange httpExchange) throws IOException {
                     if (entry.getKey().getHostString().equals(host) && entry.getKey().getPort() == port) {
                         // Call the serverGoodby2Client method in EventMeshTcp2Client to disconnect the client's connection
                         EventMeshTcp2Client.serverGoodby2Client(
-                            eventMeshTCPServer,
+                            eventMeshTCPServer.getTcpThreadPoolGroup(),
                             entry.getValue(),
                             clientSessionGroupMapping
                         );

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/AbstractRemotingServer.java

@@ -32,53 +32,20 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+/**
+ * The most basic server
+ */
 @Slf4j
 public abstract class AbstractRemotingServer {
-
+    private static final int MAX_THREADS = Runtime.getRuntime().availableProcessors();
     private static final int DEFAULT_SLEEP_SECONDS = 30;
 
     private EventLoopGroup bossGroup;
-
     private EventLoopGroup ioGroup;
-
     private EventExecutorGroup workerGroup;
 
     private int port;
 
-    private static final int MAX_THREADS = Runtime.getRuntime().availableProcessors();
-
-    public EventLoopGroup getBossGroup() {
-        return bossGroup;
-    }
-
-    public EventLoopGroup getIoGroup() {
-        return ioGroup;
-    }
-
-    public EventExecutorGroup getWorkerGroup() {
-        return workerGroup;
-    }
-
-    public int getPort() {
-        return port;
-    }
-
-    public void setBossGroup(final EventLoopGroup bossGroup) {
-        this.bossGroup = bossGroup;
-    }
-
-    public void setIoGroup(final EventLoopGroup ioGroup) {
-        this.ioGroup = ioGroup;
-    }
-
-    public void setWorkerGroup(final EventExecutorGroup workerGroup) {
-        this.workerGroup = workerGroup;
-    }
-
-    public void setPort(final int port) {
-        this.port = port;
-    }
-
     private void buildBossGroup(final String threadPrefix) {
         if (useEpoll()) {
             bossGroup = new EpollEventLoopGroup(1, new EventMeshThreadFactory(threadPrefix + "NettyEpoll-Boss", true));
@@ -106,6 +73,8 @@ public void init(final String threadPrefix) throws Exception {
         buildWorkerGroup(threadPrefix);
     }
 
+    public abstract void start() throws Exception;
+
     public void shutdown() throws Exception {
         if (bossGroup != null) {
             bossGroup.shutdownGracefully();
@@ -135,5 +104,35 @@ protected boolean useEpoll() {
         return SystemUtils.isLinuxPlatform() && Epoll.isAvailable();
     }
 
-    public abstract void start() throws Exception;
+    public EventLoopGroup getBossGroup() {
+        return bossGroup;
+    }
+
+    public EventLoopGroup getIoGroup() {
+        return ioGroup;
+    }
+
+    public EventExecutorGroup getWorkerGroup() {
+        return workerGroup;
+    }
+
+    public int getPort() {
+        return port;
+    }
+
+    public void setBossGroup(final EventLoopGroup bossGroup) {
+        this.bossGroup = bossGroup;
+    }
+
+    public void setIoGroup(final EventLoopGroup ioGroup) {
+        this.ioGroup = ioGroup;
+    }
+
+    public void setWorkerGroup(final EventExecutorGroup workerGroup) {
+        this.workerGroup = workerGroup;
+    }
+
+    public void setPort(final int port) {
+        this.port = port;
+    }
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshHttpBootstrap.java

@@ -43,6 +43,7 @@ public void init() throws Exception {
         // server init
         if (eventMeshHttpConfiguration != null) {
             eventMeshHttpServer = new EventMeshHTTPServer(eventMeshServer, eventMeshHttpConfiguration);
+            eventMeshHttpServer.init();
         }
     }
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshServer.java

@@ -76,8 +76,8 @@ public EventMeshServer() {
 
         //Initialize BOOTSTRAP_LIST based on protocols provided in configuration
         final List<String> provideServerProtocols = configuration.getEventMeshProvideServerProtocols();
-        for (final String provideServerProtocol : provideServerProtocols) {
-            switch (provideServerProtocol) {
+        for (String provideServerProtocol : provideServerProtocols) {
+            switch (provideServerProtocol.toUpperCase()) {
                 case ConfigurationContextUtil.HTTP:
                     BOOTSTRAP_LIST.add(new EventMeshHttpBootstrap(this));
                     break;

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/HTTPThreadPoolGroup.java

@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.boot;
+
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.runtime.configuration.EventMeshHTTPConfiguration;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+public class HTTPThreadPoolGroup implements ThreadPoolGroup {
+
+    private final EventMeshHTTPConfiguration eventMeshHttpConfiguration;
+
+    private ThreadPoolExecutor batchMsgExecutor;
+    private ThreadPoolExecutor sendMsgExecutor;
+    private ThreadPoolExecutor remoteMsgExecutor;
+    private ThreadPoolExecutor replyMsgExecutor;
+    private ThreadPoolExecutor pushMsgExecutor;
+    private ThreadPoolExecutor clientManageExecutor;
+    private ThreadPoolExecutor adminExecutor;
+    private ThreadPoolExecutor webhookExecutor;
+
+    public HTTPThreadPoolGroup(EventMeshHTTPConfiguration eventMeshHttpConfiguration) {
+        this.eventMeshHttpConfiguration = eventMeshHttpConfiguration;
+    }
+
+    @Override
+    public void initThreadPool() {
+
+        batchMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerBatchMsgThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerBatchMsgThreadNum(),
+                new LinkedBlockingQueue<>(eventMeshHttpConfiguration.getEventMeshServerBatchBlockQSize()),
+                "eventMesh-batchMsg", true);
+
+        sendMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerSendMsgThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerSendMsgThreadNum(),
+                new LinkedBlockingQueue<>(eventMeshHttpConfiguration.getEventMeshServerSendMsgBlockQSize()),
+                "eventMesh-sendMsg", true);
+
+        remoteMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerRemoteMsgThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerRemoteMsgThreadNum(),
+                new LinkedBlockingQueue<>(eventMeshHttpConfiguration.getEventMeshServerRemoteMsgBlockQSize()),
+                "eventMesh-remoteMsg", true);
+
+        pushMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerPushMsgThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerPushMsgThreadNum(),
+                new LinkedBlockingQueue<>(eventMeshHttpConfiguration.getEventMeshServerPushMsgBlockQSize()),
+                "eventMesh-pushMsg", true);
+
+        clientManageExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerClientManageThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerClientManageThreadNum(),
+                new LinkedBlockingQueue<>(eventMeshHttpConfiguration.getEventMeshServerClientManageBlockQSize()),
+                "eventMesh-clientManage", true);
+
+        adminExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerAdminThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerAdminThreadNum(),
+                new LinkedBlockingQueue<>(50), "eventMesh-admin",
+                true);
+
+        replyMsgExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerReplyMsgThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerReplyMsgThreadNum(),
+                new LinkedBlockingQueue<>(100),
+                "eventMesh-replyMsg", true);
+
+        webhookExecutor = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshHttpConfiguration.getEventMeshServerWebhookThreadNum(),
+                eventMeshHttpConfiguration.getEventMeshServerWebhookThreadNum(),
+                new LinkedBlockingQueue<>(100), "eventMesh-webhook", true);
+    }
+
+    @Override
+    public void shutdownThreadPool() {
+        if (batchMsgExecutor != null) {
+            batchMsgExecutor.shutdown();
+        }
+        if (adminExecutor != null) {
+            adminExecutor.shutdown();
+        }
+        if (clientManageExecutor != null) {
+            clientManageExecutor.shutdown();
+        }
+        if (sendMsgExecutor != null) {
+            sendMsgExecutor.shutdown();
+        }
+        if (remoteMsgExecutor != null) {
+            remoteMsgExecutor.shutdown();
+        }
+        if (pushMsgExecutor != null) {
+            pushMsgExecutor.shutdown();
+        }
+        if (replyMsgExecutor != null) {
+            replyMsgExecutor.shutdown();
+        }
+    }
+
+    public ThreadPoolExecutor getBatchMsgExecutor() {
+        return batchMsgExecutor;
+    }
+
+    public ThreadPoolExecutor getSendMsgExecutor() {
+        return sendMsgExecutor;
+    }
+
+    public ThreadPoolExecutor getRemoteMsgExecutor() {
+        return remoteMsgExecutor;
+    }
+
+    public ThreadPoolExecutor getReplyMsgExecutor() {
+        return replyMsgExecutor;
+    }
+
+    public ThreadPoolExecutor getPushMsgExecutor() {
+        return pushMsgExecutor;
+    }
+
+    public ThreadPoolExecutor getClientManageExecutor() {
+        return clientManageExecutor;
+    }
+
+    public ThreadPoolExecutor getAdminExecutor() {
+        return adminExecutor;
+    }
+
+    public ThreadPoolExecutor getWebhookExecutor() {
+        return webhookExecutor;
+    }
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/TCPThreadPoolGroup.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.boot;
+
+import org.apache.eventmesh.common.EventMeshThreadFactory;
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.runtime.configuration.EventMeshTCPConfiguration;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+
+public class TCPThreadPoolGroup implements ThreadPoolGroup {
+    private final EventMeshTCPConfiguration eventMeshTCPConfiguration;
+    private ScheduledExecutorService scheduler;
+    private ThreadPoolExecutor taskHandleExecutorService;
+    private ThreadPoolExecutor broadcastMsgDownstreamExecutorService;
+
+    public TCPThreadPoolGroup(EventMeshTCPConfiguration eventMeshTCPConfiguration) {
+        this.eventMeshTCPConfiguration = eventMeshTCPConfiguration;
+    }
+
+    @Override
+    public void initThreadPool() {
+
+        scheduler = ThreadPoolFactory.createScheduledExecutor(eventMeshTCPConfiguration.getEventMeshTcpGlobalScheduler(),
+                new EventMeshThreadFactory("eventMesh-tcp-scheduler", true));
+
+        taskHandleExecutorService = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshTCPConfiguration.getEventMeshTcpTaskHandleExecutorPoolSize(),
+                eventMeshTCPConfiguration.getEventMeshTcpTaskHandleExecutorPoolSize(),
+                new LinkedBlockingQueue<>(10_000),
+                new EventMeshThreadFactory("eventMesh-tcp-task-handle", true));
+
+        broadcastMsgDownstreamExecutorService = ThreadPoolFactory.createThreadPoolExecutor(
+                eventMeshTCPConfiguration.getEventMeshTcpMsgDownStreamExecutorPoolSize(),
+                eventMeshTCPConfiguration.getEventMeshTcpMsgDownStreamExecutorPoolSize(),
+                new LinkedBlockingQueue<>(10_000),
+                new EventMeshThreadFactory("eventMesh-tcp-msg-downstream", true));
+    }
+
+    @Override
+    public void shutdownThreadPool() {
+        scheduler.shutdown();
+        taskHandleExecutorService.shutdown();
+        broadcastMsgDownstreamExecutorService.shutdown();
+    }
+
+    public ScheduledExecutorService getScheduler() {
+        return scheduler;
+    }
+
+    public ThreadPoolExecutor getTaskHandleExecutorService() {
+        return taskHandleExecutorService;
+    }
+
+    public ThreadPoolExecutor getBroadcastMsgDownstreamExecutorService() {
+        return broadcastMsgDownstreamExecutorService;
+    }
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/ThreadPoolGroup.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.boot;
+
+/**
+ * The implementation class for this interface is used to assemble the thread pool required by the server
+ *
+ */
+public interface ThreadPoolGroup {
+    void initThreadPool();
+
+    void shutdownThreadPool();
+}

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/push/HTTPMessageHandler.java

@@ -68,7 +68,7 @@ private void checkTimeout() {
 
     public HTTPMessageHandler(EventMeshConsumer eventMeshConsumer) {
         this.eventMeshConsumer = eventMeshConsumer;
-        this.pushExecutor = eventMeshConsumer.getEventMeshHTTPServer().getPushMsgExecutor();
+        this.pushExecutor = eventMeshConsumer.getEventMeshHTTPServer().getHttpThreadPoolGroup().getPushMsgExecutor();
         waitingRequests.put(this.eventMeshConsumer.getConsumerGroupConf().getConsumerGroup(), Sets.newConcurrentHashSet());
         SCHEDULER.scheduleAtFixedRate(this::checkTimeout, 0, 1000, TimeUnit.MILLISECONDS);
     }
@@ -101,7 +101,7 @@ public boolean handle(final HandleMsgContext handleMsgContext) {
             return true;
         } catch (RejectedExecutionException e) {
             log.warn("pushMsgThreadPoolQueue is full, so reject, current task size {}",
-                handleMsgContext.getEventMeshHTTPServer().getPushMsgExecutor().getQueue().size(), e);
+                handleMsgContext.getEventMeshHTTPServer().getHttpThreadPoolGroup().getPushMsgExecutor().getQueue().size(), e);
             return false;
         }
     }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/EventMeshTcp2Client.java

@@ -24,7 +24,7 @@
 import org.apache.eventmesh.common.protocol.tcp.OPStatus;
 import org.apache.eventmesh.common.protocol.tcp.Package;
 import org.apache.eventmesh.common.protocol.tcp.RedirectInfo;
-import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.boot.TCPThreadPoolGroup;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.group.ClientSessionGroupMapping;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.SessionState;
@@ -45,7 +45,7 @@
 @Slf4j
 public class EventMeshTcp2Client {
 
-    public static InetSocketAddress serverGoodby2Client(EventMeshTCPServer eventMeshTCPServer,
+    public static InetSocketAddress serverGoodby2Client(TCPThreadPoolGroup tcpThreadPoolGroup,
         Session session,
         ClientSessionGroupMapping mapping) {
         log.info("serverGoodby2Client client[{}]", session.getClient());
@@ -55,7 +55,7 @@ public static InetSocketAddress serverGoodby2Client(EventMeshTCPServer eventMesh
             msg.setHeader(new Header(SERVER_GOODBYE_REQUEST, OPStatus.SUCCESS.getCode(),
                 "graceful normal quit from eventmesh", null));
 
-            eventMeshTCPServer.getScheduler().submit(new Runnable() {
+            tcpThreadPoolGroup.getScheduler().submit(new Runnable() {
                 @Override
                 public void run() {
                     long taskExecuteTime = System.currentTimeMillis();
@@ -64,30 +64,30 @@ public void run() {
             });
             InetSocketAddress address = (InetSocketAddress) session.getContext().channel().remoteAddress();
 
-            closeSessionIfTimeout(eventMeshTCPServer, session, mapping);
+            closeSessionIfTimeout(tcpThreadPoolGroup, session, mapping);
             return address;
         } catch (Exception e) {
             log.error("exception occur while serverGoodby2Client", e);
             return null;
         }
     }
 
-    public static InetSocketAddress goodBye2Client(EventMeshTCPServer eventMeshTCPServer, Session session,
-        String errMsg, int eventMeshStatus,
-        ClientSessionGroupMapping mapping) {
+    public static InetSocketAddress goodBye2Client(TCPThreadPoolGroup tcpThreadPoolGroup, Session session,
+                                                   String errMsg, int eventMeshStatus,
+                                                   ClientSessionGroupMapping mapping) {
         try {
             long startTime = System.currentTimeMillis();
             Package msg = new Package();
             msg.setHeader(new Header(SERVER_GOODBYE_REQUEST, eventMeshStatus, errMsg, null));
-            eventMeshTCPServer.getScheduler().schedule(new Runnable() {
+            tcpThreadPoolGroup.getScheduler().schedule(new Runnable() {
                 @Override
                 public void run() {
                     long taskExecuteTime = System.currentTimeMillis();
                     Utils.writeAndFlush(msg, startTime, taskExecuteTime, session.getContext(), session);
                 }
             }, 1 * 1000, TimeUnit.MILLISECONDS);
 
-            closeSessionIfTimeout(eventMeshTCPServer, session, mapping);
+            closeSessionIfTimeout(tcpThreadPoolGroup, session, mapping);
 
             return session.getRemoteAddress();
         } catch (Exception e) {
@@ -115,7 +115,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
         });
     }
 
-    public static String redirectClient2NewEventMesh(EventMeshTCPServer eventMeshTCPServer, String newEventMeshIp,
+    public static String redirectClient2NewEventMesh(TCPThreadPoolGroup tcpThreadPoolGroup, String newEventMeshIp,
         int port, Session session, ClientSessionGroupMapping mapping) {
         log.info("begin to gracefully redirect Client {}, newIPPort[{}]", session.getClient(), newEventMeshIp + ":"
             + port);
@@ -125,24 +125,24 @@ public static String redirectClient2NewEventMesh(EventMeshTCPServer eventMeshTCP
             Package pkg = new Package();
             pkg.setHeader(new Header(REDIRECT_TO_CLIENT, OPStatus.SUCCESS.getCode(), null, null));
             pkg.setBody(new RedirectInfo(newEventMeshIp, port));
-            eventMeshTCPServer.getScheduler().schedule(new Runnable() {
+            tcpThreadPoolGroup.getScheduler().schedule(new Runnable() {
                 @Override
                 public void run() {
                     long taskExecuteTime = System.currentTimeMillis();
                     Utils.writeAndFlush(pkg, startTime, taskExecuteTime, session.getContext(), session);
                 }
             }, 5 * 1000, TimeUnit.MILLISECONDS);
-            closeSessionIfTimeout(eventMeshTCPServer, session, mapping);
+            closeSessionIfTimeout(tcpThreadPoolGroup, session, mapping);
             return session.getRemoteAddress() + "--->" + newEventMeshIp + ":" + port;
         } catch (Exception e) {
             log.error("exception occur while redirectClient2NewEventMesh", e);
             return null;
         }
     }
 
-    public static void closeSessionIfTimeout(EventMeshTCPServer eventMeshTCPServer, Session session,
+    public static void closeSessionIfTimeout(TCPThreadPoolGroup tcpThreadPoolGroup, Session session,
         ClientSessionGroupMapping mapping) {
-        eventMeshTCPServer.getScheduler().schedule(new Runnable() {
+        tcpThreadPoolGroup.getScheduler().schedule(new Runnable() {
             @Override
             public void run() {
                 try {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientGroupWrapper.java

@@ -130,7 +130,7 @@ public ClientGroupWrapper(String sysId, String group,
         this.group = group;
         this.eventMeshTCPServer = eventMeshTCPServer;
         this.eventMeshTCPConfiguration = eventMeshTCPServer.getEventMeshTCPConfiguration();
-        this.eventMeshTcpRetryer = eventMeshTCPServer.getEventMeshTcpRetryer();
+        this.eventMeshTcpRetryer = eventMeshTCPServer.getTcpRetryer();
         this.eventMeshTcpMonitor =
             Preconditions.checkNotNull(eventMeshTCPServer.getEventMeshTcpMonitor());
         this.downstreamDispatchStrategy = downstreamDispatchStrategy;
@@ -627,7 +627,7 @@ public synchronized void initClientGroupBroadcastConsumer() throws Exception {
                     downStreamMsgContext.setSession(session);
 
                     //downstream broadcast msg asynchronously
-                    eventMeshTCPServer.getBroadcastMsgDownstreamExecutorService()
+                    eventMeshTCPServer.getTcpThreadPoolGroup().getBroadcastMsgDownstreamExecutorService()
                         .submit(() -> {
                             //msg put in eventmesh,waiting client ack
                             session.getPusher().unAckMsg(downStreamMsgContext.seq, downStreamMsgContext);

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/group/ClientSessionGroupMapping.java

@@ -357,7 +357,7 @@ private void shutdownClientGroupProducer(ClientGroupWrapper clientGroupWrapper)
     }
 
     private void initSessionCleaner() {
-        eventMeshTCPServer.getScheduler().scheduleAtFixedRate(
+        eventMeshTCPServer.getTcpThreadPoolGroup().getScheduler().scheduleAtFixedRate(
             () -> {
                 for (Session tmp : sessionTable.values()) {
                     if (System.currentTimeMillis() - tmp.getLastHeartbeatTime()
@@ -376,7 +376,7 @@ private void initSessionCleaner() {
     }
 
     private void initDownStreamMsgContextCleaner() {
-        eventMeshTCPServer.getScheduler().scheduleAtFixedRate(
+        eventMeshTCPServer.getTcpThreadPoolGroup().getScheduler().scheduleAtFixedRate(
             () -> {
 
                 //scan non-broadcast msg
@@ -412,15 +412,15 @@ public void shutdown() throws Exception {
         for (ClientGroupWrapper clientGroupWrapper : clientGroupMap.values()) {
             for (Session subSession : clientGroupWrapper.getGroupConsumerSessions()) {
                 try {
-                    EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer, subSession, this);
+                    EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer.getTcpThreadPoolGroup(), subSession, this);
                 } catch (Exception e) {
                     log.error("say goodbye to subSession error! {}", subSession, e);
                 }
             }
 
             for (Session pubSession : clientGroupWrapper.getGroupProducerSessions()) {
                 try {
-                    EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer, pubSession, this);
+                    EventMeshTcp2Client.serverGoodby2Client(eventMeshTCPServer.getTcpThreadPoolGroup(), pubSession, this);
                 } catch (Exception e) {
                     log.error("say goodbye to pubSession error! {}", pubSession, e);
                 }
@@ -434,7 +434,7 @@ public void shutdown() throws Exception {
 
         sessionTable.values().parallelStream().forEach(itr -> {
             try {
-                EventMeshTcp2Client.serverGoodby2Client(this.eventMeshTCPServer, itr, this);
+                EventMeshTcp2Client.serverGoodby2Client(this.eventMeshTCPServer.getTcpThreadPoolGroup(), itr, this);
             } catch (Exception e) {
                 log.error("say goodbye to session error! {}", itr, e);
             }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/GoodbyeTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/GoodbyeProcessor.java

@@ -15,16 +15,18 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import static org.apache.eventmesh.common.protocol.tcp.Command.CLIENT_GOODBYE_RESPONSE;
 
 import org.apache.eventmesh.common.protocol.tcp.Command;
 import org.apache.eventmesh.common.protocol.tcp.Header;
 import org.apache.eventmesh.common.protocol.tcp.OPStatus;
 import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.EventMeshTcp2Client;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.util.Utils;
 
 import java.util.Arrays;
@@ -35,14 +37,19 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class GoodbyeTask extends AbstractTask {
+public class GoodbyeProcessor implements TcpProcessor {
 
-    public GoodbyeTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    private EventMeshTCPServer eventMeshTCPServer;
+    private final Acl acl;
+
+    public GoodbyeProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+        this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         long taskExecuteTime = System.currentTimeMillis();
         Package msg = new Package();
         try {
@@ -58,9 +65,10 @@ public void run() {
             msg.setHeader(new Header(CLIENT_GOODBYE_RESPONSE, OPStatus.FAIL.getCode(), Arrays.toString(e.getStackTrace()),
                 pkg.getHeader().getSeq()));
         } finally {
-            this.eventMeshTCPServer.getScheduler().submit(() -> Utils.writeAndFlush(msg, startTime, taskExecuteTime, session.getContext(), session));
+            this.eventMeshTCPServer.getTcpThreadPoolGroup().getScheduler()
+                    .submit(() -> Utils.writeAndFlush(msg, startTime, taskExecuteTime, session.getContext(), session));
         }
         EventMeshTcp2Client
-            .closeSessionIfTimeout(this.eventMeshTCPServer, session, eventMeshTCPServer.getClientSessionGroupMapping());
+            .closeSessionIfTimeout(this.eventMeshTCPServer.getTcpThreadPoolGroup(), session, eventMeshTCPServer.getClientSessionGroupMapping());
     }
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/HeartBeatTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HeartBeatProcessor.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import static org.apache.eventmesh.common.protocol.tcp.Command.HEARTBEAT_REQUEST;
 import static org.apache.eventmesh.common.protocol.tcp.Command.HEARTBEAT_RESPONSE;
@@ -25,6 +25,7 @@
 import org.apache.eventmesh.common.protocol.tcp.Package;
 import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.util.RemotingHelper;
 import org.apache.eventmesh.runtime.util.Utils;
 
@@ -36,17 +37,19 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class HeartBeatTask extends AbstractTask {
+public class HeartBeatProcessor implements TcpProcessor {
 
+    private EventMeshTCPServer eventMeshTCPServer;
     private final Acl acl;
 
-    public HeartBeatTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    public HeartBeatProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
         this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         long taskExecuteTime = System.currentTimeMillis();
         Package res = new Package();
         try {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/HelloTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/HelloProcessor.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import static org.apache.eventmesh.common.protocol.tcp.Command.HELLO_RESPONSE;
 
@@ -46,19 +46,20 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class HelloTask extends AbstractTask {
+public class HelloProcessor implements TcpProcessor {
 
     private static final Logger MESSAGE_LOGGER = LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
 
+    private EventMeshTCPServer eventMeshTCPServer;
     private final Acl acl;
 
-    public HelloTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    public HelloProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
         this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
         long taskExecuteTime = System.currentTimeMillis();
         Package res = new Package();
         Session session = null;

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/ListenTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/ListenProcessor.java

@@ -15,29 +15,37 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import static org.apache.eventmesh.common.protocol.tcp.Command.LISTEN_RESPONSE;
 
 import org.apache.eventmesh.common.protocol.tcp.Header;
 import org.apache.eventmesh.common.protocol.tcp.OPStatus;
 import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 
 import io.netty.channel.ChannelHandlerContext;
 
 
 import lombok.extern.slf4j.Slf4j;
 
+
 @Slf4j
-public class ListenTask extends AbstractTask {
+public class ListenProcessor implements TcpProcessor {
+
+    private EventMeshTCPServer eventMeshTCPServer;
+    private final Acl acl;
 
-    public ListenTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    public ListenProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+        this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         long taskExecuteTime = System.currentTimeMillis();
         Header header = new Header(LISTEN_RESPONSE, OPStatus.SUCCESS.getCode(), OPStatus.SUCCESS.getDesc(), pkg.getHeader().getSeq());
         session.setListenRequestSeq(pkg.getHeader().getSeq());

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageAckTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/MessageAckProcessor.java

@@ -15,12 +15,14 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import org.apache.eventmesh.common.protocol.tcp.Command;
 import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.DownStreamMsgContext;
 
 import org.slf4j.Logger;
@@ -32,16 +34,19 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class MessageAckTask extends AbstractTask {
-
+public class MessageAckProcessor implements TcpProcessor {
     private static final Logger MESSAGE_LOGGER = LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
+    private EventMeshTCPServer eventMeshTCPServer;
+    private final Acl acl;
 
-    public MessageAckTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    public MessageAckProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+        this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         long taskExecuteTime = System.currentTimeMillis();
         String seq = pkg.getHeader().getSeq();
         Command cmd = pkg.getHeader().getCmd();

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/MessageTransferTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/MessageTransferProcessor.java

@@ -15,9 +15,10 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import static org.apache.eventmesh.common.protocol.tcp.Command.RESPONSE_TO_SERVER;
+import static org.apache.eventmesh.runtime.core.protocol.tcp.client.session.send.SessionSender.TRY_PERMIT_TIME_OUT;
 
 import org.apache.eventmesh.api.SendCallback;
 import org.apache.eventmesh.api.SendResult;
@@ -33,6 +34,7 @@
 import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.send.EventMeshTcpSendResult;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.send.EventMeshTcpSendStatus;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.send.UpStreamMsgContext;
@@ -63,21 +65,30 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class MessageTransferTask extends AbstractTask {
-
+public class MessageTransferProcessor implements TcpProcessor {
     private static final Logger MESSAGE_LOGGER = LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
-
-    private static final int TRY_PERMIT_TIME_OUT = 5;
-
+    private EventMeshTCPServer eventMeshTCPServer;
     private final Acl acl;
 
-    public MessageTransferTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    private Package pkg;
+    private ChannelHandlerContext ctx;
+    private Session session;
+    private long startTime;
+
+    public MessageTransferProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
         this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
+
+        this.pkg = pkg;
+        this.ctx = ctx;
+        this.session = session;
+        this.startTime = startTime;
+
         long taskExecuteTime = System.currentTimeMillis();
         Command cmd = pkg.getHeader().getCmd();
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/RecommendTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/RecommendProcessor.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import static org.apache.eventmesh.common.protocol.tcp.Command.RECOMMEND_RESPONSE;
 import static org.apache.eventmesh.runtime.util.Utils.writeAndFlush;
@@ -24,10 +24,12 @@
 import org.apache.eventmesh.common.protocol.tcp.OPStatus;
 import org.apache.eventmesh.common.protocol.tcp.Package;
 import org.apache.eventmesh.common.protocol.tcp.UserAgent;
+import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendImpl;
 import org.apache.eventmesh.runtime.core.protocol.tcp.client.recommend.EventMeshRecommendStrategy;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 
 import org.apache.commons.lang3.StringUtils;
 
@@ -36,16 +38,20 @@
 
 import io.netty.channel.ChannelHandlerContext;
 
-public class RecommendTask extends AbstractTask {
-
+public class RecommendProcessor implements TcpProcessor {
     private static final Logger MESSAGE_LOGGER = LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
 
-    public RecommendTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    private EventMeshTCPServer eventMeshTCPServer;
+    private final Acl acl;
+
+    public RecommendProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+        this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         long taskExecuteTime = System.currentTimeMillis();
         Package res = new Package();
         try {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/SubscribeTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/SubscribeProcessor.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import org.apache.eventmesh.api.exception.AclException;
 import org.apache.eventmesh.api.registry.bo.EventMeshAppSubTopicInfo;
@@ -27,6 +27,7 @@
 import org.apache.eventmesh.common.protocol.tcp.Subscription;
 import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.util.RemotingHelper;
 import org.apache.eventmesh.runtime.util.Utils;
 
@@ -40,17 +41,19 @@
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
-public class SubscribeTask extends AbstractTask {
+public class SubscribeProcessor implements TcpProcessor {
 
+    private EventMeshTCPServer eventMeshTCPServer;
     private final Acl acl;
 
-    public SubscribeTask(final Package pkg, final ChannelHandlerContext ctx, long startTime, final EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    public SubscribeProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
         this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         final long taskExecuteTime = System.currentTimeMillis();
 
         final Package msg = new Package();

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/AbstractTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/TcpProcessor.java

@@ -15,27 +15,15 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import org.apache.eventmesh.common.protocol.tcp.Package;
-import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
-import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 
 import io.netty.channel.ChannelHandlerContext;
 
-public abstract class AbstractTask implements Runnable {
-
-    protected Package pkg;
-    protected ChannelHandlerContext ctx;
-    protected Session session;
-    protected long startTime;
-    protected EventMeshTCPServer eventMeshTCPServer;
-
-    public AbstractTask(final Package pkg, final ChannelHandlerContext ctx, long startTime, final EventMeshTCPServer eventMeshTCPServer) {
-        this.eventMeshTCPServer = eventMeshTCPServer;
-        this.pkg = pkg;
-        this.ctx = ctx;
-        this.session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
-        this.startTime = startTime;
-    }
+/**
+ * TcpProcessor
+ */
+public interface TcpProcessor {
+    void process(final Package pkg, final ChannelHandlerContext ctx, long startTime);
 }

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/task/UnSubscribeTask.java → eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/processor/UnSubscribeProcessor.java

@@ -15,15 +15,17 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.runtime.core.protocol.tcp.client.task;
+package org.apache.eventmesh.runtime.core.protocol.tcp.client.processor;
 
 import org.apache.eventmesh.common.protocol.SubscriptionItem;
 import org.apache.eventmesh.common.protocol.tcp.Command;
 import org.apache.eventmesh.common.protocol.tcp.Header;
 import org.apache.eventmesh.common.protocol.tcp.OPStatus;
 import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.runtime.acl.Acl;
 import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
 import org.apache.eventmesh.runtime.constants.EventMeshConstants;
+import org.apache.eventmesh.runtime.core.protocol.tcp.client.session.Session;
 import org.apache.eventmesh.runtime.util.Utils;
 
 import org.apache.commons.collections4.MapUtils;
@@ -37,16 +39,20 @@
 
 import io.netty.channel.ChannelHandlerContext;
 
-public class UnSubscribeTask extends AbstractTask {
-
+public class UnSubscribeProcessor implements TcpProcessor {
     private static final Logger MESSAGE_LOGGER = LoggerFactory.getLogger(EventMeshConstants.MESSAGE);
 
-    public UnSubscribeTask(Package pkg, ChannelHandlerContext ctx, long startTime, EventMeshTCPServer eventMeshTCPServer) {
-        super(pkg, ctx, startTime, eventMeshTCPServer);
+    private EventMeshTCPServer eventMeshTCPServer;
+    private final Acl acl;
+
+    public UnSubscribeProcessor(EventMeshTCPServer eventMeshTCPServer) {
+        this.eventMeshTCPServer = eventMeshTCPServer;
+        this.acl = eventMeshTCPServer.getAcl();
     }
 
     @Override
-    public void run() {
+    public void process(final Package pkg, final ChannelHandlerContext ctx, long startTime) {
+        Session session = eventMeshTCPServer.getClientSessionGroupMapping().getSession(ctx);
         long taskExecuteTime = System.currentTimeMillis();
         Package msg = new Package();
         try {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/rebalance/EventMeshRebalanceImpl.java

@@ -167,7 +167,7 @@ private void doRedirect(String group, String purpose, int judge, List<String> ev
         for (int i = 0; i < judge; i++) {
             String newProxyIp = eventMeshRecommendResult.get(i).split(":")[0];
             String newProxyPort = eventMeshRecommendResult.get(i).split(":")[1];
-            String redirectSessionAddr = EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer, newProxyIp,
+            String redirectSessionAddr = EventMeshTcp2Client.redirectClient2NewEventMesh(eventMeshTCPServer.getTcpThreadPoolGroup(), newProxyIp,
                 Integer.parseInt(newProxyPort), sessionList.get(i), eventMeshTCPServer.getClientSessionGroupMapping());
             log.info("doRebalance,redirect sessionAddr:{}", redirectSessionAddr);
             ThreadUtils.sleep(eventMeshTCPServer.getEventMeshTCPConfiguration().getSleepIntervalInRebalanceRedirectMills(), TimeUnit.MILLISECONDS);

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/send/SessionSender.java

@@ -65,7 +65,7 @@ public class SessionSender {
 
     public final transient AtomicLong failMsgCount = new AtomicLong(0);
 
-    private static final int TRY_PERMIT_TIME_OUT = 5;
+    public static final int TRY_PERMIT_TIME_OUT = 5;
 
     @Override
     public String toString() {

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java

@@ -41,16 +41,16 @@ public class HTTPMetricsServer {
     private final transient HttpSummaryMetrics summaryMetrics;
 
     public HTTPMetricsServer(final EventMeshHTTPServer eventMeshHTTPServer,
-        final List<MetricsRegistry> metricsRegistries) {
+                             final List<MetricsRegistry> metricsRegistries) {
         Objects.requireNonNull(eventMeshHTTPServer, "EventMeshHTTPServer can not be null");
         Objects.requireNonNull(metricsRegistries, "List<MetricsRegistry> can not be null");
 
         this.eventMeshHTTPServer = eventMeshHTTPServer;
         this.metricsRegistries = metricsRegistries;
         this.summaryMetrics = new HttpSummaryMetrics(
-            eventMeshHTTPServer.getBatchMsgExecutor(),
-            eventMeshHTTPServer.getSendMsgExecutor(),
-            eventMeshHTTPServer.getPushMsgExecutor(),
+            eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor(),
+            eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor(),
+            eventMeshHTTPServer.getHttpThreadPoolGroup().getPushMsgExecutor(),
             eventMeshHTTPServer.getHttpRetryer().getFailedQueue());
 
         init();
@@ -168,9 +168,9 @@ private void logPrintServerMetrics(final HttpSummaryMetrics summaryMetrics,
 
         if (log.isInfoEnabled()) {
             log.info("batchMsgQ: {}, sendMsgQ: {}, pushMsgQ: {}, httpRetryQ: {}",
-                eventMeshHTTPServer.getBatchMsgExecutor().getQueue().size(),
-                eventMeshHTTPServer.getSendMsgExecutor().getQueue().size(),
-                eventMeshHTTPServer.getPushMsgExecutor().getQueue().size(),
+                eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor().getQueue().size(),
+                eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor().getQueue().size(),
+                eventMeshHTTPServer.getHttpThreadPoolGroup().getPushMsgExecutor().getQueue().size(),
                 eventMeshHTTPServer.getHttpRetryer().size());
         }
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/tcp/EventMeshTcpMonitor.java

@@ -84,7 +84,7 @@ public void start() throws Exception {
         });
 
         int delay = 60 * 1000;
-        monitorTpsTask = eventMeshTCPServer.getScheduler().scheduleAtFixedRate((() -> {
+        monitorTpsTask = eventMeshTCPServer.getTcpThreadPoolGroup().getScheduler().scheduleAtFixedRate((() -> {
             int msgNum = tcpSummaryMetrics.client2eventMeshMsgNum();
             tcpSummaryMetrics.resetClient2EventMeshMsgNum();
             tcpSummaryMetrics.setClient2eventMeshTPS((int) 1000.0d * msgNum / period);
@@ -121,18 +121,18 @@ public void start() throws Exception {
                 topicSet.addAll(session.getSessionContext().getSubscribeTopics().keySet());
             }
             tcpSummaryMetrics.setSubTopicNum(topicSet.size());
-            tcpSummaryMetrics.setAllConnections(eventMeshTCPServer.getEventMeshTcpConnectionHandler().getConnectionCount());
+            tcpSummaryMetrics.setAllConnections(eventMeshTCPServer.getTcpConnectionHandler().getConnectionCount());
             printAppLogger(tcpSummaryMetrics);
 
 
         }), delay, period, TimeUnit.MILLISECONDS);
 
-        monitorThreadPoolTask = eventMeshTCPServer.getScheduler().scheduleAtFixedRate(() -> {
+        monitorThreadPoolTask = eventMeshTCPServer.getTcpThreadPoolGroup().getScheduler().scheduleAtFixedRate(() -> {
             eventMeshTCPServer.getEventMeshRebalanceService().printRebalanceThreadPoolState();
-            eventMeshTCPServer.getEventMeshTcpRetryer().printRetryThreadPoolState();
+            eventMeshTCPServer.getTcpRetryer().printRetryThreadPoolState();
 
             //monitor retry queue size
-            tcpSummaryMetrics.setRetrySize(eventMeshTCPServer.getEventMeshTcpRetryer().getRetrySize());
+            tcpSummaryMetrics.setRetrySize(eventMeshTCPServer.getTcpRetryer().getRetrySize());
             appLogger.info(
                 MonitorMetricConstants.EVENTMESH_MONITOR_FORMAT_COMMON,
                 EventMeshConstants.PROTOCOL_TCP,