im-router路由im-entry指定服务地址负载均衡

README.md

@@ -5,11 +5,14 @@ TODO
 - dubbo 服务异步处理吞吐量
 - das 消息队列异步写
 - 功能开发：
-    - 消息群发
+    - 消息群发(im-router 群消息路由,批量uid一致性哈希路由)
     - 好友列表(在线状态:批量uid一致性hash, router查询)
     - 聊天记录查询
+    - 异/同设备，多地登录
 - 集群部署, docker swarm, k8s, jenkens
 - 服务监控
+- websocket 网关
+- 考虑 dubbo 使用 grpc service
 
 
 模块列表

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

@@ -0,0 +1,19 @@
+package net.sopod.soim.common.constant;
+
+/**
+ * DeviceEnum
+ * TODO 登录时客户端传递设备类型
+ * http登录接口token中包含当前账号在线顺序
+ *
+ * @author tmy
+ * @date 2022-05-02 14:09
+ */
+public enum DeviceEnum {
+
+    PC,
+
+    ANDROID,
+
+    IOS
+
+}

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

@@ -13,4 +13,14 @@ public interface DubboConstant {
      */
     String CTX_UID = "uid";
 
+    /**
+     * 请求 im-entry 服务地址
+     */
+    String IM_ENTRY_ADDR = "entry_addr";
+
+    /**
+     * im-entry 服务接口前缀
+     */
+    String IM_ENTRY_SERVICE_API_PACK = "net.sopod.soim.entry.api.service";
+
 }

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

@@ -64,4 +64,13 @@ public class Account extends NetUser {
         }
     }
 
+    @Override
+    public String toString() {
+        return "Account{" +
+                "uid=" + uid +
+                ", name='" + name + '\'' +
+                ", channel=" + channel +
+                '}';
+    }
+
 }

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

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

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

@@ -0,0 +1,39 @@
+package net.sopod.soim.entry.config;
+
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContext;
+
+/**
+ * DubboExporterListener
+ * 获取服务注册地址 im-entry
+ *
+ * @author tmy
+ * @date 2022-05-02 09:55
+ */
+public class DubboImEntryExporterListener implements ExporterListener {
+
+    private static final Logger logger = LoggerFactory.getLogger(DubboImEntryExporterListener.class);
+
+    @Override
+    public void exported(Exporter<?> exporter) throws RpcException {
+        URL invokerUrl = exporter.getInvoker().getUrl();
+        if (DubboProtocol.NAME.equals(invokerUrl.getProtocol())) {
+            if (ApplicationContextHolder.getDubboAppServiceAddr() == null) {
+                logger.info("im-entry dubbo app serverAddr: {}", invokerUrl.getAddress());
+                ApplicationContextHolder.setDubboAppServiceAddr(invokerUrl.getAddress());
+            }
+        }
+    }
+
+    @Override
+    public void unexported(Exporter<?> exporter) {
+
+    }
+
+}

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

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

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

@@ -1,6 +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.handler.NetUserMessageHandler;
 import net.sopod.soim.core.session.Account;
 import net.sopod.soim.core.session.NetUser;
