im-router, im-entry 负载均衡

README.md

@@ -2,8 +2,9 @@ TODO
 - router -> entry 负载均衡
 - router 新增节点顺时针相邻节点数据一致性哈希迁移
 - router 冗余节点存储数据不提供服务
-- dubbo 服务异步处理吞吐量
+- dubbo 服务异步处理提升吞吐量
 - das 消息队列异步写
+- entry 监控，http 查询 entry 地址返回接口
 - 功能开发：
     - 消息群发(im-router 群消息路由,批量uid一致性哈希路由)
     - 好友列表(在线状态:批量uid一致性hash, router查询)

im-client/src/main/java/net/sopod/soim/client/cmd/CmdStarter.java

@@ -31,8 +31,11 @@ public class CmdStarter {
         Logger.pre("【client】: ");
         while(this.scanner.hasNextLine()) {
             String cmd = this.scanner.nextLine();
+            try {
                 this.cmdDispatcher.dispatchCmd(cmd);
-
+            } catch (Exception e) {
+                Logger.error("cmd error: ", e.getMessage());
+            }
             // 退出
             if (CmdEnum.exit.name().equals(cmd.split("[ \t]+")[0])) {
                 Logger.info("bye bye");

im-client/src/main/java/net/sopod/soim/client/config/ClientConfig.java

@@ -17,6 +17,6 @@ public class ClientConfig {
 
     private String host = "127.0.0.1";
 
-    private Integer port = 8088;
+    private Integer port = 8087;
 
 }

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

@@ -15,6 +15,8 @@ public interface AppConstant {
 
     String APP_IM_ENTRY_NAME = "im-entry";
 
+    String APP_IM_ROUTER_NAME = "im-router";
+
     String APP_IM_HTTP_ENTRY_NAME = "im-http-entry";
     String APP_IM_DAS_USER_NAME = "im-das-user";
 

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

@@ -17,6 +17,10 @@ public interface DubboConstant {
      * 请求 im-entry 服务地址
      */
     String IM_ENTRY_ADDR = "entry_addr";
+    /**
+     * 请求 im-router 服务id
+     */
+    String IM_ROUTER_ID_KEY = "im_router_id";
 
     /**
      * im-entry 服务接口前缀

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

@@ -1,5 +1,6 @@
 package net.sopod.soim.common.util;
 
+import javax.annotation.Nullable;
 import java.util.*;
 import java.util.function.Function;
 
@@ -11,16 +12,50 @@ import java.util.function.Function;
  */
 public class Collects {
 
-    /**
+	public static boolean isEmpty(@Nullable Collection<?> collection) {
-     * 数组翻转
+		return collection == null || collection.isEmpty();
-     */
-    public static long[] revers(long[] arr) {
-        for (int i = 0, j = arr.length - 1; i < j; i++, j--) {
-            long temp = arr[i];
-            arr[i] = arr[j];
-            arr[j] = temp;
 	}
-        return arr;
+
+	public static boolean isNotEmpty(@Nullable Collection<?> collection) {
+		return !isEmpty(collection);
+	}
+
+	public static boolean isEmpty(@Nullable Object[] arr) {
+		return arr == null || arr.length > 0;
+	}
+
+	public static boolean isNotEmpty(@Nullable Object[] arr) {
+		return !isEmpty(arr);
+	}
+
+	public static <T, K> Map<K, T> collect2Map(Collection<T> collect,
+											   Function<T, K> keyProvider) {
+		return collect2Map(collect, keyProvider, new HashMap<>());
+	}
+
+	public static <T, K> Map<K, T> collect2Map(Collection<T> collect,
+													Function<T, K> keyProvider,
+													Map<K, T> resultMap) {
+		for (T item : collect) {
+			resultMap.put(keyProvider.apply(item), item);
+		}
+		return resultMap;
+	}
+
+	public static <T, K, V> Map<K, V> collect2KvMap(Collection<T> collect,
+													Function<T, K> keyProvider,
+													Function<T, V> valueProvider) {
+		return collect2KvMap(collect, keyProvider, valueProvider, new HashMap<>());
+	}
+
+	public static <T, K, V> Map<K, V> collect2KvMap(Collection<T> collect,
+												  Function<T, K> keyProvider,
+												  Function<T, V> valueProvider,
+												  Map<K, V> resultMap) {
+		for (T item : collect) {
+			resultMap.put(keyProvider.apply(item), valueProvider.apply(item));
+		}
+		return resultMap;
 	}
 
 	public static <T,K,V> Map<K, List<V>> group(Collection<T> collect,
@@ -60,4 +95,25 @@ public class Collects {
 		return result;
 	}
 
+	/**
+	 * 数组翻转
+	 */
+	public static long[] revers(long[] arr) {
+		for (int i = 0, j = arr.length - 1; i < j; i++, j--) {
+			long temp = arr[i];
+			arr[i] = arr[j];
+			arr[j] = temp;
+		}
+		return arr;
+	}
+
+	/**
+	 * 根据元素个数计算 map 容量大小
+	 * @param size 元素个数
+	 * @return map 容量大小
+	 */
+	public static int mapCapacity(int size) {
+		return Math.max(2, (int)Math.ceil(size / 0.75));
+	}
+
 }

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

@@ -20,4 +20,8 @@ public class StringUtil {
         return content == null || content.length() == 0;
     }
 
+    public static String toString(Object data) {
+        return data == null ? null : data.toString();
+    }
+
 }

im-core/src/main/java/net/sopod/soim/core/session/Account.java

@@ -14,10 +14,13 @@ public class Account extends NetUser {
 
     private final String name;
 
-    Account(NetUser netUser, long uid, String name) {
+    private final String imRouterId;
+
+    Account(NetUser netUser, long uid, String name, String imRouterId) {
         super(netUser.channel.get());
         this.uid = uid;
         this.name = name;
+        this.imRouterId = imRouterId;
     }
 
     @Override
@@ -38,9 +41,14 @@ public class Account extends NetUser {
         return name;
     }
 
+    public String getImRouterId() {
+        return imRouterId;
+    }
+
     public static class AccountBuilder {
         private long uid;
         private String name;
+        private String imRouterId;
         private NetUser netUser;
         public static AccountBuilder newBuilder() {
             return new AccountBuilder();
@@ -57,10 +65,14 @@ public class Account extends NetUser {
             this.name = name;
             return this;
         }
+        public AccountBuilder setImRouterId(String imRouterId) {
+            this.imRouterId = imRouterId;
+            return this;
+        }
         public Account build() {
             Preconditions.checkNotNull(netUser);
             Preconditions.checkNotNull(name);
-            return new Account(netUser, uid, name);
+            return new Account(netUser, uid, name, imRouterId);
         }
     }
 

im-entry/src/main/java/net/sopod/soim/entry/EntryApplication.java

@@ -1,11 +1,11 @@
 package net.sopod.soim.entry;
 
 import com.alibaba.nacos.api.exception.NacosException;
-import net.sopod.soim.entry.registry.RegistryService;
 import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.ConfigurableApplicationContext;
 
 /**
  * EntryMain
@@ -17,12 +17,8 @@ import org.springframework.context.ConfigurableApplicationContext;
 @EnableDubbo(scanBasePackages = {"net.sopod.soim.entry.service"})
 public class EntryApplication {
 
-    public static void main(String[] args) throws NacosException {
+    public static void main(String[] args) {
         SpringApplication.run(EntryApplication.class, args);
-
-        // 注册 im-entry 服务
-//        RegistryService registryService = context.getBean(RegistryService.class);
-//        registryService.registryImEntry();
     }
 
 }

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

@@ -1,6 +1,7 @@
 package net.sopod.soim.entry.config;
 
 import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.logic.common.util.RpcContextUtil;
 import org.apache.dubbo.rpc.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -20,7 +21,8 @@ public class DubboPreInvokeFilter implements Filter {
     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);
+            // 设置调用上下文 uid
+            RpcContextUtil.setContextUid(uid);
             if (logger.isDebugEnabled()) {
                 logger.debug("pre invoke filter set uid: {}", uid);
             }

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

@@ -4,10 +4,9 @@ import org.apache.dubbo.common.URL;
 import org.apache.dubbo.rpc.Exporter;
 import org.apache.dubbo.rpc.ExporterListener;
 import org.apache.dubbo.rpc.RpcException;
-import org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol;
+import org.apache.dubbo.rpc.protocol.injvm.InjvmProtocol;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.context.ApplicationContext;
 
 /**
  * DubboExporterListener
@@ -16,24 +15,24 @@ import org.springframework.context.ApplicationContext;
  * @author tmy
  * @date 2022-05-02 09:55
  */
-public class DubboImEntryExporterListener implements ExporterListener {
+public class ImEntryAPIExporterListener implements ExporterListener {
 
-    private static final Logger logger = LoggerFactory.getLogger(DubboImEntryExporterListener.class);
+    private static final Logger logger = LoggerFactory.getLogger(ImEntryAPIExporterListener.class);
 
     @Override
     public void exported(Exporter<?> exporter) throws RpcException {
         URL invokerUrl = exporter.getInvoker().getUrl();
-        if (DubboProtocol.NAME.equals(invokerUrl.getProtocol())) {
+        if (!InjvmProtocol.NAME.equals(invokerUrl.getProtocol())) {
-            if (ApplicationContextHolder.getDubboAppServiceAddr() == null) {
+            if (ImEntryAppContextHolder.getDubboAppServiceAddr() == null) {
-                logger.info("im-entry dubbo app serverAddr: {}", invokerUrl.getAddress());
+                ImEntryAppContextHolder.setDubboAppServiceAddr(invokerUrl.getAddress());
-                ApplicationContextHolder.setDubboAppServiceAddr(invokerUrl.getAddress());
+                logger.info("im-entry registry serverAddr: {}", invokerUrl.getAddress());
             }
         }
     }
 
     @Override
     public void unexported(Exporter<?> exporter) {
-
+        logger.info("unexported listener: {}", exporter.getInvoker().getInterface());
     }
 
 }

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

@@ -8,14 +8,14 @@ import org.springframework.context.ApplicationContext;
  * @author tmy
  * @date 2022-04-28 15:06
  */
-public class ApplicationContextHolder {
+public class ImEntryAppContextHolder {
 
     private static ApplicationContext applicationContext;
 
     private static String dubboAppServiceAddr;
 
     public static void setContext(ApplicationContext applicationContext) {
-        ApplicationContextHolder.applicationContext = applicationContext;
+        ImEntryAppContextHolder.applicationContext = applicationContext;
     }
 
     public static <T> T getBean(Class<T> beanType) {
@@ -23,7 +23,7 @@ public class ApplicationContextHolder {
     }
 
     public static void setDubboAppServiceAddr(String dubboAppServiceAddr) {
-        ApplicationContextHolder.dubboAppServiceAddr = dubboAppServiceAddr;
+        ImEntryAppContextHolder.dubboAppServiceAddr = dubboAppServiceAddr;
     }
 
     public static String getDubboAppServiceAddr() {

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

@@ -17,7 +17,7 @@ public class SpringApplicationContextInitialed implements ApplicationContextAwar
 
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        ApplicationContextHolder.setContext(applicationContext);
+        ImEntryAppContextHolder.setContext(applicationContext);
 
         // 注册 protobuf 消息 handler
         ProtoMessageHandlerRegistry.registerHandlerWithApplicationContext(applicationContext);

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.entry.config.ApplicationContextHolder;
+import net.sopod.soim.entry.config.ImEntryAppContextHolder;
 import net.sopod.soim.entry.handler.NetUserMessageHandler;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.core.session.NetUser;
@@ -47,10 +47,10 @@ public class ReqTokenAuthHandler extends NetUserMessageHandler<Auth.ReqTokenAuth
         logger.info("ReqTokenAuth: uid={}", msg.getUid());
         // 校验 token，
         // TODO 更新 router 状态成功后再升级为 account，主动关闭时发送一个消息到客户端，校验错误/超时
-        Boolean isValid = userAuthService.validateToken(msg.getToken(),
+        String imRouterId = userAuthService.validateToken(msg.getToken(),
-                ApplicationContextHolder.getDubboAppServiceAddr());
+                ImEntryAppContextHolder.getDubboAppServiceAddr());
 
-        if (!Boolean.TRUE.equals(isValid)) {
+        if (imRouterId == null) {
             return Auth.ResTokenAuth.newBuilder()
                     .setSuccess(false)
                     .setMessage("token校验失败,请重新登录")
@@ -61,6 +61,7 @@ public class ReqTokenAuthHandler extends NetUserMessageHandler<Auth.ReqTokenAuth
                 .setNetUser(netUser)
                 .setUid(msg.getUid())
                 .setName("") // TODO 账户名处理
+                .setImRouterId(imRouterId)
                 .build();
         netUser.upgradeAccount(account);
 

im-entry/src/main/java/net/sopod/soim/entry/service/TextChatServiceImpl.java

@@ -3,7 +3,7 @@ package net.sopod.soim.entry.service;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.data.msg.chat.Chat;
 import net.sopod.soim.entry.api.service.TextChatService;
-import net.sopod.soim.entry.config.ApplicationContextHolder;
+import net.sopod.soim.entry.config.ImEntryAppContextHolder;
 import net.sopod.soim.entry.server.AccountRegistry;
 import net.sopod.soim.logic.common.model.TextChat;
 import org.apache.dubbo.config.annotation.DubboService;
@@ -30,7 +30,7 @@ public class TextChatServiceImpl implements TextChatService {
     public Boolean sendTextChat(TextChat chat) {
         Long receiverUid = chat.getReceiverUid();
         Account account = accountRegistry.get(receiverUid);
-        logger.info("uid: {}, account: {}, {}", receiverUid, account, ApplicationContextHolder.getDubboAppServiceAddr());
+        logger.info("uid: {}, account: {}, {}", receiverUid, account, ImEntryAppContextHolder.getDubboAppServiceAddr());
         if (account == null) {
             return Boolean.FALSE;
         }

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

@@ -1 +1 @@
-im_entry_export_listener=net.sopod.soim.entry.config.DubboImEntryExporterListener
+im_entry_export_listener=net.sopod.soim.entry.config.ImEntryAPIExporterListener

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

@@ -22,6 +22,7 @@ dubbo:
     loadbalance: im_entry_loadbalance
     timeout: 2000
     retries: 0
+    register: true
 
 entry-server:
   nacos-addr: 124.222.131.236:3848

im-service-api/im-entry-api/src/main/java/net/sopod/soim/entry/api/route/ImEntryServerAddressLoadBalance.java

@@ -1,45 +0,0 @@
-package net.sopod.soim.entry.api.route;
-
-import net.sopod.soim.common.constant.DubboConstant;
-import net.sopod.soim.common.util.StringUtil;
-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;
-
-/**
- * ImEntryServerAddressLoadBalance
- *
- * @author tmy
- * @date 2022-05-02 14:45
- */
-public class ImEntryServerAddressLoadBalance extends AbstractLoadBalance {
-
-    private static final Logger logger = LoggerFactory.getLogger(ImEntryServerAddressLoadBalance.class);
-
-    @Override
-    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
-        String invokeAddr = "";
-        for (Invoker<T> invoker : invokers) {
-            invokeAddr += invoker.getUrl().getAddress() + ":" + invoker.getInterface() + ",";
-        }
-        logger.info("invokers: {}", invokeAddr);
-        String entryAddr = invocation.getAttachment(DubboConstant.IM_ENTRY_ADDR);
-        if (StringUtil.isEmpty(entryAddr)) {
-            throw new IllegalCallerException("上下文 im-entry 服务地址不能为空");
-        }
-        logger.info("entryAddr: {}", entryAddr);
-        for (Invoker<T> invoker : invokers) {
-            if (entryAddr.equals(invoker.getUrl().getAddress())) {
-                logger.info("invoker address: {}", invoker.getUrl().getAddress());
-                return invoker;
-            }
-        }
-        throw new IllegalCallerException("没有地址为 " + entryAddr + " im-entry 服务");
-    }
-
-}

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

@@ -1 +0,0 @@
-im_entry_loadbalance=net.sopod.soim.entry.api.route.ImEntryServerAddressLoadBalance

im-service-api/im-logic-common/src/main/java/net/sopod/soim/logic/common/util/RpcContextUtil.java

@@ -0,0 +1,32 @@
+package net.sopod.soim.logic.common.util;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.common.util.StringUtil;
+import org.apache.dubbo.rpc.RpcContext;
+
+/**
+ * RpcContextUtil
+ *
+ * @author tmy
+ * @date 2022-05-04 22:44
+ */
+public class RpcContextUtil {
+
+    public static void setContextUid(Long uid) {
+        setContextUid(StringUtil.toString(uid));
+    }
+
+    public static void setContextUid(String uid) {
+        RpcContext.getServiceContext().setAttachment(DubboConstant.CTX_UID, uid);
+    }
+
+    public static String getContextUid() {
+        return RpcContext.getServiceContext().getAttachment(DubboConstant.CTX_UID);
+    }
+
+    public static Long getContextUidNum() {
+        String uid = getContextUid();
+        return uid == null ? null : Long.valueOf(uid);
+    }
+
+}

im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/auth/service/UserAuthService.java

@@ -18,6 +18,6 @@ public interface UserAuthService {
      * @param imEntryAddr entry 节点地址，检验通过后存储到 im-router
      * @return 是否校验成功
      */
-    Boolean validateToken(String token, String imEntryAddr);
+    String validateToken(String token, String imEntryAddr);
 
 }

im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RegistryRes.java

@@ -0,0 +1,24 @@
+package net.sopod.soim.router.api.model;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+import java.io.Serializable;
+
+/**
+ * RegistryRes
+ *
+ * @author tmy
+ * @date 2022-05-04 16:05
+ */
+@Data
+@Accessors(chain = true)
+public class RegistryRes implements Serializable {
+
+    private static final long serialVersionUID = -4367288919640496421L;
+
+    private Boolean success;
+
+    private String imRouterId;
+
+}

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

@@ -1,7 +1,7 @@
 package net.sopod.soim.router.api.route;
 
 import net.sopod.soim.common.constant.DubboConstant;
-import net.sopod.soim.common.util.HashAlgorithms;
+import net.sopod.soim.common.util.Collects;
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
@@ -9,9 +9,9 @@ import org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.TreeMap;
 
 /**
  * ConsistentHashRoute
@@ -26,12 +26,7 @@ public class ImRouterConsistentHashRoute extends AbstractLoadBalance {
 
     private static final Logger logger = LoggerFactory.getLogger(ImRouterConsistentHashRoute.class);
 
-    /**
+    private volatile UidConsistentHashSelector<?> selector;
-     * 一致性 hash 每个节点的虚拟节点数量
-     */
-    private static final int VIRTUAL_NODE_SIZE = 120;
-
-    private volatile ConsistentHashSelector<?> selector;
 
     /**
      * 后续如有接口版本号，构建 map 每个方法版本，对应一个 Selector
@@ -40,56 +35,22 @@ public class ImRouterConsistentHashRoute extends AbstractLoadBalance {
     @SuppressWarnings("unchecked")
     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
         int invokersHash = getInvokersHash(invokers);
-        logger.info("invokers hash: {}", invokersHash);
+        // 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) {
+        if (selector == null || selector.getIdentityHashCode() != invokersHash) {
-        return invokers.hashCode();
+            // 调用节点有变化，构建新的 hash 表
+            Map<String, Invoker<T>> serverAddrInvokerMap = Collects.collect2Map(invokers,
+                    invoker -> invoker.getUrl().getAddress(),
+                    new HashMap<>(6));
+            selector = new UidConsistentHashSelector<>(serverAddrInvokerMap, invokersHash);
         }
-
-    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) {
+        return (Invoker<T>) selector.select(uid);
-                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();
     }
 
+    private <T> int getInvokersHash(List<Invoker<T>> invokers) {
+        return invokers.hashCode();
     }
 
 }

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

@@ -0,0 +1,44 @@
+package net.sopod.soim.router.api.route;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.common.util.StringUtil;
+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;
+
+/**
+ * ImRouterDirectLoadBalance
+ * 通过上下文指定的 router 服务地址调用 im-router 服务接口
+ *
+ * @author tmy
+ * @date 2022-05-04 15:26
+ */
+public class ImRouterDirectLoadBalance extends AbstractLoadBalance {
+
+    public static final String NAME = "im_router_direct";
+
+    private static final Logger logger = LoggerFactory.getLogger(ImRouterDirectLoadBalance.class);
+
+    @Override
+    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        String imRouterId = invocation.getAttachment(DubboConstant.IM_ROUTER_ID_KEY);
+        logger.info("invocation router id: {}", imRouterId);
+        if (StringUtil.isEmpty(imRouterId)) {
+            throw new IllegalCallerException("上下文im-router服务id不能为空");
+        }
+        for (Invoker<T> invoker : invokers) {
+            String invokerId = invoker.getUrl().getParameter(DubboConstant.IM_ROUTER_ID_KEY);
+            logger.info("invoker router id: {}", invokerId);
+            if (imRouterId.equals(invokerId)) {
+                return invoker;
+            }
+        }
+        throw new IllegalCallerException("没有id为" + imRouterId + "的im-router服务");
+    }
+
+}

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

@@ -0,0 +1,62 @@
+package net.sopod.soim.router.api.route;
+
+import net.sopod.soim.common.util.HashAlgorithms;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * UidConsistentHashSelector
+ *
+ * @author tmy
+ * @date 2022-05-04 10:37
+ */
+public class UidConsistentHashSelector<V> {
+
+    /**
+     * 一致性 hash 每个节点的虚拟节点数量
+     */
+    private static final int VIRTUAL_NODE_SIZE = 120;
+
+    private final TreeMap<Long, V> virtualNodeMap;
+
+    private final int identityHashCode;
+
+    public UidConsistentHashSelector(Map<String, V> serverAddressMap, int identityHashCode) {
+        this.identityHashCode = identityHashCode;
+        this.virtualNodeMap = new TreeMap<>();
+
+        // 构建虚拟节点一致性 hash 表
+        for (Map.Entry<String, V> entry : serverAddressMap.entrySet()) {
+            String serverAddr = entry.getKey();
+            V value = entry.getValue();
+            for (int i = 0, len = VIRTUAL_NODE_SIZE / 4; i < len; i++) {
+                for (int h = 0; h < 4; h++) {
+                    long hash = hash(serverAddr + i, h);
+                    this.virtualNodeMap.put(hash, value);
+                }
+            }
+        }
+    }
+
+    public V select(String 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, V> entry = virtualNodeMap.ceilingEntry(hash);
+        if (entry == null) {
+            entry = virtualNodeMap.firstEntry();
+        }
+        return entry.getValue();
+    }
+
+    private static long hash(String value, int number) {
+        return HashAlgorithms.md5Hash(value, number);
+    }
+
+    public int getIdentityHashCode() {
+        return identityHashCode;
+    }
+
+}

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

@@ -3,7 +3,7 @@ package net.sopod.soim.router.api.service;
 import net.sopod.soim.router.api.model.CacheRes;
 
 /**
- * RouterService
+ * AuthStoreService
  *
  * @author tmy
  * @date 2022-04-14 11:22

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

@@ -3,19 +3,20 @@ package net.sopod.soim.router.api.service;
 import net.sopod.soim.logic.common.model.TextChat;
 import net.sopod.soim.router.api.model.CacheRes;
 import net.sopod.soim.logic.common.model.UserInfo;
+import net.sopod.soim.router.api.model.RegistryRes;
 
 import java.util.List;
 
 /**
- * AccountStoreService
+ * UserRouteService
  * 缓存登录账号信息，entry 节点信息...
  *
  * @author tmy
  * @date 2022-04-14 15:22
  */
-public interface UserEntryRegistryService {
+public interface UserRouteService {
 
-    CacheRes registryUserEntry(Long uid, String imEntryAddr);
+    RegistryRes registryUserEntry(Long uid, String imEntryAddr);
 
     /**
      * 查询在线用户列表

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

@@ -1 +1,2 @@
 im_route_consistent_hash=net.sopod.soim.router.api.route.ImRouterConsistentHashRoute
+im_router_direct=net.sopod.soim.router.api.route.ImRouterDirectLoadBalance

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

@@ -1,13 +1,12 @@
 package net.sopod.soim.logic.user.service;
 
-import net.sopod.soim.common.constant.DubboConstant;
 import net.sopod.soim.das.user.api.model.entity.ImUser;
 import net.sopod.soim.das.user.api.service.UserDasService;
 import net.sopod.soim.logic.common.model.TextChat;
-import net.sopod.soim.router.api.service.UserEntryRegistryService;
+import net.sopod.soim.logic.common.util.RpcContextUtil;
+import net.sopod.soim.router.api.service.UserRouteService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
-import org.apache.dubbo.rpc.RpcContext;
 
 import java.util.Objects;
 
@@ -21,7 +20,7 @@ import java.util.Objects;
 public class ChatServiceImpl implements ChatService {
 
     @DubboReference
-    private UserEntryRegistryService userEntryRegistryService;
+    private UserRouteService userRouteService;
 
     @DubboReference
     private UserDasService userDasService;
@@ -36,8 +35,9 @@ public class ChatServiceImpl implements ChatService {
             }
             textChat.setReceiverUid(receiverUser.getId());
         }
-        RpcContext.getServiceContext().setAttachment(DubboConstant.CTX_UID, String.valueOf(textChat.getReceiverUid()));
+        // 设置调用 router 为消息接受者地址
-        return userEntryRegistryService.routeTextChat(textChat);
+        RpcContextUtil.setContextUid(textChat.getReceiverUid());
+        return userRouteService.routeTextChat(textChat);
     }
 
 }

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

@@ -8,8 +8,8 @@ import net.sopod.soim.das.user.api.service.UserDasService;
 import net.sopod.soim.logic.user.auth.model.ImAuth;
 import net.sopod.soim.logic.user.auth.service.UserAuthService;
 import net.sopod.soim.logic.user.config.AuthConfig;
-import net.sopod.soim.router.api.model.CacheRes;
+import net.sopod.soim.router.api.model.RegistryRes;
-import net.sopod.soim.router.api.service.UserEntryRegistryService;
+import net.sopod.soim.router.api.service.UserRouteService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
 import org.apache.dubbo.rpc.RpcContext;
@@ -32,7 +32,7 @@ public class UserAuthServiceImpl implements UserAuthService {
     private UserDasService userDasService;
 
     @DubboReference
-    private UserEntryRegistryService userEntryRegistryService;
+    private UserRouteService userRouteService;
 
     @Resource
     private AuthConfig authConfig;
@@ -61,17 +61,18 @@ public class UserAuthServiceImpl implements UserAuthService {
     }
 
     @Override
-    public Boolean validateToken(String token, String imEntryAddr) {
+    public String validateToken(String token, String imEntryAddr) {
         TokenUtil.Payload payload = TokenUtil.validateAndParse(token);
         if (payload == null) {
-            return Boolean.FALSE;
+            return null;
         }
         // 注册记录用户登录的 entry 节点
         RpcContext.getServiceContext()
                 .setAttachment(DubboConstant.CTX_UID, String.valueOf(payload.getUserId()));
-        CacheRes cacheRes = userEntryRegistryService
+        RegistryRes registryRes = userRouteService
                 .registryUserEntry(payload.getUserId(), imEntryAddr);
-        return cacheRes.getSuccess();
+
+        return !Boolean.TRUE.equals(registryRes.getSuccess()) ? null : registryRes.getImRouterId();
     }
 
 }

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

@@ -1,7 +1,7 @@
 package net.sopod.soim.logic.user.service;
 
 import net.sopod.soim.logic.common.model.UserInfo;
-import net.sopod.soim.router.api.service.UserEntryRegistryService;
+import net.sopod.soim.router.api.service.UserRouteService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
 import org.apache.dubbo.rpc.RpcContext;
@@ -23,7 +23,7 @@ public class UserBizServiceImpl implements UserBizService {
     private static final Logger logger = LoggerFactory.getLogger(UserBizServiceImpl.class);
 
     @DubboReference
-    private UserEntryRegistryService userEntryRegistryService;
+    private UserRouteService userRouteService;
 
     @Override
     public CompletableFuture<String> sayHi(String name) {
@@ -38,7 +38,7 @@ public class UserBizServiceImpl implements UserBizService {
     @Override
     public List<UserInfo> onlineUserList(String keyword) {
         logger.info("service context uid: {}", RpcContext.getServiceContext().getAttachment("uid"));
-        return userEntryRegistryService.onlineUserList(keyword);
+        return userRouteService.onlineUserList(keyword);
     }
 
 }

im-service/im-router/pom.xml

@@ -81,6 +81,10 @@
             <groupId>com.github.ben-manes.caffeine</groupId>
             <artifactId>caffeine</artifactId>
         </dependency>
+        <dependency>
+            <groupId>cglib</groupId>
+            <artifactId>cglib</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -0,0 +1,26 @@
+package net.sopod.soim.router.cache;
+
+/**
+ * BiSync
+ *
+ * @author tmy
+ * @date 2022-05-04 17:22
+ */
+public interface DataSync {
+
+    /** 不是更新数据的方法开头 */
+    String[] nonUpdateMethodStart = new String[]{"get", "select", "list"};
+
+    /**
+     * 如果是更新方法会同步数据，到新增节点或备份节点
+     */
+    default boolean isUpdateMethod(String methodName) {
+        for (String nonUpdateStart : nonUpdateMethodStart) {
+            if (methodName.startsWith(nonUpdateStart)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+}

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

@@ -0,0 +1,66 @@
+package net.sopod.soim.router.cache;
+
+import net.sf.cglib.proxy.Enhancer;
+import net.sf.cglib.proxy.MethodInterceptor;
+import net.sf.cglib.proxy.MethodProxy;
+import net.sopod.soim.common.util.Collects;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+import java.lang.reflect.Method;
+
+/**
+ * BiSyncProxyManager
+ * 设置属性的时候，将设置方法同步处理
+ *
+ * @author tmy
+ * @date 2022-05-04 17:10
+ */
+public class DataSyncProxyFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(DataSyncProxyFactory.class);
+
+    @SuppressWarnings("unchecked")
+    public static <T extends DataSync> T newProxyInstance(Class<T> biSyncClazz) {
+        Enhancer enhancer = new Enhancer();
+        enhancer.setSuperclass(biSyncClazz);
+        enhancer.setCallback(new DataSyncProxyCallback());
+        return (T) enhancer.create();
+    }
+
+    public static class DataSyncProxyCallback implements MethodInterceptor {
+
+        @Override
+        public Object intercept(Object instance, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
+            String methodName = method.getName();
+            // 是判断是否更新的方法跳过
+            if ("isUpdateMethod".equals(methodName)) {
+                return methodProxy.invokeSuper(instance, args);
+            }
+            boolean isUpdateMethod = ((DataSync) instance).isUpdateMethod(methodName);
+            if (!isUpdateMethod) {
+                return methodProxy.invokeSuper(instance, args);
+            }
+            if (Collects.isNotEmpty(args)) {
+                // TODO 参数值可能为 null，检查参数可序列化放在生成代理对象时
+                for (int i = 0; i < args.length; i++) {
+                    if (!(args[i] instanceof Serializable)) {
+                        logger.error("类:{} 更新方法:{} 第{}i个参数不可序列化", instance.getClass(), methodName, i+1);
+                    }
+                }
+            }
+            // TODO 记录更新操作(方法和参数)，查询数据订阅者，异步同步数据
+            System.out.println("intercept invoke:" + methodName);
+            return methodProxy.invokeSuper(instance, args);
+        }
+
+    }
+
+    public static void main(String[] args) {
+        RouterUser routerUser = newProxyInstance(RouterUser.class);
+        routerUser.setAccount("日月光");
+        System.out.println(routerUser.getAccount());
+    }
+
+}

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

@@ -1,4 +1,4 @@
-package net.sopod.soim.router.api.model;
+package net.sopod.soim.router.cache;
 
 import lombok.Data;
 import lombok.experimental.Accessors;
@@ -11,7 +11,7 @@ import lombok.experimental.Accessors;
  */
 @Data
 @Accessors(chain = true)
-public class RouterUser {
+public class RouterUser implements DataSync {
 
     private long uid;
 

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

@@ -1,7 +1,5 @@
 package net.sopod.soim.router.cache;
 
-import net.sopod.soim.router.api.model.RouterUser;
-
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 

im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/DataSyncIgnore.java

@@ -0,0 +1,12 @@
+package net.sopod.soim.router.cache.annotation;
+
+/**
+ * BiSyncIgnore
+ * 忽略方法同步
+ *
+ * @author tmy
+ * @date 2022-05-04 17:39
+ */
+public @interface DataSyncIgnore {
+
+}

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

@@ -0,0 +1,50 @@
+package net.sopod.soim.router.config;
+
+import net.sopod.soim.common.util.HashAlgorithms;
+import net.sopod.soim.common.util.StringUtil;
+import org.apache.dubbo.common.URL;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+/**
+ * ImRouterContextHolder
+ * im-router 应用上下文信息
+ * TODO 启动指定 im-router 服务类型 service/backup
+ * TODO backup 指定备份 im-router 或随机备份未备份 im-router 服务
+ *
+ * @author tmy
+ * @date 2022-05-04 09:21
+ */
+public class ImRouterAppContextHolder {
+
+    /**
+     * 要注册的 provider 服务的 url 列表
+     */
+    private static final List<URL> registryInvokerUrls = new CopyOnWriteArrayList<>();
+
+    private static String appServiceAddr;
+
+    public static final String IM_ROUTER_ID;
+
+    static {
+        IM_ROUTER_ID = String.valueOf(HashAlgorithms.md5Hash(StringUtil.randomUUID()));
+    }
+
+    public static void addRegistryInvokerUrl(URL registryInvokerUrl) {
+        registryInvokerUrls.add(registryInvokerUrl);
+    }
+
+    public static List<URL> getRegistryInvokerUrls() {
+        return registryInvokerUrls;
+    }
+
+    public static void setAppServiceAddr(String appServiceAddr) {
+        ImRouterAppContextHolder.appServiceAddr = appServiceAddr;
+    }
+
+    public static String getAppServiceAddr() {
+        return appServiceAddr;
+    }
+
+}

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

@@ -0,0 +1,126 @@
+package net.sopod.soim.router.config;
+
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.exception.NacosException;
+import com.alibaba.nacos.api.naming.NamingService;
+import com.alibaba.nacos.api.naming.pojo.Instance;
+import net.sopod.soim.common.constant.AppConstant;
+import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.common.util.Collects;
+import net.sopod.soim.router.api.route.UidConsistentHashSelector;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.registry.Registry;
+import org.apache.dubbo.registry.support.RegistryManager;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.apache.dubbo.rpc.proxy.AbstractProxyInvoker;
+import org.apache.dubbo.spring.boot.context.event.AwaitingNonWebApplicationListener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.Ordered;
+
+import java.util.*;
+
+/**
+ * AppcationInitialed
+ * 参考 dubbo {@link AwaitingNonWebApplicationListener}
+ *
+ * @author tmy
+ * @date 2022-05-04 09:51
+ */
+@Configuration
+public class ImRouterAppOnReady implements ApplicationListener<ApplicationReadyEvent>, Ordered {
+
+    private static final Logger logger = LoggerFactory.getLogger(ImRouterAppOnReady.class);
+
+    /**
+     * 同步一致性hash临近节点数据
+     */
+    @Override
+    public void onApplicationEvent(ApplicationReadyEvent event) {
+        // 服务已可用进行注册
+        this.doRegistry();
+    }
+
+    /**
+     * 注册 im-router 的API接口服务
+     */
+    private void doRegistry() {
+        RegistryManager registryManager = ApplicationModel.defaultModel().getBeanFactory()
+                .getBean(RegistryManager.class);
+        Collection<Registry> registries = registryManager.getRegistries();
+        List<URL> registryInvokerUrls = ImRouterAppContextHolder.getRegistryInvokerUrls();
+        if (Collects.isNotEmpty(registries)
+                && Collects.isNotEmpty(registryInvokerUrls)) {
+            for (Registry registry : registries) {
+                for (URL invokerUrl : registryInvokerUrls) {
+                    // 添加 im-router 服务id参数，生成新的 url
+                    URL url = invokerUrl.addParameter(
+                            DubboConstant.IM_ROUTER_ID_KEY,
+                            ImRouterAppContextHolder.IM_ROUTER_ID
+                    );
+                    registry.register(url);
+                }
+            }
+        }
+    }
+
+    @Override
+    public int getOrder() {
+        return Ordered.HIGHEST_PRECEDENCE;
+    }
+
+    /**
+     * 一致性hash，使用了虚拟节点会导致迁移多个数据节点
+     *
+     * 应用启动后，进行新增 im-router 节点逻辑处理
+     * 0.获取分布式全局锁，成功开始同步数据(TODO nacos不支持，想其他办法)
+     * 1.获取现有所有 im-router 节点，加上当前节点，生成新的一致性hash虚拟节点表
+     * 2.计算所有需要迁移数据的节点，当前虚拟节点的顺时针临近节点
+     * 3.依次调用数据迁移节点，分页拉取router缓存用户数据（如有数据节点获取失败，探测是否可用，不可用从第1步重新开始；）
+     *   如已拉取数据有更新需要双写？？[CGLib], 服务注册前需要被调用，开netty http服务与dubbo服务端口偏移量1000
+     *
+     * 4.数据迁移完成后，注册dubbo服务，依次调用所有迁移数据节点可清空用户数据（用户数据已移过来了失败不用管）
+     */
+    private void nameServerTestCode() {
+        String serverAddr = "124.222.131.236:3848";
+        Properties properties = new Properties();
+        properties.put("serverAddr", serverAddr);
+
+        // 同步一致性 hash 临近节点数据
+        try {
+            // 获取当前服务实例
+            NamingService namingService = NacosFactory.createNamingService(properties);
+            List<Instance> imRouterInstances = namingService.getAllInstances(AppConstant.APP_IM_ROUTER_NAME);
+
+            //namingService.registerInstance();
+            Instance instance1 = new Instance();
+            instance1.addMetadata("LOCK_VAL", "123123");
+            // instance1.getMetadata();
+
+            logger.info("instances: {}", imRouterInstances);
+            if (!Collects.isEmpty(imRouterInstances)) {
+                Map<String, String> consistentHashNodeMap = new HashMap<>();
+                Map<String, Instance> stringInstanceMap = Collects.collect2Map(imRouterInstances,
+                        Instance::toInetAddr,
+                        new HashMap<>(Collects.mapCapacity(imRouterInstances.size()))
+                );
+                Map<String, String> map = new HashMap<>(Collects.mapCapacity(imRouterInstances.size()));
+                for (Instance instance : imRouterInstances) {
+                    String serverInetAddr = instance.toInetAddr();// 服务地址, 如: 192.168.56.1:3031
+                    map.put(serverInetAddr, serverInetAddr);
+                }
+                UidConsistentHashSelector<String> selector = new UidConsistentHashSelector<>(map, imRouterInstances.hashCode());
+
+                logger.info("instances addr: {}", imRouterInstances.get(0).toInetAddr());
+                logger.info("instances addr: {}:{}", imRouterInstances.get(0).getIp(), imRouterInstances.get(0).getPort());
+            }
+            logger.info("application ready event...");
+        } catch (NacosException e) {
+            e.printStackTrace();
+        }
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/config/filter/InvokeImEntryFilter.java

@@ -1,7 +1,7 @@
-package net.sopod.soim.router.config;
+package net.sopod.soim.router.config.filter;
 
 import net.sopod.soim.common.constant.DubboConstant;
-import net.sopod.soim.router.api.model.RouterUser;
+import net.sopod.soim.router.cache.RouterUser;
 import net.sopod.soim.router.cache.SoImUserCache;
 import org.apache.dubbo.rpc.*;
 import org.slf4j.Logger;
@@ -10,10 +10,12 @@ import org.slf4j.LoggerFactory;
 /**
  * InvokeImEntryFilter
  * 调用 im-entry 服务 serverAddr 上下文设置
+ * 弃用: loadbalance 先于 filter 调用
  *
  * @author tmy
  * @date 2022-05-02 14:56
  */
+@Deprecated
 public class InvokeImEntryFilter implements Filter {
 
     private static final Logger logger = LoggerFactory.getLogger(InvokeImEntryFilter.class);
@@ -37,11 +39,10 @@ public class InvokeImEntryFilter implements Filter {
             if (imEntryServerAddr == null) {
                 throw new IllegalCallerException("调用im-entry服务接口,上下文服务地址未指定");
             }
-            String uid2 = RpcContext.getServerContext().getAttachment(DubboConstant.CTX_UID);
+            // 设置调用 im-entry 服务地址
-            String uid3 = RpcContext.getServiceContext().getAttachment(DubboConstant.CTX_UID);
+            RpcContext.getServiceContext().setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr);
-            logger.info("ctxUid: {}, {}, {}", ctxUid, uid2, uid3);
+            // invocation.setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr);
-            // 设置 im-entry 服务地址
+            logger.info("invoke im-entry: uid={}, entryAddr={}", ctxUid, imEntryServerAddr);
-            invocation.setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr);
         }
         return invoker.invoke(invocation);
     }

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

@@ -0,0 +1,40 @@
+package net.sopod.soim.router.config.listener;
+
+import net.sopod.soim.router.config.ImRouterAppContextHolder;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.rpc.Exporter;
+import org.apache.dubbo.rpc.ExporterListener;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.protocol.injvm.InjvmProtocol;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * ImRouterExportListener
+ * 监听需要注册的服务URL，存储到context，后根据需要注册到注册中心
+ *
+ * @author tmy
+ * @date 2022-05-01 16:31
+ */
+public class ImRouterAPIExportListener implements ExporterListener {
+
+    private static final Logger logger = LoggerFactory.getLogger(ImRouterAPIExportListener.class);
+
+    @Override
+    public void exported(Exporter<?> exporter) throws RpcException {
+        URL invokerUrl = exporter.getInvoker().getUrl();
+        if (!InjvmProtocol.NAME.equals(invokerUrl.getProtocol())) {
+            ImRouterAppContextHolder.addRegistryInvokerUrl(invokerUrl);
+            if (ImRouterAppContextHolder.getAppServiceAddr() == null) {
+                ImRouterAppContextHolder.setAppServiceAddr(invokerUrl.getAddress());
+                logger.info("im-router registry serverAddr: {}", invokerUrl.getAddress());
+            }
+        }
+    }
+
+    @Override
+    public void unexported(Exporter<?> exporter) {
+        logger.info("unexported listener: {}", exporter.getInvoker().getInterface());
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/config/loadbalance/ImEntryServerAddressLoadBalance.java

@@ -0,0 +1,56 @@
+package net.sopod.soim.router.config.loadbalance;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.common.util.StringUtil;
+import net.sopod.soim.logic.common.util.RpcContextUtil;
+import net.sopod.soim.router.cache.RouterUser;
+import net.sopod.soim.router.cache.SoImUserCache;
+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;
+
+/**
+ * ImEntryServerAddressLoadBalance
+ *
+ * @author tmy
+ * @date 2022-05-02 14:45
+ */
+public class ImEntryServerAddressLoadBalance extends AbstractLoadBalance {
+
+    private static final Logger logger = LoggerFactory.getLogger(ImEntryServerAddressLoadBalance.class);
+
+    @Override
+    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
+        // 调用 im-entry 服务的地址
+        String entryAddr = null;
+        // 获取请求链路用户id
+        // String ctxUid = invocation.getAttachment(DubboConstant.CTX_UID);
+        String ctxUid = RpcContextUtil.getContextUid();
+        if (StringUtil.isEmpty(ctxUid)) {
+            throw new IllegalCallerException("调用im-entry服务接口,上下文uid未指定");
+        }
+        SoImUserCache soImUserCache = SoImUserCache.getInstance();
+        RouterUser routerUser = soImUserCache.get(Long.valueOf(ctxUid));
+        if (routerUser != null) {
+            entryAddr = routerUser.getImEntryAddr();
+        }
+        if (entryAddr == null) {
+            throw new IllegalCallerException("调用im-entry服务接口,上下文entry服务地址未指定");
+        }
+        logger.info("invoke im-entry: uid={}, im-entry addr={}", ctxUid, entryAddr);
+        // 返回地址为 entryAddr 的 invoker
+        for (Invoker<T> invoker : invokers) {
+            if (entryAddr.equals(invoker.getUrl().getAddress())) {
+                return invoker;
+            }
+        }
+        // TODO 自定义异常，用户重新登录
+        throw new IllegalCallerException("没有地址为 " + entryAddr + " im-entry 服务");
+    }
+
+}

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

@@ -1,25 +0,0 @@
-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/service/OnlineUserServiceImpl.java

@@ -1,7 +1,7 @@
 package net.sopod.soim.router.service;
 
 import net.sopod.soim.entry.api.service.OnlineUserService;
-import net.sopod.soim.router.api.model.RouterUser;
+import net.sopod.soim.router.cache.RouterUser;
 import net.sopod.soim.router.cache.SoImUserCache;
 import org.apache.dubbo.config.annotation.DubboService;
 

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

@@ -8,13 +8,12 @@ import net.sopod.soim.das.user.api.service.UserDasService;
 import net.sopod.soim.entry.api.service.OnlineUserService;
 import net.sopod.soim.entry.api.service.TextChatService;
 import net.sopod.soim.logic.common.model.TextChat;
-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.api.model.RegistryRes;
+import net.sopod.soim.router.cache.RouterUser;
+import net.sopod.soim.router.api.service.UserRouteService;
 import net.sopod.soim.router.cache.SoImUserCache;
-import net.sopod.soim.router.util.RpcContextUtil;
+import net.sopod.soim.router.config.ImRouterAppContextHolder;
-import net.sopod.soim.router.util.ServerContext;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
 import org.apache.dubbo.rpc.RpcContext;
@@ -22,20 +21,19 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
- * AccountStoreServiceImpl
+ * UserRouteServiceImpl
  *
  * @author tmy
  * @date 2022-04-28 9:47
  */
-@DubboService
+@DubboService()
-public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
+public class UserRouteServiceImpl implements UserRouteService {
 
-    private static final Logger logger = LoggerFactory.getLogger(UserEntryRegistryServiceImpl.class);
+    private static final Logger logger = LoggerFactory.getLogger(UserRouteServiceImpl.class);
 
     @DubboReference
     private UserDasService userDasService;
@@ -47,15 +45,19 @@ public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
     private OnlineUserService onlineUserService;
 
     @Override
-    public CacheRes registryUserEntry(Long uid, String imEntryAddr) {
+    public RegistryRes registryUserEntry(Long uid, String imEntryAddr) {
         ImUser imUser = userDasService.getUserById(uid);
+        logger.info("registry user:{}, {}", uid, imUser.getAccount());
         RouterUser routerUser = new RouterUser().setUid(uid)
                 .setAccount(imUser.getAccount())
                 .setIsOnline(Boolean.TRUE)
                 .setOnlineTime(ImClock.millis())
                 .setImEntryAddr(imEntryAddr);
         SoImUserCache.getInstance().put(uid, routerUser);
-        return CacheRes.success(0L);
+        // 接口返回 im_router_id，后续调用 im-router 负载均衡指向当前router服务
+        return new RegistryRes()
+                .setSuccess(true)
+                .setImRouterId(ImRouterAppContextHolder.IM_ROUTER_ID);
     }
 
     @Override
@@ -72,35 +74,18 @@ public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
                 .collect(Collectors.toList());
     }
 
+    /**
+     * 调用该方法时，将到 im-router 服务的路由 uid 设置为消息接受者的 uid
+     */
     @Override
     public Boolean routeTextChat(TextChat textChat) {
-        Long receiverUid = textChat.getReceiverUid();
-        // 查询 receiverUid 对应 im-entry 地址
-        RouterUser receiverUser = SoImUserCache.getInstance().get(textChat.getReceiverUid());
-        String receiverImEntryAddr = null;
-        if (receiverUser != null) {
-            receiverImEntryAddr = receiverUser.getImEntryAddr();
-            logger.info("local im-router service invoke: {}, {}", receiverUid, receiverImEntryAddr);
-        }
         // 本服务实例没有存储接收者用户信息，查询其他服务
 //        if (receiverImEntryAddr == null) {
 //            RpcContext.getServiceContext().setAttachment(DubboConstant.CTX_UID, String.valueOf(receiverUid));
 //            receiverImEntryAddr = onlineUserService.getImEntryAddrByUid(receiverUid);
 //            logger.info("other im-router service invoke: {}, {}", receiverUid, receiverImEntryAddr);
 //        }
-        // 调用该方法时，将到 im-router 服务的路由 uid 设置为消息接受者的 uid
+
-        if (receiverImEntryAddr == null) {
-            logger.info("消息接受者im-entry服务连接地址为找到");
-            return false;
-        }
-        logger.info("receiver user: {}, {}", receiverUid, receiverImEntryAddr);
-        // TODO 设置im-entry服务调用地址，优化集成到过滤器或工具类
-        RpcContext.getServiceContext().setAttachment(DubboConstant.IM_ENTRY_ADDR, receiverImEntryAddr);
-//        boolean set = RpcContextUtil.setImEntryRouteServerAddrByUid(textChat.getReceiverUid());
-//        if (!set) {
-//            logger.info("im-entry服务地址设置失败");
-//            return false;
-//        }
         Boolean send = textChatService.sendTextChat(textChat);
         if (!Boolean.TRUE.equals(send)) {
             // 未送到，消息存储，重发...

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

@@ -1,7 +1,7 @@
 package net.sopod.soim.router.util;
 
 import net.sopod.soim.common.constant.DubboConstant;
-import net.sopod.soim.router.api.model.RouterUser;
+import net.sopod.soim.router.cache.RouterUser;
 import net.sopod.soim.router.cache.SoImUserCache;
 import org.apache.dubbo.rpc.RpcContext;
 

im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.ExporterListener

@@ -0,0 +1 @@
+im_router_api_export_listener=net.sopod.soim.router.config.listener.ImRouterAPIExportListener

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

@@ -1 +1 @@
-invoke_im_entry_filter=net.sopod.soim.router.config.InvokeImEntryFilter
+invoke_im_entry_filter=net.sopod.soim.router.config.filter.InvokeImEntryFilter

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

@@ -0,0 +1 @@
+im_entry_loadbalance=net.sopod.soim.router.config.loadbalance.ImEntryServerAddressLoadBalance

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

@@ -20,11 +20,13 @@ dubbo:
     group: so-im
   protocol:
     name: dubbo
-    port: 3032
+    port: 3031
   consumer:
     check: false
-    # filter: invoke_im_entry_filter
+    # filter: invoke_im_entry_filter # 调用im-entry时设置调用地址，配合im_entry_loadbalance路由到用户对应连接的im-entry
   provider:
-    loadbalance: im_route_consistent_hash
+    listener: im_router_api_export_listener # 监听im-router需要暴露的服务，启动时不注册，数据迁移后再注册为可用服务
+    loadbalance: im_route_consistent_hash # 调用im-router时根据上下文uid路由到用户对应的im-router
     retries: 0 # 这里服务重试时会路由到非uid所在对应im-router
     timeout: 2000
+    register: false # 不自动注册，数据初始化后注册

pom.xml

@@ -249,6 +249,11 @@
                 <artifactId>spring-boot-starter-amqp</artifactId>
                 <version>${spring-boot.version}</version>
             </dependency>
+            <dependency>
+                <groupId>cglib</groupId>
+                <artifactId>cglib</artifactId>
+                <version>3.3.0</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>