diff --git a/README.md b/README.md index a5dfffc..9cee990 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,18 @@ +TODO +- router -> entry 负载均衡 +- router 新增节点顺时针相邻节点数据一致性哈希迁移 +- router 冗余节点存储数据不提供服务 +- dubbo 服务异步处理吞吐量 +- das 消息队列异步写 +- 功能开发： + - 消息群发 + - 好友列表(在线状态:批量uid一致性hash, router查询) + - 聊天记录查询 +- 集群部署, docker swarm, k8s, jenkens +- 服务监控 + + +模块列表 - 接入层 entry - 逻辑层 logic - 内存存储层 router diff --git a/im-common/src/main/java/net/sopod/soim/common/constant/DubboConstant.java b/im-common/src/main/java/net/sopod/soim/common/constant/DubboConstant.java new file mode 100644 index 0000000..0f40b29 --- /dev/null +++ b/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"; + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/util/Func.java b/im-common/src/main/java/net/sopod/soim/common/util/Func.java new file mode 100644 index 0000000..392787f --- /dev/null +++ b/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 { + +} diff --git a/im-common/src/main/java/net/sopod/soim/common/util/HashAlgorithms.java b/im-common/src/main/java/net/sopod/soim/common/util/HashAlgorithms.java index 2b88c5c..c2760cc 100644 --- a/im-common/src/main/java/net/sopod/soim/common/util/HashAlgorithms.java +++ b/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 - * - * @param key 字符串 - * @param prime 一个质数 - * @return hash结果 - */ - public static int additiveHash(String key, int prime) { - int hash, i; - for (hash = key.length(), i = 0; i < key.length(); i++) - hash += key.charAt(i); - return (hash % prime); - } - - /** - * 旋转hash - * - * @param key 输入字符串 - * @param prime 质数 - * @return hash值 - */ - public static int rotatingHash(String key, int prime) { - int hash, i; - 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; - - /** - * 一次一个hash - * - * @param key 输入字符串 - * @return 输出hash值 + * MD5 Hash */ - 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); + public static long md5Hash(String value) { + MessageDigest md5; + try { + md5 = MessageDigest.getInstance("MD5"); + } catch (NoSuchAlgorithmException e) { + throw new RuntimeException("MD5 not supported", e); } - hash += (hash << 3); - hash ^= (hash >> 11); - hash += (hash << 15); - // return (hash & M_MASK); - return hash; - } - - /** - * 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; + md5.reset(); + byte[] keyBytes = value.getBytes(StandardCharsets.UTF_8); + md5.update(keyBytes); + byte[] digest = md5.digest(); + // hash code, Truncate to 32-bits + long hashCode = ((long) (digest[3] & 0xFF) << 24) + | ((long) (digest[2] & 0xFF) << 16) + | ((long) (digest[1] & 0xFF) << 8) + | (digest[0] & 0xFF); + return hashCode & 0xffffffffL; + } + + public static long md5Hash(String value, int number) { + MessageDigest md5; + try { + md5 = MessageDigest.getInstance("MD5"); + } catch (NoSuchAlgorithmException e) { + throw new RuntimeException("MD5 not supported", e); + } + md5.reset(); + byte[] keyBytes = value.getBytes(StandardCharsets.UTF_8); + md5.update(keyBytes); + byte[] digest = md5.digest(); + // 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; } - // - /**///// Pearson's Hash - // char pearson(char[]key, ub4 len, char tab[256]) - // { - // char hash; - // ub4 i; - // for (hash=len, i=0; i> 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; - 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)); + 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)); 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 - *