@@ -47,7 +48,7 @@ public class ReqTokenAuthHandler extends NetUserMessageHandler<Auth.ReqTokenAuth
         // 校验 token，
         // TODO 更新 router 状态成功后再升级为 account，主动关闭时发送一个消息到客户端，校验错误/超时
         Boolean isValid = userAuthService.validateToken(msg.getToken(),
-                entryServerConfig.getIp() + ":" + entryServerConfig.getPort());
+                ApplicationContextHolder.getDubboAppServiceAddr());
 
         if (!Boolean.TRUE.equals(isValid)) {
             return Auth.ResTokenAuth.newBuilder()

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

@@ -6,7 +6,7 @@ import net.sopod.soim.data.msg.hello.HelloPB;
 import net.sopod.soim.entry.delay.NetUserDelayTaskManager;
 import net.sopod.soim.entry.handler.AccountMessageHandler;
 import net.sopod.soim.logic.api.segmentid.core.SegmentIdGenerator;
-import net.sopod.soim.logic.user.service.UserService;
+import net.sopod.soim.logic.user.service.UserBizService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.Method;
 import org.apache.dubbo.rpc.RpcContext;
@@ -30,7 +30,7 @@ public class HelloHandler extends AccountMessageHandler<HelloPB.Hello> {
     private static final Logger logger = LoggerFactory.getLogger(HelloHandler.class);
 
     @DubboReference(methods = {@Method(name = "sayHello", async = true)})
-    private UserService userService;
+    private UserBizService userBizService;
 
     @Autowired
     private SegmentIdGenerator segmentIdGenerator;
@@ -39,11 +39,11 @@ public class HelloHandler extends AccountMessageHandler<HelloPB.Hello> {
     public MessageLite handle(Account account, HelloPB.Hello msg) {
         logger.info("get hello message: {}", segmentIdGenerator.nextId("im-entry-hello"));
         logger.info("msg={}, {}", msg.getId(), msg.getStr());
-        CompletableFuture<String> hi = userService.sayHi("黄绿");
+        CompletableFuture<String> hi = userBizService.sayHi("黄绿");
         hi.whenComplete((res, err) -> {
            logger.info("async hi, {}", res);
         });
-        String hello = userService.sayHello("lastJet");
+        String hello = userBizService.sayHello("lastJet");
         logger.info("hello:{}", hello);
         RpcContext.getServiceContext().getCompletableFuture().whenComplete((res, err) -> {
             logger.info("async sayHello, {}", res);

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

@@ -5,7 +5,7 @@ import net.sopod.soim.core.session.Account;
 import net.sopod.soim.data.msg.user.UserGroup;
 import net.sopod.soim.entry.handler.AccountMessageHandler;
 import net.sopod.soim.logic.common.model.UserInfo;
-import net.sopod.soim.logic.user.service.UserService;
+import net.sopod.soim.logic.user.service.UserBizService;
 import org.apache.dubbo.config.annotation.DubboReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -26,11 +26,11 @@ public class ReqOnlineUserListHandler extends AccountMessageHandler<UserGroup.Re
     private static final Logger logger = LoggerFactory.getLogger(ReqOnlineUserListHandler.class);
 
     @DubboReference
-    private UserService userService;
+    private UserBizService userBizService;
 
     @Override
     public MessageLite handle(Account account, UserGroup.ReqOnlineUserList msg) {
-        List<UserInfo> userInfos = userService.onlineUserList(msg.getKeyword());
+        List<UserInfo> userInfos = userBizService.onlineUserList(msg.getKeyword());
 
         List<UserGroup.UserInfo> resUserInfos = userInfos.stream().map(user -> UserGroup.UserInfo.newBuilder()
                 .setUid(user.getUid())

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

@@ -1,6 +1,8 @@
 package net.sopod.soim.entry.server;
 
 import net.sopod.soim.core.session.Account;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import java.util.concurrent.ConcurrentHashMap;
@@ -14,13 +16,17 @@ import java.util.concurrent.ConcurrentHashMap;
 @Service
 public class AccountRegistry {
 
+    private static final Logger logger = LoggerFactory.getLogger(AccountRegistry.class);
+
     private ConcurrentHashMap<Long, Account> accounts = new ConcurrentHashMap<>();
 
     public void put(Account account) {
+        logger.info("registry account: {}", account);
         accounts.put(account.getUid(), account);
     }
 
     public Account get(Long uid) {
+        logger.info("account list: {}", accounts);
         return accounts.get(uid);
     }
 

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

@@ -3,9 +3,12 @@ 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.server.AccountRegistry;
 import net.sopod.soim.logic.common.model.TextChat;
 import org.apache.dubbo.config.annotation.DubboService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Resource;
 
@@ -18,6 +21,8 @@ import javax.annotation.Resource;
 @DubboService
 public class TextChatServiceImpl implements TextChatService {
 
+    private static final Logger logger = LoggerFactory.getLogger(TextChatServiceImpl.class);
+
     @Resource
     private AccountRegistry accountRegistry;
 
@@ -25,6 +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());
         if (account == null) {
             return Boolean.FALSE;
         }
@@ -36,6 +42,7 @@ public class TextChatServiceImpl implements TextChatService {
                 .setTime(chat.getTime())
                 .build();
         account.channel().writeAndFlush(resTextChat);
+        logger.info("write client msg: {}", resTextChat);
         return Boolean.TRUE;
     }
 

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

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

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

@@ -9,12 +9,20 @@ dubbo:
     group: so-im
   protocol:
     name: dubbo
+    # 启动参数 -DDUBBO_IP_TO_REGISTRY=192.168.51.7 或环境变量指定服务注册ip
+    # host: 192.168.56.8 # 规则 NetUtil.isInvalidLocalHost,不能是127.x、localhost本机ip
   # provider:
     # protocol: injvm # 服务提供只供jvm内部使用, 不暴露在外面 (不启动dubbo provider服务)
-    port: 3008
+    port: 3009
   consumer:
     check: false
     filter: pre_invoke_filter
+  provider:
+    listener: im_entry_export_listener
+    loadbalance: im_entry_loadbalance
+    timeout: 2000
+    retries: 0
 
 entry-server:
   nacos-addr: 124.222.131.236:3848
+  port: 8089

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

@@ -18,6 +18,17 @@
             <groupId>net.sopod</groupId>
             <version>1.0.0</version>
         </dependency>
+        <dependency>
+            <artifactId>im-common</artifactId>
+            <groupId>net.sopod</groupId>
+            <version>1.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-cluster</artifactId>
+            <version>${dubbo.version}</version>
+            <scope>provided</scope>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -0,0 +1,45 @@
+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/java/net/sopod/soim/entry/api/service/OnlineUserService.java

@@ -0,0 +1,13 @@
+package net.sopod.soim.entry.api.service;
+
+/**
+ * OnlineUserService
+ *
+ * @author tmy
+ * @date 2022-05-02 22:44
+ */
+public interface OnlineUserService {
+
+    String getImEntryAddrByUid(Long uid);
+
+}

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

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

im-service-api/im-logic-common/pom.xml

@@ -11,5 +11,18 @@
 
     <artifactId>im-logic-common</artifactId>
 
+    <dependencies>
+        <dependency>
+            <artifactId>im-common</artifactId>
+            <groupId>net.sopod</groupId>
+            <version>1.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-cluster</artifactId>
+            <version>${dubbo.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
 
 </project>

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

@@ -12,7 +12,7 @@ import java.util.concurrent.CompletableFuture;
  * @author tmy
  * @date 2022-03-27 22:42
  */
-public interface UserService {
+public interface UserBizService {
 
     /**
      * 异步接口测试

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

@@ -22,4 +22,6 @@ public class RouterUser {
     /** 在线时间戳 */
     private long onlineTime;
 
+    private String imEntryAddr;
+
 }

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

@@ -1,11 +1,13 @@
 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 org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.dubbo.config.annotation.DubboService;
+import org.apache.dubbo.rpc.RpcContext;
 
 import java.util.Objects;
 
@@ -28,12 +30,13 @@ public class ChatServiceImpl implements ChatService {
     public Boolean textChat(TextChat textChat) {
         if (textChat.getReceiverUid() == null
                 || Objects.equals(textChat.getReceiverUid(), 0L)) {
-            ImUser imUser = userDasService.getNormalUserByAccount(textChat.getReceiverName());
+            ImUser receiverUser = userDasService.getNormalUserByAccount(textChat.getReceiverName());
-            if (imUser == null) {
+            if (receiverUser == null) {
                 return false;
             }
-            textChat.setReceiverUid(imUser.getId());
+            textChat.setReceiverUid(receiverUser.getId());
         }
+        RpcContext.getServiceContext().setAttachment(DubboConstant.CTX_UID, String.valueOf(textChat.getReceiverUid()));
         return userEntryRegistryService.routeTextChat(textChat);
     }
 

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

@@ -18,9 +18,9 @@ import java.util.concurrent.CompletableFuture;
  * @date 2022-04-04 10:03
  */
 @DubboService
-public class UserServiceImpl implements UserService {
+public class UserBizServiceImpl implements UserBizService {
 
-    private static final Logger logger = LoggerFactory.getLogger(UserServiceImpl.class);
+    private static final Logger logger = LoggerFactory.getLogger(UserBizServiceImpl.class);
 
     @DubboReference
     private UserEntryRegistryService userEntryRegistryService;
@@ -37,8 +37,7 @@ public class UserServiceImpl implements UserService {
 
     @Override
     public List<UserInfo> onlineUserList(String keyword) {
-        logger.info("client context uid: {}", RpcContext.getClientAttachment().getAttachment("uid"));
+        logger.info("service context uid: {}", RpcContext.getServiceContext().getAttachment("uid"));
-        logger.info("server context uid: {}", RpcContext.getServerAttachment().getAttachment("uid"));
         return userEntryRegistryService.onlineUserList(keyword);
     }
 

im-service/im-logic-user/src/main/resources/application.yml

@@ -12,3 +12,6 @@ dubbo:
     port: 3004
   consumer:
     check: false
+  provider:
+    retries: 0
+    timeout: 2000

im-service/im-router/README.md

@@ -0,0 +1,5 @@
+
+其他服务到 router 的服务，通过 uid 进行一致性 hash 负载均衡，
+
+router 到 entry 的服务，通过指定 user 所登录的 entry 节点 serverAddr 地址进行指向性路由负载均衡。
+

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

@@ -0,0 +1,37 @@
+package net.sopod.soim.router.cache;
+
+import net.sopod.soim.router.api.model.RouterUser;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * ImUserCache
+ * 在线用户属性缓存
+ *
+ * @author tmy
+ * @date 2022-05-02 14:07
+ */
+public class SoImUserCache {
+
+    private static final SoImUserCache INSTANCE = new SoImUserCache();
+
+    private final ConcurrentHashMap<Long, RouterUser> routerUserMap = new ConcurrentHashMap<>(128);
+
+    public static SoImUserCache getInstance() {
+        return INSTANCE;
+    }
+
+    public void put(Long uid, RouterUser routerUser) {
+        routerUserMap.put(uid, routerUser);
+    }
+
+    public RouterUser get(Long uid) {
+        return routerUserMap.get(uid);
+    }
+
+    public Map<Long, RouterUser> getRouterUserMap() {
+        return routerUserMap;
+    }
+
+}

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

@@ -0,0 +1,49 @@
+package net.sopod.soim.router.config;
+
+import net.sopod.soim.common.constant.DubboConstant;
+import net.sopod.soim.router.api.model.RouterUser;
+import net.sopod.soim.router.cache.SoImUserCache;
+import org.apache.dubbo.rpc.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * InvokeImEntryFilter
+ * 调用 im-entry 服务 serverAddr 上下文设置
+ *
+ * @author tmy
+ * @date 2022-05-02 14:56
+ */
+public class InvokeImEntryFilter implements Filter {
+
+    private static final Logger logger = LoggerFactory.getLogger(InvokeImEntryFilter.class);
+
+    @Override
+    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
+        String serviceInterface = invoker.getUrl().getServiceInterface();
+        logger.info("invoke interface: {}", serviceInterface);
+        // 判断是 im-entry 服务接口
+        if (serviceInterface.startsWith(DubboConstant.IM_ENTRY_SERVICE_API_PACK)) {
+            String imEntryServerAddr = null;
+            // 获取请求链路用户id
+            String ctxUid = invocation.getAttachment(DubboConstant.CTX_UID);
+            if (ctxUid != null) {
+                SoImUserCache soImUserCache = SoImUserCache.getInstance();
+                RouterUser routerUser = soImUserCache.get(Long.valueOf(ctxUid));
+                if (routerUser != null) {
+                    imEntryServerAddr = routerUser.getImEntryAddr();
+                }
+            }
+            if (imEntryServerAddr == null) {
+                throw new IllegalCallerException("调用im-entry服务接口,上下文服务地址未指定");
+            }
+            String uid2 = RpcContext.getServerContext().getAttachment(DubboConstant.CTX_UID);
+            String uid3 = RpcContext.getServiceContext().getAttachment(DubboConstant.CTX_UID);
+            logger.info("ctxUid: {}, {}, {}", ctxUid, uid2, uid3);
+            // 设置 im-entry 服务地址
+            invocation.setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr);
+        }
+        return invoker.invoke(invocation);
+    }
+
+}

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

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

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

@@ -0,0 +1,23 @@
+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.SoImUserCache;
+import org.apache.dubbo.config.annotation.DubboService;
+
+/**
+ * SoImUserServiceImpl
+ *
+ * @author tmy
+ * @date 2022-05-02 22:45
+ */
+@DubboService
+public class OnlineUserServiceImpl implements OnlineUserService {
+
+    @Override
+    public String getImEntryAddrByUid(Long uid) {
+        RouterUser routerUser = SoImUserCache.getInstance().get(uid);
+        return routerUser == null ? null : routerUser.getImEntryAddr();
+    }
+
+}

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

@@ -1,18 +1,23 @@
 package net.sopod.soim.router.service;
 
+import net.sopod.soim.common.constant.DubboConstant;
 import net.sopod.soim.common.util.ImClock;
 import net.sopod.soim.common.util.StringUtil;
 import net.sopod.soim.das.user.api.model.entity.ImUser;
 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.cache.SoImUserCache;
+import net.sopod.soim.router.util.RpcContextUtil;
 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;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -32,17 +37,14 @@ public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
 
     private static final Logger logger = LoggerFactory.getLogger(UserEntryRegistryServiceImpl.class);
 
-    private final ConcurrentHashMap<Long, RouterUser> uidImEntryStore;
-
     @DubboReference
     private UserDasService userDasService;
 
     @DubboReference
     private TextChatService textChatService;
 
-    public UserEntryRegistryServiceImpl() {
+    @DubboReference
-        this.uidImEntryStore = new ConcurrentHashMap<>();
+    private OnlineUserService onlineUserService;
-    }
 
     @Override
     public CacheRes registryUserEntry(Long uid, String imEntryAddr) {
@@ -50,16 +52,18 @@ public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
         RouterUser routerUser = new RouterUser().setUid(uid)
                 .setAccount(imUser.getAccount())
                 .setIsOnline(Boolean.TRUE)
-                .setOnlineTime(ImClock.millis());
+                .setOnlineTime(ImClock.millis())
-        this.uidImEntryStore.put(uid, routerUser);
+                .setImEntryAddr(imEntryAddr);
+        SoImUserCache.getInstance().put(uid, routerUser);
         return CacheRes.success(0L);
     }
 
-
+    @Override
     public List<UserInfo> onlineUserList(String keyword) {
-        logger.info("client context uid: {}", ServerContext.getContextUid());
+        //logger.info("client context uid: {}", ServerContext.getContextUid());
+        logger.info("client context uid: {}", RpcContext.getServiceContext().getAttachment(DubboConstant.CTX_UID));
 
-        Stream<RouterUser> stream = uidImEntryStore.values().stream();
+        Stream<RouterUser> stream = SoImUserCache.getInstance().getRouterUserMap().values().stream();
         if (!StringUtil.isEmpty(keyword)) {
             // 根据关键词过滤
             stream = stream.filter(user -> user.getAccount().contains(keyword));
@@ -71,11 +75,32 @@ public class UserEntryRegistryServiceImpl implements UserEntryRegistryService {
     @Override
     public Boolean routeTextChat(TextChat textChat) {
         Long receiverUid = textChat.getReceiverUid();
-        // TODO 负载均衡路由 receiver 所在 entry
+        // 查询 receiverUid 对应 im-entry 地址
-        RouterUser routerUser = uidImEntryStore.get(receiverUid);
+        RouterUser receiverUser = SoImUserCache.getInstance().get(textChat.getReceiverUid());
-        if (routerUser == null) {
+        String receiverImEntryAddr = null;
-            return Boolean.FALSE;
+        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

@@ -0,0 +1,31 @@
+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.SoImUserCache;
+import org.apache.dubbo.rpc.RpcContext;
+
+import java.util.Objects;
+
+/**
+ * RpcContextUtil
+ *
+ * @author tmy
+ * @date 2022-05-02 22:27
+ */
+public class RpcContextUtil {
+
+    public static boolean setImEntryRouteServerAddrByUid(Long uid) {
+        Objects.requireNonNull(uid, "设置im-entry路由参数uid不能为空");
+        RpcContext.getServiceContext().setAttachment(DubboConstant.CTX_UID, uid);
+        RouterUser routerUser = SoImUserCache.getInstance().get(uid);
+        String imEntryAddr;
+        if (routerUser == null
+            || null == (imEntryAddr = routerUser.getImEntryAddr())) {
+            return false;
+        }
+        RpcContext.getServiceContext().setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryAddr);
+        return true;
+    }
+
+}

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

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

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

@@ -20,8 +20,11 @@ dubbo:
     group: so-im
   protocol:
     name: dubbo
-    port: 3033
+    port: 3032
   consumer:
     check: false
+    # filter: invoke_im_entry_filter
   provider:
     loadbalance: im_route_consistent_hash
+    retries: 0 # 这里服务重试时会路由到非uid所在对应im-router
+    timeout: 2000