im-router根据请求uid一致性hash负载均衡

README.md

@@ -1,3 +1,18 @@
+TODO 
+- router -> entry 负载均衡
+- router 新增节点顺时针相邻节点数据一致性哈希迁移
+- router 冗余节点存储数据不提供服务
+- dubbo 服务异步处理吞吐量
+- das 消息队列异步写
+- 功能开发：
+    - 消息群发
+    - 好友列表(在线状态:批量uid一致性hash, router查询)
+    - 聊天记录查询
+- 集群部署, docker swarm, k8s, jenkens
+- 服务监控
+
+
+模块列表
 - 接入层 entry
 - 逻辑层 logic
 - 内存存储层 router

im-common/src/main/java/net/sopod/soim/common/constant/DubboConstant.java

@@ -0,0 +1,16 @@
+package net.sopod.soim.common.constant;
+
+/**
+ * DubboConstant
+ *
+ * @author tmy
+ * @date 2022-05-01 15:26
+ */
+public interface DubboConstant {
+
+    /**
+     * 服务调用上下文用户id
+     */
+    String CTX_UID = "uid";
+
+}

im-common/src/main/java/net/sopod/soim/common/util/Func.java

@@ -0,0 +1,11 @@
+package net.sopod.soim.common.util;
+
+/**
+ * Func
+ *
+ * @author tmy
+ * @date 2022-05-01 15:28
+ */
+public class Func {
+
+}

im-common/src/main/java/net/sopod/soim/common/util/HashAlgorithms.java

@@ -1,8 +1,12 @@
 package net.sopod.soim.common.util;
 
+import java.io.UnsupportedEncodingException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
 /**
  * hash函数
- * https://zhuanlan.zhihu.com/p/101390996
  *
  * @author tangmingyou
  * @date 2021-10-28 10:33
@@ -10,94 +14,46 @@ package net.sopod.soim.common.util;
 public class HashAlgorithms {
 
     /**
-     * 加法hash
+     * MD5 Hash
-     *
-     * @param key   字符串
-     * @param prime 一个质数
-     * @return hash结果
      */