- * 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; - } - } diff --git a/im-core/pom.xml b/im-core/pom.xml index e354d05..5529817 100644 --- a/im-core/pom.xml +++ b/im-core/pom.xml @@ -31,6 +31,11 @@ spring-context provided + + org.apache.dubbo + dubbo + provided + \ No newline at end of file diff --git a/im-core/src/main/java/net/sopod/soim/core/handler/ProtoMessageHandler.java b/im-core/src/main/java/net/sopod/soim/core/handler/ProtoMessageHandler.java deleted file mode 100644 index 46e62e9..0000000 --- a/im-core/src/main/java/net/sopod/soim/core/handler/ProtoMessageHandler.java +++ /dev/null @@ -1,26 +0,0 @@ -package net.sopod.soim.core.handler; - -import com.google.protobuf.MessageLite; -import net.sopod.soim.core.session.NetUser; - -/** - * ProtoMessageHandler - * - * - * @author tmy - * @date 2022-04-10 19:19 - */ -public abstract class ProtoMessageHandler { - - public final void exec(NetUser netUser, T msg) { - MessageLite res = handle(msg); - if (res != null) { - netUser.write(res); - } - } - - public abstract Class type(); - - public abstract MessageLite handle(T msg); - -} diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java b/im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java new file mode 100644 index 0000000..95f98ed --- /dev/null +++ b/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); + } + +} diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/MessageHandlerContext.java b/im-entry/src/main/java/net/sopod/soim/entry/config/MessageHandlerContext.java new file mode 100644 index 0000000..d8335b5 --- /dev/null +++ b/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> HOLDER = new FastThreadLocal<>(); + + public static void setAttribute(String key, String value) { + Map 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 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 attrMap; + if ((attrMap = HOLDER.get()) != null) { + attrMap.remove(key); + } + } + +} diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/ContextHolder.java b/im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextHolder.java similarity index 80% rename from im-entry/src/main/java/net/sopod/soim/entry/config/ContextHolder.java rename to im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextHolder.java index f9afe52..dd58cca 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/ContextHolder.java +++ b/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 getBean(Class beanType) { diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextInitialed.java b/im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextInitialed.java similarity index 83% rename from im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextInitialed.java rename to im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextInitialed.java index a5c8b7e..6d3968d 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextInitialed.java +++ b/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); diff --git a/im-core/src/main/java/net/sopod/soim/core/handler/AccountMessageHandler.java b/im-entry/src/main/java/net/sopod/soim/entry/handler/AccountMessageHandler.java similarity index 81% rename from im-core/src/main/java/net/sopod/soim/core/handler/AccountMessageHandler.java rename to im-entry/src/main/java/net/sopod/soim/entry/handler/AccountMessageHandler.java index 8b94430..8288855 100644 --- a/im-core/src/main/java/net/sopod/soim/core/handler/AccountMessageHandler.java +++ b/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 diff --git a/im-core/src/main/java/net/sopod/soim/core/handler/NetUserMessageHandler.java b/im-entry/src/main/java/net/sopod/soim/entry/handler/NetUserMessageHandler.java similarity index 84% rename from im-core/src/main/java/net/sopod/soim/core/handler/NetUserMessageHandler.java rename to im-entry/src/main/java/net/sopod/soim/entry/handler/NetUserMessageHandler.java index ce216ed..5915efe 100644 --- a/im-core/src/main/java/net/sopod/soim/core/handler/NetUserMessageHandler.java +++ b/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; /** diff --git a/im-entry/src/main/java/net/sopod/soim/entry/handler/auth/ReqTokenAuthHandler.java b/im-entry/src/main/java/net/sopod/soim/entry/handler/auth/ReqTokenAuthHandler.java index be44405..dd6efca 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/handler/auth/ReqTokenAuthHandler.java +++ b/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 { + private static final Logger logger = LoggerFactory.getLogger(ReqOnlineUserListHandler.class); + @DubboReference private UserService userService; @Override public MessageLite handle(Account account, UserGroup.ReqOnlineUserList msg) { List userInfos = userService.onlineUserList(msg.getKeyword()); + List resUserInfos = userInfos.stream().map(user -> UserGroup.UserInfo.newBuilder() .setUid(user.getUid()) .setAccount(user.getAccount()) diff --git a/im-entry/src/main/java/net/sopod/soim/entry/server/ProtoMessageDispatcher.java b/im-entry/src/main/java/net/sopod/soim/entry/server/ProtoMessageDispatcher.java index 8a8c3c0..c37db98 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/server/ProtoMessageDispatcher.java +++ b/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 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(); + } + }); } } diff --git a/im-entry/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter b/im-entry/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter new file mode 100644 index 0000000..e4e982d --- /dev/null +++ b/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 \ No newline at end of file diff --git a/im-entry/src/main/resources/application.yml b/im-entry/src/main/resources/application.yml index 18e6075..c755377 100644 --- a/im-entry/src/main/resources/application.yml +++ b/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 \ No newline at end of file + nacos-addr: 124.222.131.236:3848 diff --git a/im-service-api/im-router-api/pom.xml b/im-service-api/im-router-api/pom.xml index 117f248..e42f208 100644 --- a/im-service-api/im-router-api/pom.xml +++ b/im-service-api/im-router-api/pom.xml @@ -17,6 +17,11 @@ net.sopod 1.0.0 + + im-common + net.sopod + 1.0.0 + org.apache.dubbo dubbo-cluster diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashRoute.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashRoute.java deleted file mode 100644 index b8b455d..0000000 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashRoute.java +++ /dev/null @@ -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 Invoker doSelect(List> invokers, URL url, Invocation invocation) { - Invoker invoker = invokers.get(0); - return null; - } - -} diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashTest.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ConsistentHashTest.java new file mode 100644 index 0000000..58c093e --- /dev/null +++ b/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 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 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 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 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(); + } + + +} diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java new file mode 100644 index 0000000..a53d592 --- /dev/null +++ b/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 Invoker doSelect(List> 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)selector).select(invocation); + } + + private int getInvokersHash(List> invokers) { + return invokers.hashCode(); + } + + static class ConsistentHashSelector { + + private final TreeMap> virtualInvokers; + + private final int identityHashCode; + + ConsistentHashSelector(List> invokers, int identityHashCode) { + this.identityHashCode = identityHashCode; + this.virtualInvokers = new TreeMap<>(); + + for (Invoker 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 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> entry = virtualInvokers.ceilingEntry(hash); + if (entry == null) { + entry = virtualInvokers.firstEntry(); + } + return entry.getValue(); + } + + } + +} diff --git a/im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance b/im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance new file mode 100644 index 0000000..4502309 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java index 6ebfe83..85ac866 100644 --- a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java +++ b/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(); diff --git a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserServiceImpl.java b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserServiceImpl.java index 89ce99d..9ee8319 100644 --- a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserServiceImpl.java +++ b/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 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); } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java b/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java new file mode 100644 index 0000000..85e0f21 --- /dev/null +++ b/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) { + + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterServiceRegistry.java b/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterServiceRegistry.java new file mode 100644 index 0000000..d849ede --- /dev/null +++ b/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]; + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserEntryRegistryServiceImpl.java b/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserEntryRegistryServiceImpl.java index 2de1982..9748d32 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/service/UserEntryRegistryServiceImpl.java +++ b/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 onlineUserList(String keyword) { + logger.info("client context uid: {}", ServerContext.getContextUid()); + Stream stream = uidImEntryStore.values().stream(); if (!StringUtil.isEmpty(keyword)) { // 根据关键词过滤 diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/util/ServerContext.java b/im-service/im-router/src/main/java/net/sopod/soim/router/util/ServerContext.java new file mode 100644 index 0000000..9aa4691 --- /dev/null +++ b/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); + } + +} diff --git a/im-service/im-router/src/main/resources/application.yml b/im-service/im-router/src/main/resources/application.yml index 3e86641..c8bc2f0 100644 --- a/im-service/im-router/src/main/resources/application.yml +++ b/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 \ No newline at end of file + loadbalance: im_route_consistent_hash