perf[net]: 重构net的task dispatcher

README.md

@@ -176,8 +176,9 @@ feat[module]: 新增某一项功能
 perf[module]: 优化了模块代码或者优化了什么功能
 fix[module]: 修改了什么bug
 test[module]: 测试了什么东西
-doc[module]: 增加了什么文档
 del[module]: 删除了某些功能或者无用代码
+ref[module]: 重命名或者重构了模块
+doc[module]: 增加了什么文档
 ```
 
 Ⅶ. License

event/src/main/java/com/zfoo/event/manager/EventBus.java

@@ -91,7 +91,7 @@ public abstract class EventBus {
         return executors[RandomUtils.randomInt(EXECUTORS_SIZE)];
     }
 
-    public static Executor asyncExecute(int hashcode) {
+    public static Executor execute(int hashcode) {
         return executors[Math.abs(hashcode % EXECUTORS_SIZE)];
     }
 

net/src/main/java/com/zfoo/net/NetContext.java

@@ -21,7 +21,7 @@ import com.zfoo.net.dispatcher.manager.IPacketDispatcher;
 import com.zfoo.net.packet.service.IPacketService;
 import com.zfoo.net.session.manager.ISessionManager;
 import com.zfoo.net.session.model.Session;
-import com.zfoo.net.task.TaskManager;
+import com.zfoo.net.task.TaskBus;
 import com.zfoo.protocol.collection.ArrayUtils;
 import com.zfoo.protocol.exception.ExceptionUtils;
 import com.zfoo.protocol.util.IOUtils;
@@ -133,10 +133,10 @@ public class NetContext implements ApplicationListener<ApplicationContextEvent>,
 
         // 关闭TaskManager
         try {
-            Field field = TaskManager.class.getDeclaredField("executors");
+            Field field = TaskBus.class.getDeclaredField("executors");
             ReflectionUtils.makeAccessible(field);
 
-            var executors = (ExecutorService[]) ReflectionUtils.getField(field, TaskManager.getInstance());
+            var executors = (ExecutorService[]) ReflectionUtils.getField(field, null);
             for (ExecutorService executor : executors) {
                 ThreadUtils.shutdown(executor);
             }

net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java

@@ -31,7 +31,7 @@ import com.zfoo.net.packet.model.IPacketAttachment;
 import com.zfoo.net.packet.model.SignalPacketAttachment;
 import com.zfoo.net.session.model.AttributeType;
 import com.zfoo.net.session.model.Session;
-import com.zfoo.net.task.TaskManager;
+import com.zfoo.net.task.TaskBus;
 import com.zfoo.net.task.model.ReceiveTask;
 import com.zfoo.protocol.IPacket;
 import com.zfoo.protocol.exception.ExceptionUtils;
@@ -136,7 +136,7 @@ public class PacketDispatcher implements IPacketDispatcher {
         }
 
         // 正常发送消息的接收
-        TaskManager.getInstance().addTask(new ReceiveTask(session, packet, packetAttachment));
+        TaskBus.submit(new ReceiveTask(session, packet, packetAttachment));
     }
 
     @Override
@@ -265,7 +265,7 @@ public class PacketDispatcher implements IPacketDispatcher {
                             }
                         }
 
-                    }, TaskManager.getInstance().getExecutorByConsistentHash(executorConsistentHash));
+                    }, TaskBus.executor(executorConsistentHash));
 
 
             PacketSignal.addSignalAttachment(clientAttachment);

net/src/main/java/com/zfoo/net/task/ITaskManager.java

@@ -1,29 +0,0 @@
-/*
- * Copyright (C) 2020 The zfoo Authors
- *
- * Licensed 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 com.zfoo.net.task;
-
-import com.zfoo.net.task.model.ReceiveTask;
-
-import java.util.concurrent.ExecutorService;
-
-/**
- * @author jaysunxiao
- * @version 3.0
- */
-public interface ITaskManager {
-
-    void addTask(ReceiveTask task);
-
-    ExecutorService getExecutorByConsistentHash(int hash);
-}

net/src/main/java/com/zfoo/net/task/TaskBus.java

@@ -14,33 +14,28 @@
 package com.zfoo.net.task;
 
 import com.zfoo.net.NetContext;
-import com.zfoo.net.task.model.AbstractTaskDispatch;
-import com.zfoo.net.task.model.ITaskDispatch;
 import com.zfoo.net.task.model.ReceiveTask;
+import com.zfoo.net.task.route.AbstractTaskRoute;
+import com.zfoo.net.task.route.ITaskRoute;
 import com.zfoo.protocol.util.StringUtils;
-import io.netty.util.concurrent.FastThreadLocalThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author jaysunxiao
  * @version 3.0
  */
-public final class TaskManager implements ITaskManager {
+public final class TaskBus {
 
-    private static final Logger logger = LoggerFactory.getLogger(TaskManager.class);
-
-    private static final TaskManager INSTANCE = new TaskManager();
+    private static final Logger logger = LoggerFactory.getLogger(TaskBus.class);
 
     // 线程池的大小
     public static final int EXECUTOR_SIZE;
 
-    private static final ITaskDispatch taskDispatch;
+    private static final ITaskRoute taskRoute;
 
 
     /**
@@ -57,7 +52,7 @@ public final class TaskManager implements ITaskManager {
                 ? "default" : providerConfig.getDispatchThread();
 
         EXECUTOR_SIZE = "default".equals(dispatchThread) ? (Runtime.getRuntime().availableProcessors() + 1) : Integer.parseInt(dispatchThread);
-        taskDispatch = AbstractTaskDispatch.valueOf(dispatch);
+        taskRoute = AbstractTaskRoute.valueOf(dispatch);
 
         executors = new ExecutorService[EXECUTOR_SIZE];
         for (int i = 0; i < executors.length; i++) {
@@ -66,42 +61,12 @@ public final class TaskManager implements ITaskManager {
         }
     }
 
-    private static class TaskThreadFactory implements ThreadFactory {
-        private static final AtomicInteger poolNumber = new AtomicInteger(1);
-        private final ThreadGroup group;
-        private final AtomicInteger threadNumber = new AtomicInteger(1);
-        private final String namePrefix;
 
-        TaskThreadFactory() {
-            var s = System.getSecurityManager();
-            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
-            namePrefix = "task-p" + poolNumber.getAndIncrement() + "-t";
-        }
-
-        @Override
-        public Thread newThread(Runnable runnable) {
-            var t = new FastThreadLocalThread(group, runnable, namePrefix + threadNumber.getAndIncrement(), 0);
-            t.setDaemon(false);
-            t.setPriority(Thread.NORM_PRIORITY);
-            t.setUncaughtExceptionHandler((thread, e) -> logger.error(thread.toString(), e));
-            return t;
-        }
+    public static void submit(ReceiveTask task) {
+        taskRoute.getExecutor(task).execute(task);
     }
 
-    private TaskManager() {
-    }
-
-    public static TaskManager getInstance() {
-        return INSTANCE;
-    }
-
-    @Override
-    public void addTask(ReceiveTask task) {
-        taskDispatch.getExecutor(task).execute(task);
-    }
-
-    @Override
-    public ExecutorService getExecutorByConsistentHash(int executorConsistentHash) {
+    public static ExecutorService executor(int executorConsistentHash) {
         return executors[Math.abs(executorConsistentHash % EXECUTOR_SIZE)];
     }
 }

net/src/main/java/com/zfoo/net/task/TaskThreadFactory.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright (C) 2020 The zfoo Authors
+ * Licensed 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 com.zfoo.net.task;
+
+import io.netty.util.concurrent.FastThreadLocalThread;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+public class TaskThreadFactory implements ThreadFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(TaskThreadFactory.class);
+
+    private static final AtomicInteger poolNumber = new AtomicInteger(1);
+    private final ThreadGroup group;
+    private final AtomicInteger threadNumber = new AtomicInteger(1);
+    private final String namePrefix;
+
+    TaskThreadFactory() {
+        var s = System.getSecurityManager();
+        group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
+        namePrefix = "task-p" + poolNumber.getAndIncrement() + "-t";
+    }
+
+    @Override
+    public Thread newThread(Runnable runnable) {
+        var t = new FastThreadLocalThread(group, runnable, namePrefix + threadNumber.getAndIncrement(), 0);
+        t.setDaemon(false);
+        t.setPriority(Thread.NORM_PRIORITY);
+        t.setUncaughtExceptionHandler((thread, e) -> logger.error(thread.toString(), e));
+        return t;
+    }
+
+}

net/src/main/java/com/zfoo/net/task/route/AbstractTaskRoute.java

@@ -1,6 +1,5 @@
 /*
  * Copyright (C) 2020 The zfoo Authors
- *
  * Licensed 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
  *
@@ -9,9 +8,10 @@
  * 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 com.zfoo.net.task.model;
+package com.zfoo.net.task.route;
 
 import com.zfoo.protocol.util.StringUtils;
 
@@ -19,16 +19,16 @@ import com.zfoo.protocol.util.StringUtils;
  * @author jaysunxiao
  * @version 3.0
  */
-public abstract class AbstractTaskDispatch implements ITaskDispatch {
+public abstract class AbstractTaskRoute implements ITaskRoute {
 
-    public static ITaskDispatch valueOf(String taskDispatchName) {
+    public static ITaskRoute valueOf(String taskDispatchName) {
         switch (taskDispatchName) {
             case "random":
-                return new RandomTaskDispatch();
+                return new RandomTaskRoute();
             case "sessionId":
-                return new SessionIdTaskDispatch();
+                return new SessionIdTaskRoute();
             case "consistent-hash":
-                return new ConsistentHashTaskDispatch();
+                return new ConsistentHashTaskRoute();
             default:
                 throw new RuntimeException(StringUtils.format("没有找到对应的taskDispatch[{}]", taskDispatchName));
         }

net/src/main/java/com/zfoo/net/task/route/ConsistentHashTaskRoute.java

@@ -1,6 +1,5 @@
 /*
  * Copyright (C) 2020 The zfoo Authors
- *
  * Licensed 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
  *
@@ -9,11 +8,13 @@
  * 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 com.zfoo.net.task.model;
+package com.zfoo.net.task.route;
 
-import com.zfoo.net.task.TaskManager;
+import com.zfoo.net.task.TaskBus;
+import com.zfoo.net.task.model.ReceiveTask;
 
 import java.util.concurrent.ExecutorService;
 
@@ -21,11 +22,11 @@ import java.util.concurrent.ExecutorService;
  * @author jaysunxiao
  * @version 3.0
  */
-public class ConsistentHashTaskDispatch extends AbstractTaskDispatch {
+public class ConsistentHashTaskRoute extends AbstractTaskRoute {
 
-    private static ConsistentHashTaskDispatch INSTANCE = new ConsistentHashTaskDispatch();
+    private static ConsistentHashTaskRoute INSTANCE = new ConsistentHashTaskRoute();
 
-    public static ConsistentHashTaskDispatch getINSTANCE() {
+    public static ConsistentHashTaskRoute getINSTANCE() {
         return INSTANCE;
     }
 
@@ -34,10 +35,10 @@ public class ConsistentHashTaskDispatch extends AbstractTaskDispatch {
         var packetAttachment = receiveTask.getPacketAttachment();
 
         if (packetAttachment == null) {
-            return SessionIdTaskDispatch.getInstance().getExecutor(receiveTask);
+            return SessionIdTaskRoute.getInstance().getExecutor(receiveTask);
         }
 
-        return TaskManager.getInstance().getExecutorByConsistentHash(packetAttachment.executorConsistentHash());
+        return TaskBus.executor(packetAttachment.executorConsistentHash());
     }
 
 }

net/src/main/java/com/zfoo/net/task/route/ITaskRoute.java

@@ -1,6 +1,5 @@
 /*
  * Copyright (C) 2020 The zfoo Authors
- *
  * Licensed 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
  *
@@ -9,9 +8,12 @@
  * 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 com.zfoo.net.task.model;
+package com.zfoo.net.task.route;
+
+import com.zfoo.net.task.model.ReceiveTask;
 
 import java.util.concurrent.ExecutorService;
 
@@ -19,7 +21,7 @@ import java.util.concurrent.ExecutorService;
  * @author jaysunxiao
  * @version 3.0
  */
-public interface ITaskDispatch {
+public interface ITaskRoute {
 
     ExecutorService getExecutor(ReceiveTask receiveTask);
 

net/src/main/java/com/zfoo/net/task/route/RandomTaskRoute.java

@@ -1,6 +1,5 @@
 /*
  * Copyright (C) 2020 The zfoo Authors
- *
  * Licensed 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
  *
@@ -9,11 +8,13 @@
  * 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 com.zfoo.net.task.model;
+package com.zfoo.net.task.route;
 
-import com.zfoo.net.task.TaskManager;
+import com.zfoo.net.task.TaskBus;
+import com.zfoo.net.task.model.ReceiveTask;
 
 import java.util.concurrent.ExecutorService;
 
@@ -21,17 +22,17 @@ import java.util.concurrent.ExecutorService;
  * @author jaysunxiao
  * @version 3.0
  */
-public class RandomTaskDispatch extends AbstractTaskDispatch {
+public class RandomTaskRoute extends AbstractTaskRoute {
 
-    private static final RandomTaskDispatch INSTANCE = new RandomTaskDispatch();
+    private static final RandomTaskRoute INSTANCE = new RandomTaskRoute();
 
-    public static RandomTaskDispatch getInstance() {
+    public static RandomTaskRoute getInstance() {
         return INSTANCE;
     }
 
     @Override
     public ExecutorService getExecutor(ReceiveTask receiveTask) {
-        return TaskManager.getInstance().getExecutorByConsistentHash(-1);
+        return TaskBus.executor(-1);
     }
 
 }

net/src/main/java/com/zfoo/net/task/route/SessionIdTaskRoute.java

@@ -1,6 +1,5 @@
 /*
  * Copyright (C) 2020 The zfoo Authors
- *
  * Licensed 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
  *
@@ -9,11 +8,13 @@
  * 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 com.zfoo.net.task.model;
+package com.zfoo.net.task.route;
 
-import com.zfoo.net.task.TaskManager;
+import com.zfoo.net.task.TaskBus;
+import com.zfoo.net.task.model.ReceiveTask;
 import com.zfoo.util.math.HashUtils;
 
 import java.util.concurrent.ExecutorService;
@@ -24,18 +25,18 @@ import java.util.concurrent.ExecutorService;
  * @author jaysunxiao
  * @version 3.0
  */
-public class SessionIdTaskDispatch extends AbstractTaskDispatch {
+public class SessionIdTaskRoute extends AbstractTaskRoute {
 
-    private static final SessionIdTaskDispatch INSTANCE = new SessionIdTaskDispatch();
+    private static final SessionIdTaskRoute INSTANCE = new SessionIdTaskRoute();
 
-    public static SessionIdTaskDispatch getInstance() {
+    public static SessionIdTaskRoute getInstance() {
         return INSTANCE;
     }
 
     @Override
     public ExecutorService getExecutor(ReceiveTask receiveTask) {
         var session = receiveTask.getSession();
-        return TaskManager.getInstance().getExecutorByConsistentHash(HashUtils.fnvHash(session.getSid()));
+        return TaskBus.executor(HashUtils.fnvHash(session.getSid()));
     }
 
 }

net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java

@@ -54,7 +54,7 @@ public class PacketSignalTest {
 
         var countDownLatch = new CountDownLatch(executorSize);
         for (var i = 0; i < executorSize; i++) {
-            EventBus.asyncExecute(i).execute(new Runnable() {
+            EventBus.execute(i).execute(new Runnable() {
                 @Override
                 public void run() {
                     addAndRemoveArray();
@@ -72,7 +72,7 @@ public class PacketSignalTest {
 
         var countDownLatch = new CountDownLatch(executorSize);
         for (int i = 0; i < executorSize; i++) {
-            EventBus.asyncExecute(i).execute(new Runnable() {
+            EventBus.execute(i).execute(new Runnable() {
                 @Override
                 public void run() {
                     addAndRemoveMap();

orm/src/main/java/com/zfoo/orm/model/persister/CronOrmPersister.java

@@ -78,7 +78,7 @@ public class CronOrmPersister extends AbstractOrmPersister {
                 @Override
                 public void run() {
                     if (!OrmContext.isStop()) {
-                        EventBus.asyncExecute(entityDef.getClazz().hashCode()).execute(new Runnable() {
+                        EventBus.execute(entityDef.getClazz().hashCode()).execute(new Runnable() {
                             @Override
                             public void run() {
                                 entityCaches.persistAll();

orm/src/main/java/com/zfoo/orm/model/persister/TimeOrmPersister.java

@@ -47,7 +47,7 @@ public class TimeOrmPersister extends AbstractOrmPersister {
             @Override
             public void run() {
                 if (!OrmContext.isStop()) {
-                    EventBus.asyncExecute(entityDef.getClazz().hashCode()).execute(new Runnable() {
+                    EventBus.execute(entityDef.getClazz().hashCode()).execute(new Runnable() {
                         @Override
                         public void run() {
                             entityCaches.persistAll();