-    public static int additiveHash(String key, int prime) {
+    public static long md5Hash(String value) {
-        int hash, i;
+        MessageDigest md5;
-        for (hash = key.length(), i = 0; i < key.length(); i++)
+        try {
-            hash += key.charAt(i);
+            md5 = MessageDigest.getInstance("MD5");
-        return (hash % prime);
+        } catch (NoSuchAlgorithmException e) {
+            throw new RuntimeException("MD5 not supported", e);
         }
-
+        md5.reset();
-    /**
+        byte[] keyBytes = value.getBytes(StandardCharsets.UTF_8);
-     * 旋转hash
+        md5.update(keyBytes);
-     *
+        byte[] digest = md5.digest();
-     * @param key   输入字符串
+        // hash code, Truncate to 32-bits
-     * @param prime 质数
+        long hashCode = ((long) (digest[3] & 0xFF) << 24)
-     * @return hash值
+                | ((long) (digest[2] & 0xFF) << 16)
-     */
+                | ((long) (digest[1] & 0xFF) << 8)
-    public static int rotatingHash(String key, int prime) {
+                | (digest[0] & 0xFF);
-        int hash, i;
+        return hashCode & 0xffffffffL;
-        for (hash = key.length(), i = 0; i < key.length(); ++i)
-            hash = (hash << 4) ^ (hash >> 28) ^ key.charAt(i);
-        return (hash % prime);
-        //  return (hash ^ (hash>>10) ^ (hash>>20));
     }
-    // 替代：
-    // 使用：hash = (hash ^ (hash>>10) ^ (hash>>20)) & mask;
-    // 替代：hash %= prime;
-    /**/
-    /**
-     * MASK值，随便找一个值，最好是质数
-     */
-    static int M_MASK = 0x8765fed1;
 
-    /**
+    public static long md5Hash(String value, int number) {
-     * 一次一个hash
+        MessageDigest md5;
-     *
+        try {
-     * @param key 输入字符串
+            md5 = MessageDigest.getInstance("MD5");
-     * @return 输出hash值
+        } catch (NoSuchAlgorithmException e) {
-     */
+            throw new RuntimeException("MD5 not supported", e);
-    public static int oneByOneHash(String key) {
-        int hash, i;
-        for (hash = 0, i = 0; i < key.length(); ++i) {
-            hash += key.charAt(i);
-            hash += (hash << 10);
-            hash ^= (hash >> 6);
         }
-        hash += (hash << 3);
+        md5.reset();
-        hash ^= (hash >> 11);
+        byte[] keyBytes = value.getBytes(StandardCharsets.UTF_8);
-        hash += (hash << 15);
+        md5.update(keyBytes);
-        //  return (hash & M_MASK);
+        byte[] digest = md5.digest();
-        return hash;
+        // dubbo md5 hash algorithms
+        return (((long) (digest[3 + number * 4] & 0xFF) << 24)
+                | ((long) (digest[2 + number * 4] & 0xFF) << 16)
+                | ((long) (digest[1 + number * 4] & 0xFF) << 8)
+                | (digest[number * 4] & 0xFF))
+                & 0xFFFFFFFFL;
     }
 
-    /**
-     * Bernstein's hash
-     *
-     * @param key 输入字节数组
-     * @return 结果hash
-     */
-    public static int bernstein(String key) {
-        int hash = 0;
-        int i;
-        for (i = 0; i < key.length(); ++i) hash = 33 * hash + key.charAt(i);
-        return hash;
-    }
-
-    //
-    /**///// Pearson's Hash
-    // char pearson(char[]key, ub4 len, char tab[256])
-    // {
-    //  char hash;
-    //  ub4 i;
-    //  for (hash=len, i=0; i<len; ++i)
-    //   hash=tab[hash^key[i]];
-    //  return (hash);
-    // }
-    /**///// CRC Hashing，计算crc,具体代码见其他
-    // ub4 crc(char *key, ub4 len, ub4 mask, ub4 tab[256])
-    // {
-    //  ub4 hash, i;
-    //  for (hash=len, i=0; i<len; ++i)
-    //   hash = (hash >> 8) ^ tab[(hash & 0xff) ^ key[i]];
-    //  return (hash & mask);
-    // }
-    /**/
 
     /**
      * CRC系列算法本身并非查表，可是，查表是它的一种最快的实现方式。以下是CRC32的实现
@@ -108,14 +64,17 @@ public class HashAlgorithms {
 
         for (int n = 0; n != 256; ++n) {
             c = n;
+            for (int i = 0; i < 8; i++) {
                 c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+            }
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
-            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
+//            c = ((c & 1) != 0 ? (-306674912 ^ (c >>> 1)) : (c >>> 1));
             table[n] = c;
         }
         return table;
@@ -151,246 +110,4 @@ public class HashAlgorithms {
         return ~C;
     }
 
-    /**
-     * Universal Hashing
-     */
-    public static int universal(char[] key, int mask, int[] tab) {
-        int hash = key.length, i, len = key.length;
-        for (i = 0; i < (len << 3); i += 8) {
-            char k = key[i >> 3];
-            if ((k & 0x01) == 0) hash ^= tab[i + 0];
-            if ((k & 0x02) == 0) hash ^= tab[i + 1];
-            if ((k & 0x04) == 0) hash ^= tab[i + 2];
-            if ((k & 0x08) == 0) hash ^= tab[i + 3];
-            if ((k & 0x10) == 0) hash ^= tab[i + 4];
-            if ((k & 0x20) == 0) hash ^= tab[i + 5];
-            if ((k & 0x40) == 0) hash ^= tab[i + 6];
-            if ((k & 0x80) == 0) hash ^= tab[i + 7];
-        }
-        return (hash & mask);
-    }
-
-    /**
-     * Zobrist Hashing
-     */
-    public static int zobrist(char[] key, int mask, int[][] tab) {
-        int hash, i;
-        for (hash = key.length, i = 0; i < key.length; ++i)
-            hash ^= tab[i][key[i]];
-        return (hash & mask);
-    }
-
-    // LOOKUP3
-    // 见Bob Jenkins(3).c文件
-    // 32位FNV算法
-    static int M_SHIFT = 0;
-
-    /**
-     * 32位的FNV算法
-     *
-     * @param data 数组
-     * @return int值
-     */
-    public static int FNVHash(byte[] data) {
-        int hash = (int) 2166136261L;
-        for (byte b : data)
-            hash = (hash * 16777619) ^ b;
-        if (M_SHIFT == 0)
-            return hash;
-        return (hash ^ (hash >> M_SHIFT)) & M_MASK;
-    }
-
-    /**
-     * 改进的32位FNV算法1
-     *
-     * @param data 数组
-     * @return int值
-     */
-    public static int FNVHash1(byte[] data) {
-        final int p = 16777619;
-        int hash = (int) 2166136261L;
-        for (byte b : data)
-            hash = (hash ^ b) * p;
-        hash += hash << 13;
-        hash ^= hash >> 7;
-        hash += hash << 3;
-        hash ^= hash >> 17;
-        hash += hash << 5;
-        return hash;
-    }
-
-    /**
-     * FNV-1a HASH
-     * 改进的32位FNV算法1
-     * <p>
-     * costarring 与...碰撞 liquid
-     * declinate 与...碰撞 macallums
-     * altarage 与...碰撞 zinke
-     * altarages 与...碰撞 zinkes
-     *
-     * @param data 字符串
-     * @return int值
-     */
-    public static int FNVHash1a(String data) {
-        final int p = 16777619;
-        int hash = (int) 2166136261L;
-        for (int i = 0; i < data.length(); i++)
-            hash = (hash ^ data.charAt(i)) * p;
-        hash += hash << 13;
-        hash ^= hash >> 7;
-        hash += hash << 3;
-        hash ^= hash >> 17;
-        hash += hash << 5;
-        return hash;
-    }
-
-    /**
-     * Thomas Wang的算法，整数hash
-     */
-    public static int intHash(int key) {
-        key += ~(key << 15);
-        key ^= (key >>> 10);
-        key += (key << 3);
-        key ^= (key >>> 6);
-        key += ~(key << 11);
-        key ^= (key >>> 16);
-        return key;
-    }
-
-    /**
-     * RS算法hash
-     *
-     * @param str 字符串
-     */
-    public static int RSHash(String str) {
-        int b = 378551;
-        int a = 63689;
-        int hash = 0;
-        for (int i = 0; i < str.length(); i++) {
-            hash = hash * a + str.charAt(i);
-            a = a * b;
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-    /* End Of RS Hash Function */
-
-    /**
-     * JS算法
-     */
-    public static int JSHash(String str) {
-        int hash = 1315423911;
-        for (int i = 0; i < str.length(); i++) {
-            hash ^= ((hash << 5) + str.charAt(i) + (hash >> 2));
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-    /**//* End Of JS Hash Function */
-
-    /**
-     * PJW算法
-     */
-    public static int PJWHash(String str) {
-        int BitsInUnsignedInt = 32;
-        int ThreeQuarters = (BitsInUnsignedInt * 3) / 4;
-        int OneEighth = BitsInUnsignedInt / 8;
-        int HighBits = 0xFFFFFFFF << (BitsInUnsignedInt - OneEighth);
-        int hash = 0;
-        int test = 0;
-        for (int i = 0; i < str.length(); i++) {
-            hash = (hash << OneEighth) + str.charAt(i);
-            if ((test = hash & HighBits) != 0) {
-                hash = ((hash ^ (test >> ThreeQuarters)) & (~HighBits));
-            }
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-    /* End Of P. J. Weinberger Hash Function */
-
-    /**
-     * ELF算法
-     */
-    public static int ELFHash(String str) {
-        int hash = 0;
-        int x = 0;
-        for (int i = 0; i < str.length(); i++) {
-            hash = (hash << 4) + str.charAt(i);
-            if ((x = (int) (hash & 0xF0000000L)) != 0) {
-                hash ^= (x >> 24);
-                hash &= ~x;
-            }
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-
-    /**
-     * BKDR算法
-     */
-    public static int BKDRHash(String str) {
-        int seed = 131; // 31 131 1313 13131 131313 etc..
-        int hash = 0;
-        for (int i = 0; i < str.length(); i++) {
-            hash = (hash * seed) + str.charAt(i);
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-
-    /**
-     * SDBM算法
-     */
-    public static int SDBMHash(String str) {
-        int hash = 0;
-        for (int i = 0; i < str.length(); i++) {
-            hash = str.charAt(i) + (hash << 6) + (hash << 16) - hash;
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-
-    /**
-     * DJB算法
-     */
-    public static int DJBHash(String str) {
-        int hash = 5381;
-        for (int i = 0; i < str.length(); i++) {
-            hash = ((hash << 5) + hash) + str.charAt(i);
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-
-    /**
-     * DEK算法
-     */
-    public static int DEKHash(String str) {
-        int hash = str.length();
-        for (int i = 0; i < str.length(); i++) {
-            hash = ((hash << 5) ^ (hash >> 27)) ^ str.charAt(i);
-        }
-        return (hash & 0x7FFFFFFF);
-    }
-
-    /**
-     * AP算法
-     */
-    public static int APHash(String str) {
-        int hash = 0;
-        for (int i = 0; i < str.length(); i++) {
-            hash ^= ((i & 1) == 0) ? ((hash << 7) ^ str.charAt(i) ^ (hash >> 3)) :
-                    (~((hash << 11) ^ str.charAt(i) ^ (hash >> 5)));
-        }
-        //    return (hash & 0x7FFFFFFF);
-        return hash;
-    }
-
-    /**
-     * JAVA自己带的算法
-     */
-    public static int java(String str) {
-        int h = 0;
-        int off = 0;
-        int len = str.length();
-        for (int i = 0; i < len; i++) {
-            h = 31 * h + str.charAt(off++);
-        }
-        return h;
-    }
-
 }

im-core/pom.xml

@@ -31,6 +31,11 @@
             <artifactId>spring-context</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo</artifactId>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
 </project>

im-core/src/main/java/net/sopod/soim/core/handler/ProtoMessageHandler.java

@@ -1,26 +0,0 @@
-package net.sopod.soim.core.handler;
-
-import com.google.protobuf.MessageLite;
-import net.sopod.soim.core.session.NetUser;
-
-/**
- * ProtoMessageHandler
- * <T extends MessageLite>
- *
- * @author tmy
- * @date 2022-04-10 19:19
- */
-public abstract class ProtoMessageHandler<T> {
-
-    public final void exec(NetUser netUser, T msg) {
-        MessageLite res = handle(msg);
-        if (res != null) {
-            netUser.write(res);
-        }
-    }
-
-    public abstract Class<T> type();
-
-    public abstract MessageLite handle(T msg);
-
-}

im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java

@@ -0,0 +1,31 @@
+package net.sopod.soim.entry.config;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import org.apache.dubbo.rpc.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * DubboPreInvokeFilter
+ * 每次调用服务前设置上下文属性
+ *
+ * @author tmy
+ * @date 2022-05-01 08:55
+ */
+public class DubboPreInvokeFilter implements Filter {
+
+    private static final Logger logger = LoggerFactory.getLogger(DubboPreInvokeFilter.class);
+
+    @Override
+    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
+        String uid = MessageHandlerContext.getAttribute(DubboConstant.CTX_UID);
+        if (uid != null) {
+            RpcContext.getServiceContext().setAttachment(DubboConstant.CTX_UID, uid);
+            if (logger.isDebugEnabled()) {
+                logger.debug("pre invoke filter set uid: {}", uid);
+            }
+        }
+        return invoker.invoke(invocation);
+    }
+
+}

im-entry/src/main/java/net/sopod/soim/entry/config/MessageHandlerContext.java

@@ -0,0 +1,53 @@
+package net.sopod.soim.entry.config;
+
+import io.netty.util.concurrent.FastThreadLocal;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * MsgHandlerContext
+ *
+ * 线程上下文属性设置工具
+ * 这里主要用于 {@link net.sopod.soim.core.handler.MessageHandler} 与 dubbo RpcContext 上下文参数传递
+ *
+ * @author tmy
+ * @date 2022-05-01 22:42
+ */
+public class MessageHandlerContext {
+
+    private static final FastThreadLocal<Map<String, String>> HOLDER = new FastThreadLocal<>();
+
+    public static void setAttribute(String key, String value) {
+        Map<String, String> attrMap;
+        if ((attrMap = HOLDER.get()) == null) {
+            synchronized (Thread.currentThread()) {
+                if ((attrMap = HOLDER.get()) == null) {
+                    attrMap = new ConcurrentHashMap<>(3);
+                    HOLDER.set(attrMap);
+                }
+            }
+        }
+        attrMap.put(key, value);
+    }
+
+    public static String getAttribute(String key) {
+        Map<String, String> attrMap;
+        if ((attrMap = HOLDER.get()) != null) {
+            return attrMap.get(key);
+        }
+        return null;
+    }
+
+    public static void remove() {
+        HOLDER.remove();
+    }
+
+    public static void removeAttribute(String key) {
+        Map<String, String> attrMap;
+        if ((attrMap = HOLDER.get()) != null) {
+            attrMap.remove(key);
+        }
+    }
+
+}

im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextHolder.java

@@ -8,12 +8,12 @@ import org.springframework.context.ApplicationContext;
  * @author tmy
  * @date 2022-04-28 15:06
  */
-public class ContextHolder {
+public class SpringContextHolder {
 
     private static ApplicationContext applicationContext;
 
     public static void setContext(ApplicationContext applicationContext) {
-        ContextHolder.applicationContext = applicationContext;
+        SpringContextHolder.applicationContext = applicationContext;
     }
 
     public static <T> T getBean(Class<T> beanType) {

im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextInitialed.java

@@ -13,11 +13,11 @@ import org.springframework.context.annotation.Configuration;
  * @date 2022-04-10 22:20
  */
 @Configuration
-public class ApplicationContextInitialed implements ApplicationContextAware {
+public class SpringContextInitialed implements ApplicationContextAware {
 
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        ContextHolder.setContext(applicationContext);
+        SpringContextHolder.setContext(applicationContext);
 
         // 注册 protobuf 消息 handler
         ProtoMessageHandlerRegistry.registerHandlerWithApplicationContext(applicationContext);

im-entry/src/main/java/net/sopod/soim/entry/handler/AccountMessageHandler.java

@@ -1,11 +1,14 @@
-package net.sopod.soim.core.handler;
+package net.sopod.soim.entry.handler;
 
 import com.google.protobuf.MessageLite;
+import net.sopod.soim.core.handler.MessageHandler;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.core.session.NetUser;
+import org.apache.dubbo.rpc.RpcContext;
 
 /**
  * AccountMessageHandler
+ * 每次调用服务前
  *
  * @author tmy
  * @date 2022-04-10 23:41

im-entry/src/main/java/net/sopod/soim/entry/handler/NetUserMessageHandler.java

@@ -1,6 +1,7 @@
-package net.sopod.soim.core.handler;
+package net.sopod.soim.entry.handler;
 
 import com.google.protobuf.MessageLite;
+import net.sopod.soim.core.handler.MessageHandler;
 import net.sopod.soim.core.session.NetUser;
 
 /**

im-entry/src/main/java/net/sopod/soim/entry/handler/auth/ReqTokenAuthHandler.java

@@ -1,7 +1,7 @@
 package net.sopod.soim.entry.handler.auth;
 
 import com.google.protobuf.MessageLite;
-import net.sopod.soim.core.handler.NetUserMessageHandler;
+import net.sopod.soim.entry.handler.NetUserMessageHandler;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.core.session.NetUser;
 import net.sopod.soim.data.msg.auth.Auth;
@@ -44,7 +44,8 @@ public class ReqTokenAuthHandler extends NetUserMessageHandler<Auth.ReqTokenAuth
     @Override
     public MessageLite handle(NetUser netUser, Auth.ReqTokenAuth msg) {
         logger.info("ReqTokenAuth: uid={}", msg.getUid());
-        // 校验 token
+        // 校验 token，
+        // TODO 更新 router 状态成功后再升级为 account，主动关闭时发送一个消息到客户端，校验错误/超时
         Boolean isValid = userAuthService.validateToken(msg.getToken(),
                 entryServerConfig.getIp() + ":" + entryServerConfig.getPort());
 

im-entry/src/main/java/net/sopod/soim/entry/handler/chat/ReqTextChatHandler.java

@@ -1,7 +1,7 @@
 package net.sopod.soim.entry.handler.chat;
 
 import com.google.protobuf.MessageLite;
-import net.sopod.soim.core.handler.AccountMessageHandler;
+import net.sopod.soim.entry.handler.AccountMessageHandler;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.data.msg.chat.Chat;
 import net.sopod.soim.logic.common.model.TextChat;

im-entry/src/main/java/net/sopod/soim/entry/handler/task/NetUserDelayCloseHandler.java

@@ -1,7 +1,7 @@
 package net.sopod.soim.entry.handler.task;
 
 import com.google.protobuf.MessageLite;
-import net.sopod.soim.core.handler.NetUserMessageHandler;
+import net.sopod.soim.entry.handler.NetUserMessageHandler;
 import net.sopod.soim.core.session.NetUser;
 import net.sopod.soim.data.msg.task.Tasks;
 import org.springframework.stereotype.Service;

im-entry/src/main/java/net/sopod/soim/entry/handler/user/HelloHandler.java

@@ -1,12 +1,10 @@
-package net.sopod.soim.entry.handler;
+package net.sopod.soim.entry.handler.user;
 
 import com.google.protobuf.MessageLite;
-import net.sopod.soim.core.handler.AccountMessageHandler;
-import net.sopod.soim.core.handler.NetUserMessageHandler;
 import net.sopod.soim.core.session.Account;
-import net.sopod.soim.core.session.NetUser;
 import net.sopod.soim.data.msg.hello.HelloPB;
 import net.sopod.soim.entry.delay.NetUserDelayTaskManager;
+import net.sopod.soim.entry.handler.AccountMessageHandler;
 import net.sopod.soim.logic.api.segmentid.core.SegmentIdGenerator;
 import net.sopod.soim.logic.user.service.UserService;
 import org.apache.dubbo.config.annotation.DubboReference;

im-entry/src/main/java/net/sopod/soim/entry/handler/user/ReqOnlineUserListHandler.java

@@ -1,12 +1,14 @@
 package net.sopod.soim.entry.handler.user;
 
 import com.google.protobuf.MessageLite;
-import net.sopod.soim.core.handler.AccountMessageHandler;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.data.msg.user.UserGroup;
-import net.sopod.soim.logic.user.service.UserService;
+import net.sopod.soim.entry.handler.AccountMessageHandler;
 import net.sopod.soim.logic.common.model.UserInfo;
+import net.sopod.soim.logic.user.service.UserService;
 import org.apache.dubbo.config.annotation.DubboReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
@@ -21,12 +23,15 @@ import java.util.stream.Collectors;
 @Service
 public class ReqOnlineUserListHandler extends AccountMessageHandler<UserGroup.ReqOnlineUserList> {
 
+    private static final Logger logger = LoggerFactory.getLogger(ReqOnlineUserListHandler.class);
+
     @DubboReference
     private UserService userService;
 
     @Override
     public MessageLite handle(Account account, UserGroup.ReqOnlineUserList msg) {
         List<UserInfo> userInfos = userService.onlineUserList(msg.getKeyword());
+
         List<UserGroup.UserInfo> resUserInfos = userInfos.stream().map(user -> UserGroup.UserInfo.newBuilder()
                 .setUid(user.getUid())
                 .setAccount(user.getAccount())

im-entry/src/main/java/net/sopod/soim/entry/server/ProtoMessageDispatcher.java

@@ -1,9 +1,12 @@
 package net.sopod.soim.entry.server;
 
 import com.google.protobuf.MessageLite;
+import net.sopod.soim.common.constant.DubboConstant;
 import net.sopod.soim.core.handler.MessageHandler;
 import net.sopod.soim.core.registry.ProtoMessageHandlerRegistry;
+import net.sopod.soim.core.session.Account;
 import net.sopod.soim.core.session.NetUser;
+import net.sopod.soim.entry.config.MessageHandlerContext;
 import net.sopod.soim.entry.worker.Worker;
 import net.sopod.soim.entry.worker.WorkerGroup;
 
@@ -18,8 +21,21 @@ public class ProtoMessageDispatcher {
     public static void dispatch(NetUser netUser, MessageLite message) {
         MessageHandler<MessageLite> typeHandler = ProtoMessageHandlerRegistry
                 .getTypeHandler(message.getClass());
+        if (typeHandler == null) {
+            throw new IllegalCallerException("no handler for message : " + message.getClass());
+        }
         Worker worker = WorkerGroup.next();
-        worker.execute(() -> { typeHandler.exec(netUser, message); });
+        worker.execute(() -> {
+            if (netUser.isAccount()) {
+                Account account = (Account)netUser;
+                MessageHandlerContext.setAttribute(DubboConstant.CTX_UID, String.valueOf(account.getUid()));
+            }
+            try {
+                typeHandler.exec(netUser, message);
+            } finally {
+                MessageHandlerContext.remove();
+            }
+        });
     }
 
 }

im-entry/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter

@@ -0,0 +1 @@
+pre_invoke_filter=net.sopod.soim.entry.config.DubboPreInvokeFilter

im-entry/src/main/resources/application.yml

@@ -14,6 +14,7 @@ dubbo:
     port: 3008
   consumer:
     check: false
+    filter: pre_invoke_filter
 
 entry-server:
   nacos-addr: 124.222.131.236:3848

im-service-api/im-router-api/pom.xml

@@ -17,6 +17,11 @@
             <groupId>net.sopod</groupId>
             <version>1.0.0</version>
         </dependency>
+        <dependency>
+            <artifactId>im-common</artifactId>
+            <groupId>net.sopod</groupId>
+            <version>1.0.0</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-cluster</artifactId>

im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashRoute.java

@@ -1,26 +0,0 @@
-package net.sopod.soim.router.api.route;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.rpc.Invocation;
-import org.apache.dubbo.rpc.Invoker;
-import org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance;
-
-import java.util.List;
-
-/**
- * ConsistentHashRoute
- * 一致性 hash 服务路由
- *
- *
- * @author tmy
- * @date 2022-04-29 14:41
- */
-public class ConsistentHashRoute extends AbstractLoadBalance {
-
-    @Override
-    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
-        Invoker<T> invoker = invokers.get(0);
-        return null;
-    }
-
-}

im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashTest.java

@@ -0,0 +1,100 @@
+package net.sopod.soim.router.api.route;
+
+import net.sopod.soim.common.util.HashAlgorithms;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * ConsistentHashTest
+ *
+ * @author tmy
+ * @date 2022-05-01 15:33
+ */
+public class ConsistentHashTest {
+
+    private static final int VIRTUAL_NODE_SIZE = 120;
+
+    private TreeMap<Long, String> virtualInvokers = new TreeMap<>();
+
+    public void add(String key, String value) {
+        // 增加虚拟节点n个
+        for (int i = 0; i < VIRTUAL_NODE_SIZE / 4; i++) {
+            for (int h = 0; h < 4; h++) {
+                long virHash = hash(key + i);
+                virtualInvokers.put(virHash, value);
+            }
+//            long virHash = hash("vir-" + key + i);
+//            virtualInvokers.put(virHash, value);
+        }
+        // virtualInvokers.put(hash(key), value);
+    }
+
+    public String getFirstNode(String value) {
+        long hash = hash(value);
+        Map.Entry<Long, String> entry = virtualInvokers.ceilingEntry(hash);
+        if (entry != null) {
+            return entry.getValue();
+        }
+        if (virtualInvokers.size() == 0) {
+            throw new IllegalStateException("server is not available");
+        }
+        return virtualInvokers.firstEntry().getValue();
+    }
+
+    /**
+     * hash 运算
+     */
+    public static long hash(String value) {
+        return HashAlgorithms.md5Hash(value);
+    }
+
+    public static long hash(String value, int number) {
+        return HashAlgorithms.md5Hash(value, number);
+    }
+
+    private static void testConsistentHash() {
+        String[] nodes = {
+                "192.168.31.156:3032",
+                "192.168.31.156:3031",
+//                "192.168.1.101",
+//                "192.168.1.102",
+//                "192.168.1.103",
+//                "192.168.1.104",
+//                "192.168.1.105",
+        };
+        ConsistentHashTest route = new ConsistentHashTest();
+        for (String node : nodes) {
+            route.add(node, node);
+        }
+        Map<String, AtomicInteger> counter = new HashMap<>();
+        for (long i = 71000L; i < 72000L; i++) {
+            String node = route.getFirstNode(String.valueOf(i));
+            AtomicInteger count = counter.computeIfAbsent(node, k -> new AtomicInteger());
+            count.incrementAndGet();
+        }
+        System.out.println(route.virtualInvokers);
+        System.out.println(counter);
+    }
+
+    private static void testTreeMap() {
+        TreeMap<Integer, String> map = new TreeMap<>();
+        map.put(10, "沧海1");
+        map.put(20, "沧海2");
+        map.put(30, "沧海3");
+        map.put(40, "沧海4");
+        map.put(50, "沧海5");
+        System.out.println(map);
+        System.out.println(map.ceilingKey(30));
+        System.out.println(map.higherEntry(30));
+    }
+
+    public static void main(String[] args) {
+        // {192.168.1.103=32, 192.168.1.100=26, 192.168.1.101=19, 192.168.1.102=23}
+        testConsistentHash();
+    }
+
+
+}

im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java

@@ -0,0 +1,95 @@
+package net.sopod.soim.router.api.route;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.common.util.HashAlgorithms;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * ConsistentHashRoute
+ * 一致性 hash 服务路由
+ *
+ * {@link org.apache.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance}
+ *
+ * @author tmy
+ * @date 2022-04-29 14:41
+ */
+public class ImRouterConsistentHashRoute extends AbstractLoadBalance {
+
+    private static final Logger logger = LoggerFactory.getLogger(ImRouterConsistentHashRoute.class);
+
+    /**
+     * 一致性 hash 每个节点的虚拟节点数量
+     */
+    private static final int VIRTUAL_NODE_SIZE = 120;
+
+    private volatile ConsistentHashSelector<?> selector;
+
+    /**
+     * 后续如有接口版本号，构建 map 每个方法版本，对应一个 Selector
+     */
+    @Override
+    @SuppressWarnings("unchecked")
+    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        int invokersHash = getInvokersHash(invokers);
+        logger.info("invokers hash: {}", invokersHash);
+        if (selector == null || selector.identityHashCode != invokersHash) {
+            selector = new ConsistentHashSelector<>(invokers, invokersHash);
+        }
+        return ((ConsistentHashSelector<T>)selector).select(invocation);
+    }
+
+    private <T> int getInvokersHash(List<Invoker<T>> invokers) {
+        return invokers.hashCode();
+    }
+
+    static class ConsistentHashSelector<T> {
+
+        private final TreeMap<Long, Invoker<T>> virtualInvokers;
+
+        private final int identityHashCode;
+
+        ConsistentHashSelector(List<Invoker<T>> invokers, int identityHashCode) {
+            this.identityHashCode = identityHashCode;
+            this.virtualInvokers = new TreeMap<>();
+
+            for (Invoker<T> invoker : invokers) {
+                String address = invoker.getUrl().getAddress();
+                for (int i = 0, len = VIRTUAL_NODE_SIZE / 4; i < len; i++) {
+                    for (int h = 0; h < 4; h++) {
+                        long hash = hash(address + i, h);
+                        this.virtualInvokers.put(hash, invoker);
+                    }
+                }
+            }
+        }
+
+        private static long hash(String value, int number) {
+            return HashAlgorithms.md5Hash(value, number);
+        }
+
+        public Invoker<T> select(Invocation invocation) {
+            // 获取上下文 uid, 同 RpcContext
+            String uid = invocation.getAttachment(DubboConstant.CTX_UID);
+            if (uid == null) {
+                throw new IllegalStateException("im-router consistent hash route, ctx uid can not be null!");
+            }
+            long hash = hash(uid, 0);
+            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
+            if (entry == null) {
+                entry = virtualInvokers.firstEntry();
+            }
+            return entry.getValue();
+        }
+
+    }
+
+}

im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance

@@ -0,0 +1 @@
+im_route_consistent_hash=net.sopod.soim.router.api.route.ImRouterConsistentHashRoute

im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java

@@ -1,5 +1,6 @@
 package net.sopod.soim.logic.user.service;
 
+import net.sopod.soim.common.constant.DubboConstant;
 import net.sopod.soim.common.util.ImClock;
 import net.sopod.soim.common.util.TokenUtil;
 import net.sopod.soim.das.user.api.model.entity.ImUser;
@@ -11,6 +12,7 @@ import net.sopod.soim.router.api.model.CacheRes;
 import net.sopod.soim.router.api.service.UserEntryRegistryService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
+import org.apache.dubbo.rpc.RpcContext;
 import org.springframework.util.StringUtils;
 
 import javax.annotation.Resource;
@@ -65,6 +67,8 @@ public class UserAuthServiceImpl implements UserAuthService {
             return Boolean.FALSE;
         }
         // 注册记录用户登录的 entry 节点
+        RpcContext.getServiceContext()
+                .setAttachment(DubboConstant.CTX_UID, String.valueOf(payload.getUserId()));
         CacheRes cacheRes = userEntryRegistryService
                 .registryUserEntry(payload.getUserId(), imEntryAddr);
         return cacheRes.getSuccess();

im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserServiceImpl.java

@@ -4,6 +4,9 @@ import net.sopod.soim.logic.common.model.UserInfo;
 import net.sopod.soim.router.api.service.UserEntryRegistryService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
+import org.apache.dubbo.rpc.RpcContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -17,6 +20,8 @@ import java.util.concurrent.CompletableFuture;
 @DubboService
 public class UserServiceImpl implements UserService {
 
+    private static final Logger logger = LoggerFactory.getLogger(UserServiceImpl.class);
+
     @DubboReference
     private UserEntryRegistryService userEntryRegistryService;
 
@@ -32,6 +37,8 @@ public class UserServiceImpl implements UserService {
 
     @Override
     public List<UserInfo> onlineUserList(String keyword) {
+        logger.info("client context uid: {}", RpcContext.getClientAttachment().getAttachment("uid"));
+        logger.info("server context uid: {}", RpcContext.getServerAttachment().getAttachment("uid"));
         return userEntryRegistryService.onlineUserList(keyword);
     }
 

im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java

@@ -0,0 +1,25 @@
+package net.sopod.soim.router.listener;
+
+import org.apache.dubbo.rpc.Exporter;
+import org.apache.dubbo.rpc.ExporterListener;
+import org.apache.dubbo.rpc.RpcException;
+
+/**
+ * ImRouterExportListener
+ *
+ * @author tmy
+ * @date 2022-05-01 16:31
+ */
+public class ImRouterExportListener implements ExporterListener {
+
+    @Override
+    public void exported(Exporter<?> exporter) throws RpcException {
+
+    }
+
+    @Override
+    public void unexported(Exporter<?> exporter) {
+
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterServiceRegistry.java

@@ -0,0 +1,39 @@
+package net.sopod.soim.router.listener;
+
+import java.rmi.*;
+import java.rmi.registry.Registry;
+
+/**
+ * ImRouterServiceRegistry
+ *
+ * @author tmy
+ * @date 2022-05-01 16:34
+ */
+public class ImRouterServiceRegistry implements Registry {
+
+    @Override
+    public Remote lookup(String s) throws RemoteException, NotBoundException, AccessException {
+        return null;
+    }
+
+    @Override
+    public void bind(String s, Remote remote) throws RemoteException, AlreadyBoundException, AccessException {
+
+    }
+
+    @Override
+    public void unbind(String s) throws RemoteException, NotBoundException, AccessException {
+
+    }
+
+    @Override
+    public void rebind(String s, Remote remote) throws RemoteException, AccessException {
+
+    }
+
+    @Override
+    public String[] list() throws RemoteException, AccessException {
+        return new String[0];
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/service/UserEntryRegistryServiceImpl.java

@@ -10,6 +10,7 @@ import net.sopod.soim.router.api.model.CacheRes;
 import net.sopod.soim.router.api.model.RouterUser;
 import net.sopod.soim.logic.common.model.UserInfo;
 import net.sopod.soim.router.api.service.UserEntryRegistryService;
+import net.sopod.soim.router.util.ServerContext;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
 import org.slf4j.Logger;
@@ -56,6 +57,8 @@ public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
 
 
     public List<UserInfo> onlineUserList(String keyword) {
+        logger.info("client context uid: {}", ServerContext.getContextUid());
+
         Stream<RouterUser> stream = uidImEntryStore.values().stream();
         if (!StringUtil.isEmpty(keyword)) {
             // 根据关键词过滤

im-service/im-router/src/main/java/net/sopod/soim/router/util/ServerContext.java

@@ -0,0 +1,19 @@
+package net.sopod.soim.router.util;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import org.apache.dubbo.rpc.RpcContext;
+
+/**
+ * SessionContext
+ *
+ * @author tmy
+ * @date 2022-05-01 23:08
+ */
+public class ServerContext {
+
+    public static Long getContextUid() {
+        String uidStr = RpcContext.getServerContext().getAttachment(DubboConstant.CTX_UID);
+        return Long.valueOf(uidStr);
+    }
+
+}

im-service/im-router/src/main/resources/application.yml

@@ -20,8 +20,8 @@ dubbo:
     group: so-im
   protocol:
     name: dubbo
-    port: 3032
+    port: 3033
   consumer:
     check: false
   provider:
-    loadbalance: consistenthash
+    loadbalance: im_route_consistent_hash