del[net]: remove difficult and infrequently used task-dispatcher configurations

net/src/main/java/com/zfoo/net/config/model/ProviderConfig.java

@@ -26,12 +26,7 @@ import java.util.Objects;
  */
 public class ProviderConfig {
 
-    public static transient final int DEFAULT_PORT = 12400;
-
-    /**
-     * 对应于ITaskDispatch
-     */
-    private String taskDispatch;
+    public static final int DEFAULT_PORT = 12400;
 
     private String thread;
 
@@ -55,14 +50,6 @@ public class ProviderConfig {
         return HostAndPort.valueOf(address);
     }
 
-    public String getTaskDispatch() {
-        return taskDispatch;
-    }
-
-    public void setTaskDispatch(String taskDispatch) {
-        this.taskDispatch = taskDispatch;
-    }
-
     public String getThread() {
         return thread;
     }

net/src/main/java/com/zfoo/net/router/Router.java

@@ -34,8 +34,8 @@ import com.zfoo.net.router.route.PacketBus;
 import com.zfoo.net.router.route.SignalBridge;
 import com.zfoo.net.session.model.AttributeType;
 import com.zfoo.net.session.model.Session;
+import com.zfoo.net.task.PacketReceiverTask;
 import com.zfoo.net.task.TaskBus;
-import com.zfoo.net.task.model.PacketReceiverTask;
 import com.zfoo.protocol.IPacket;
 import com.zfoo.protocol.exception.ExceptionUtils;
 import com.zfoo.protocol.util.JsonUtils;
@@ -151,7 +151,7 @@ public class Router implements IRouter {
 
         // 正常发送消息的接收,把客户端的业务请求包装下到路由策略指定的线程进行业务处理
         // 注意：像客户端以asyncAsk发送请求，在服务器处理完后返回结果，在请求方也是进入这个receive方法，但是attachment不为空，会提前return掉不会走到这
-        TaskBus.submit(new PacketReceiverTask(session, packet, attachment));
+        TaskBus.dispatch(new PacketReceiverTask(session, packet, attachment));
     }
 
     @Override

net/src/main/java/com/zfoo/net/schema/NamespaceHandler.java

@@ -16,7 +16,7 @@ package com.zfoo.net.schema;
 import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
 
 /**
- * @author jaysunxiao
+ * @author godotg
  * @version 3.0
  */
 public class NamespaceHandler extends NamespaceHandlerSupport {

net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java

@@ -32,7 +32,7 @@ import org.springframework.beans.factory.xml.ParserContext;
 import org.w3c.dom.Element;
 
 /**
- * @author jaysunxiao
+ * @author godotg
  * @version 3.0
  */
 public class NetDefinitionParser implements BeanDefinitionParser {
@@ -179,7 +179,6 @@ public class NetDefinitionParser implements BeanDefinitionParser {
         var clazz = ProviderConfig.class;
         var builder = BeanDefinitionBuilder.rootBeanDefinition(clazz);
 
-        resolvePlaceholder("task-dispatch", "taskDispatch", builder, element, parserContext);
         resolvePlaceholder("thread", "thread", builder, element, parserContext);
         resolvePlaceholder("address", "address", builder, element, parserContext);
 

net/src/main/java/com/zfoo/net/task/PacketReceiverTask.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
  *
@@ -11,7 +10,7 @@
  * 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;
 
 import com.zfoo.net.NetContext;
 import com.zfoo.net.router.attachment.IAttachment;
@@ -19,7 +18,7 @@ import com.zfoo.net.session.model.Session;
 import com.zfoo.protocol.IPacket;
 
 /**
- * @author jaysunxiao
+ * @author godotg
  * @version 3.0
  */
 public final class PacketReceiverTask implements Runnable {

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

@@ -15,9 +15,7 @@ package com.zfoo.net.task;
 
 import com.zfoo.event.manager.EventBus;
 import com.zfoo.net.NetContext;
-import com.zfoo.net.task.dispatcher.AbstractTaskDispatch;
-import com.zfoo.net.task.dispatcher.ITaskDispatch;
-import com.zfoo.net.task.model.PacketReceiverTask;
+import com.zfoo.net.session.model.AttributeType;
 import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject;
 import com.zfoo.protocol.util.AssertionUtils;
 import com.zfoo.protocol.util.StringUtils;
@@ -29,8 +27,10 @@ import io.netty.util.concurrent.FastThreadLocalThread;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Map;
-import java.util.concurrent.*;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -46,9 +46,6 @@ public final class TaskBus {
     // 线程池的大小，也可以通过provider thread配置指定
     public static final int EXECUTOR_SIZE;
 
-    private static final ITaskDispatch taskDispatch;
-
-
     /**
      * 使用不同的线程池，让线程池之间实现隔离，互不影响
      */
@@ -58,11 +55,7 @@ public final class TaskBus {
         var localConfig = NetContext.getConfigManager().getLocalConfig();
         var providerConfig = localConfig.getProvider();
 
-        taskDispatch = AbstractTaskDispatch.valueOf(providerConfig == null ? "consistent-hash" : providerConfig.getTaskDispatch());
-
-        EXECUTOR_SIZE = (providerConfig == null || StringUtils.isBlank(providerConfig.getThread()))
-                ? (Runtime.getRuntime().availableProcessors() + 1)
-                : Integer.parseInt(providerConfig.getThread());
+        EXECUTOR_SIZE = (providerConfig == null || StringUtils.isBlank(providerConfig.getThread())) ? (Runtime.getRuntime().availableProcessors() + 1) : Integer.parseInt(providerConfig.getThread());
 
         executors = new ExecutorService[EXECUTOR_SIZE];
         for (int i = 0; i < executors.length; i++) {
@@ -116,9 +109,20 @@ public final class TaskBus {
      * GatewayAttachment：默认是executorConsistentHash等于用户活玩家的uid，也可以通过IGatewayLoadBalancer接口指定
      * SignalAttachment：executorConsistentHash通过IRouter和IConsumer的argument参数指定
      */
-    public static void submit(PacketReceiverTask task) {
-        // 里面会看到是：其中一致性hash是根据附加包记录的hashId进行选择哪个线程进行业务处理
-        taskDispatch.getExecutor(executors, task).execute(task);
+    public static void dispatch(PacketReceiverTask task) {
+        var attachment = task.getAttachment();
+
+        if (attachment == null) {
+            var session = task.getSession();
+            var uid = session.getAttribute(AttributeType.UID);
+            if (uid == null) {
+                execute((int) session.getSid(), task);
+            } else {
+                execute((int) uid, task);
+            }
+        } else {
+            execute(attachment.executorConsistentHash(), task);
+        }
     }
 
     public static int executorIndex(int executorConsistentHash) {

net/src/main/java/com/zfoo/net/task/dispatcher/AbstractTaskDispatch.java

@@ -1,37 +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.dispatcher;
-
-import com.zfoo.protocol.util.StringUtils;
-
-/**
- * @author jaysunxiao
- * @version 3.0
- */
-public abstract class AbstractTaskDispatch implements ITaskDispatch {
-
-    public static ITaskDispatch valueOf(String taskDispatchName) {
-        switch (taskDispatchName) {
-            case "random":
-                return new RandomTaskDispatch();
-            case "sessionId":
-                return new SessionIdTaskDispatch();
-            case "consistent-hash":
-                return new ConsistentHashTaskDispatch();
-            default:
-                throw new RuntimeException(StringUtils.format("没有找到对应的taskDispatch[{}]", taskDispatchName));
-        }
-    }
-
-}

net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java

@@ -1,55 +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.dispatcher;
-
-import com.zfoo.net.session.model.AttributeType;
-import com.zfoo.net.task.TaskBus;
-import com.zfoo.net.task.model.PacketReceiverTask;
-import com.zfoo.util.math.HashUtils;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-
-/**
- * @author godotg
- * @version 3.0
- */
-public class ConsistentHashTaskDispatch extends AbstractTaskDispatch {
-
-    private static final ConsistentHashTaskDispatch INSTANCE = new ConsistentHashTaskDispatch();
-
-    public static ConsistentHashTaskDispatch getINSTANCE() {
-        return INSTANCE;
-    }
-
-    @Override
-    public Executor getExecutor(ExecutorService[] executors, PacketReceiverTask packetReceiverTask) {
-        var attachment = packetReceiverTask.getAttachment();
-
-        if (attachment == null) {
-            var session = packetReceiverTask.getSession();
-            var uid = session.getAttribute(AttributeType.UID);
-
-            if (uid == null) {
-                return SessionIdTaskDispatch.getInstance().getExecutor(executors, packetReceiverTask);
-            } else {
-                return executors[TaskBus.executorIndex(HashUtils.fnvHash(uid))];
-            }
-        }
-
-        // 可见最终是根据附加包的信息选择服务端由哪个线程执行这个业务
-        return executors[TaskBus.executorIndex(attachment.executorConsistentHash())];
-    }
-
-}

net/src/main/java/com/zfoo/net/task/dispatcher/ITaskDispatch.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.dispatcher;
-
-import com.zfoo.net.task.model.PacketReceiverTask;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-
-/**
- * @author godotg
- * @version 3.0
- */
-public interface ITaskDispatch {
-
-    Executor getExecutor(ExecutorService[] executors, PacketReceiverTask packetReceiverTask);
-
-}

net/src/main/java/com/zfoo/net/task/dispatcher/RandomTaskDispatch.java

@@ -1,40 +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.dispatcher;
-
-import com.zfoo.net.task.TaskBus;
-import com.zfoo.net.task.model.PacketReceiverTask;
-import com.zfoo.util.math.RandomUtils;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-
-/**
- * @author godotg
- * @version 3.0
- */
-public class RandomTaskDispatch extends AbstractTaskDispatch {
-
-    private static final RandomTaskDispatch INSTANCE = new RandomTaskDispatch();
-
-    public static RandomTaskDispatch getInstance() {
-        return INSTANCE;
-    }
-
-    @Override
-    public Executor getExecutor(ExecutorService[] executors, PacketReceiverTask packetReceiverTask) {
-        return executors[TaskBus.executorIndex(RandomUtils.randomInt())];
-    }
-
-}

net/src/main/java/com/zfoo/net/task/dispatcher/SessionIdTaskDispatch.java

@@ -1,43 +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.dispatcher;
-
-import com.zfoo.net.task.TaskBus;
-import com.zfoo.net.task.model.PacketReceiverTask;
-import com.zfoo.util.math.HashUtils;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
-
-/**
- * 同一个session总是分配到同一个线程池执行
- *
- * @author godotg
- * @version 3.0
- */
-public class SessionIdTaskDispatch extends AbstractTaskDispatch {
-
-    private static final SessionIdTaskDispatch INSTANCE = new SessionIdTaskDispatch();
-
-    public static SessionIdTaskDispatch getInstance() {
-        return INSTANCE;
-    }
-
-    @Override
-    public Executor getExecutor(ExecutorService[] executors, PacketReceiverTask packetReceiverTask) {
-        var session = packetReceiverTask.getSession();
-        return executors[TaskBus.executorIndex(HashUtils.fnvHash(session.getSid()))];
-    }
-
-}

net/src/main/resources/net-1.0.xsd

@@ -27,7 +27,6 @@
         <xsd:sequence>
             <xsd:element name="provider" maxOccurs="unbounded" type="providerAttributeType" minOccurs="0"/>
         </xsd:sequence>
-        <xsd:attribute name="task-dispatch" type="xsd:string" default="consistent-hash"/>
         <xsd:attribute name="thread" type="xsd:string" use="optional"/>
         <xsd:attribute name="address" type="xsd:string" use="optional"/>
     </xsd:complexType>

net/src/test/resources/provider/provider_config.xml

@@ -25,7 +25,7 @@
         </net:registry>
 
         <!--1.这里声明自己是服务提供者 2.提供接口是providerTest模块下的接口 3.提供者的名字是myProvider1-->
-        <net:providers task-dispatch="consistent-hash">
+        <net:providers>
             <net:provider protocol-module="providerTest" provider="myProvider1"/>
             <net:provider protocol-module="providerTest" provider="myProvider2"/>
         </net:providers>