From 1bec7081450b35f79cd022a502d1b3022657df49 Mon Sep 17 00:00:00 2001 From: jianan <1072772483@qq.com> Date: Thu, 28 Jul 2022 10:36:30 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix[module]=20=E4=BF=AE=E5=A4=8D=EF=BC=9A?= =?UTF-8?q?=E5=86=8D=E6=9C=89=E8=BF=9E=E6=8E=A5=E6=97=B6=EF=BC=8C=E4=BC=98?= =?UTF-8?q?=E5=85=88=E4=BB=A5uid=E4=BD=9C=E4=B8=BAhashId?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/java/com/zfoo/net/session/model/Session.java | 4 ++-- .../task/dispatcher/ConsistentHashTaskDispatch.java | 10 +++++++++- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/session/model/Session.java b/net/src/main/java/com/zfoo/net/session/model/Session.java index 54994588..9d0f268c 100644 --- a/net/src/main/java/com/zfoo/net/session/model/Session.java +++ b/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 getAttribute(AttributeType key) { + return (T) attributes.get(key); } public Channel getChannel() { diff --git a/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java index 17d0cd8b..e26a2994 100644 --- a/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java +++ b/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(Math.toIntExact(uid)); + } } // 可见最终是根据附加包的信息选择服务端由哪个线程执行这个业务 From 4dba2cc35f4b6acdaca11a42609b7b677f507bef Mon Sep 17 00:00:00 2001 From: jianan <1072772483@qq.com> Date: Thu, 28 Jul 2022 10:40:35 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix[module]=20=E4=BF=AE=E5=A4=8D=EF=BC=9A?= =?UTF-8?q?=E4=BC=A0=E5=85=A5Object=EF=BC=8C=E6=96=B9=E4=BE=BF=E5=8F=96has?= =?UTF-8?q?h?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- net/src/main/java/com/zfoo/net/task/TaskBus.java | 5 +++-- .../zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java | 2 +- .../com/zfoo/net/task/dispatcher/SessionIdTaskDispatch.java | 2 +- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/task/TaskBus.java b/net/src/main/java/com/zfoo/net/task/TaskBus.java index af12e816..84759249 100644 --- a/net/src/main/java/com/zfoo/net/task/TaskBus.java +++ b/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线程执行回调任务 diff --git a/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java index e26a2994..f723d63e 100644 --- a/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java +++ b/net/src/main/java/com/zfoo/net/task/dispatcher/ConsistentHashTaskDispatch.java @@ -42,7 +42,7 @@ public class ConsistentHashTaskDispatch extends AbstractTaskDispatch { if (uid == null) { return SessionIdTaskDispatch.getInstance().getExecutor(packetReceiverTask); } else { - return TaskBus.executor(Math.toIntExact(uid)); + return TaskBus.executor(uid); } } diff --git a/net/src/main/java/com/zfoo/net/task/dispatcher/SessionIdTaskDispatch.java b/net/src/main/java/com/zfoo/net/task/dispatcher/SessionIdTaskDispatch.java index 18e136dc..0a67eb39 100644 --- a/net/src/main/java/com/zfoo/net/task/dispatcher/SessionIdTaskDispatch.java +++ b/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()); } }