Merge branch 'main' of https://github.com/zfoo-project/zfoo

net/src/main/java/com/zfoo/net/session/model/Session.java

@@ -96,8 +96,8 @@ public class Session implements Closeable {
     }
 
 
-    public Object getAttribute(AttributeType key) {
-        return attributes.get(key);
+    public <T> T getAttribute(AttributeType key) {
+        return (T) attributes.get(key);
     }
 
     public Channel getChannel() {

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

@@ -19,6 +19,7 @@ import com.zfoo.net.task.dispatcher.ITaskDispatch;
 import com.zfoo.net.task.model.PacketReceiverTask;
 import com.zfoo.protocol.util.AssertionUtils;
 import com.zfoo.protocol.util.StringUtils;
+import com.zfoo.util.math.HashUtils;
 import com.zfoo.util.math.RandomUtils;
 import io.netty.util.concurrent.FastThreadLocalThread;
 import org.slf4j.Logger;
@@ -122,8 +123,8 @@ public final class TaskBus {
         taskDispatch.getExecutor(task).execute(task);
     }
 
-    public static ExecutorService executor(int executorConsistentHash) {
-        return executors[Math.abs(executorConsistentHash % EXECUTOR_SIZE)];
+    public static ExecutorService executor(Object hashObj) {
+        return executors[Math.abs(HashUtils.fnvHash(hashObj.hashCode()) % EXECUTOR_SIZE)];
     }
 
     // 在task线程的异步请求，请求成功过后依然在相同的task线程执行回调任务

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

@@ -13,6 +13,7 @@
 
 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;
 
@@ -35,7 +36,14 @@ public class ConsistentHashTaskDispatch extends AbstractTaskDispatch {
         var attachment = packetReceiverTask.getAttachment();
 
         if (attachment == null) {
-            return SessionIdTaskDispatch.getInstance().getExecutor(packetReceiverTask);
+            var session = packetReceiverTask.getSession();
+            Long uid = session.getAttribute(AttributeType.UID);
+
+            if (uid == null) {
+                return SessionIdTaskDispatch.getInstance().getExecutor(packetReceiverTask);
+            } else {
+                return TaskBus.executor(uid);
+            }
         }
 
         // 可见最终是根据附加包的信息选择服务端由哪个线程执行这个业务

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

@@ -36,7 +36,7 @@ public class SessionIdTaskDispatch extends AbstractTaskDispatch {
     @Override
     public ExecutorService getExecutor(PacketReceiverTask packetReceiverTask) {
         var session = packetReceiverTask.getSession();
-        return TaskBus.executor(HashUtils.fnvHash(session.getSid()));
+        return TaskBus.executor(session.getSid());
     }
 
 }