diff --git a/README.md b/README.md index 9cee990..7eff52f 100644 --- a/README.md +++ b/README.md @@ -5,11 +5,14 @@ TODO - dubbo 服务异步处理吞吐量 - das 消息队列异步写 - 功能开发： - - 消息群发 + - 消息群发(im-router 群消息路由,批量uid一致性哈希路由) - 好友列表(在线状态:批量uid一致性hash, router查询) - 聊天记录查询 + - 异/同设备，多地登录 - 集群部署, docker swarm, k8s, jenkens - 服务监控 +- websocket 网关 +- 考虑 dubbo 使用 grpc service 模块列表 diff --git a/im-common/src/main/java/net/sopod/soim/common/constant/DeviceEnum.java b/im-common/src/main/java/net/sopod/soim/common/constant/DeviceEnum.java new file mode 100644 index 0000000..d003169 --- /dev/null +++ b/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 + +} 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 index 0f40b29..c12c75f 100644 --- 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 @@ -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"; + } diff --git a/im-core/src/main/java/net/sopod/soim/core/session/Account.java b/im-core/src/main/java/net/sopod/soim/core/session/Account.java index 1661cdc..05e9b12 100644 --- a/im-core/src/main/java/net/sopod/soim/core/session/Account.java +++ b/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 + + '}'; + } + } diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextHolder.java b/im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextHolder.java similarity index 50% rename from im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextHolder.java rename to im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextHolder.java index dd58cca..65ec490 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextHolder.java +++ b/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 getBean(Class beanType) { return applicationContext.getBean(beanType); } + public static void setDubboAppServiceAddr(String dubboAppServiceAddr) { + ApplicationContextHolder.dubboAppServiceAddr = dubboAppServiceAddr; + } + + public static String getDubboAppServiceAddr() { + return dubboAppServiceAddr; + } + } diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/DubboImEntryExporterListener.java b/im-entry/src/main/java/net/sopod/soim/entry/config/DubboImEntryExporterListener.java new file mode 100644 index 0000000..60c926e --- /dev/null +++ b/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) { + + } + +} diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextInitialed.java b/im-entry/src/main/java/net/sopod/soim/entry/config/SpringApplicationContextInitialed.java similarity index 82% rename from im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextInitialed.java rename to im-entry/src/main/java/net/sopod/soim/entry/config/SpringApplicationContextInitialed.java index 6d3968d..7a00db1 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/SpringContextInitialed.java +++ b/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); 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 dd6efca..28c7388 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,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 { 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 { 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 hi = userService.sayHi("黄绿"); + CompletableFuture 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); diff --git a/im-entry/src/main/java/net/sopod/soim/entry/handler/user/ReqOnlineUserListHandler.java b/im-entry/src/main/java/net/sopod/soim/entry/handler/user/ReqOnlineUserListHandler.java index 26405fc..e91aab9 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/handler/user/ReqOnlineUserListHandler.java +++ b/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 userInfos = userService.onlineUserList(msg.getKeyword()); + List userInfos = userBizService.onlineUserList(msg.getKeyword()); List resUserInfos = userInfos.stream().map(user -> UserGroup.UserInfo.newBuilder() .setUid(user.getUid()) diff --git a/im-entry/src/main/java/net/sopod/soim/entry/server/AccountRegistry.java b/im-entry/src/main/java/net/sopod/soim/entry/server/AccountRegistry.java index 271c0b7..dfd3717 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/server/AccountRegistry.java +++ b/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 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); } diff --git a/im-entry/src/main/java/net/sopod/soim/entry/service/TextChatServiceImpl.java b/im-entry/src/main/java/net/sopod/soim/entry/service/TextChatServiceImpl.java index 2b2e688..3ae4045 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/service/TextChatServiceImpl.java +++ b/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; } diff --git a/im-entry/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.ExporterListener b/im-entry/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.ExporterListener new file mode 100644 index 0000000..0b779a8 --- /dev/null +++ b/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 \ 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 c755377..13411f7 100644 --- a/im-entry/src/main/resources/application.yml +++ b/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 diff --git a/im-service-api/im-entry-api/pom.xml b/im-service-api/im-entry-api/pom.xml index 43a8f59..ce753d5 100644 --- a/im-service-api/im-entry-api/pom.xml +++ b/im-service-api/im-entry-api/pom.xml @@ -18,6 +18,17 @@ net.sopod 1.0.0 + + im-common + net.sopod + 1.0.0 + + + org.apache.dubbo + dubbo-cluster + ${dubbo.version} + provided + diff --git a/im-service-api/im-entry-api/src/main/java/net/sopod/soim/entry/api/route/ImEntryServerAddressLoadBalance.java b/im-service-api/im-entry-api/src/main/java/net/sopod/soim/entry/api/route/ImEntryServerAddressLoadBalance.java new file mode 100644 index 0000000..20518de --- /dev/null +++ b/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 Invoker doSelect(List> invokers, URL url, Invocation invocation) { + String invokeAddr = ""; + for (Invoker 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 invoker : invokers) { + if (entryAddr.equals(invoker.getUrl().getAddress())) { + logger.info("invoker address: {}", invoker.getUrl().getAddress()); + return invoker; + } + } + throw new IllegalCallerException("没有地址为 " + entryAddr + " im-entry 服务"); + } + +} diff --git a/im-service-api/im-entry-api/src/main/java/net/sopod/soim/entry/api/service/OnlineUserService.java b/im-service-api/im-entry-api/src/main/java/net/sopod/soim/entry/api/service/OnlineUserService.java new file mode 100644 index 0000000..a714e14 --- /dev/null +++ b/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); + +} diff --git a/im-service-api/im-entry-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance b/im-service-api/im-entry-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance new file mode 100644 index 0000000..81277d1 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/im-service-api/im-logic-common/pom.xml b/im-service-api/im-logic-common/pom.xml index b2f24bd..de95ab7 100644 --- a/im-service-api/im-logic-common/pom.xml +++ b/im-service-api/im-logic-common/pom.xml @@ -11,5 +11,18 @@ im-logic-common + + + im-common + net.sopod + 1.0.0 + + + org.apache.dubbo + dubbo-cluster + ${dubbo.version} + provided + + \ No newline at end of file diff --git a/im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/service/UserService.java b/im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/service/UserBizService.java similarity index 92% rename from im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/service/UserService.java rename to im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/service/UserBizService.java index 055606b..62354d3 100644 --- a/im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/service/UserService.java +++ b/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 { /** * 异步接口测试 diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RouterUser.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RouterUser.java index f2c1721..0f7fe74 100644 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RouterUser.java +++ b/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; + } diff --git a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java index fa05b3a..1b9594e 100644 --- a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/ChatServiceImpl.java +++ b/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()); - if (imUser == null) { + ImUser receiverUser = userDasService.getNormalUserByAccount(textChat.getReceiverName()); + 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); } 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/UserBizServiceImpl.java similarity index 79% rename from im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserServiceImpl.java rename to im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserBizServiceImpl.java index 9ee8319..73c78ef 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/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 onlineUserList(String keyword) { - logger.info("client context uid: {}", RpcContext.getClientAttachment().getAttachment("uid")); - logger.info("server context uid: {}", RpcContext.getServerAttachment().getAttachment("uid")); + logger.info("service context uid: {}", RpcContext.getServiceContext().getAttachment("uid")); return userEntryRegistryService.onlineUserList(keyword); } diff --git a/im-service/im-logic-user/src/main/resources/application.yml b/im-service/im-logic-user/src/main/resources/application.yml index 42f50c4..2c0d44f 100644 --- a/im-service/im-logic-user/src/main/resources/application.yml +++ b/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 \ No newline at end of file diff --git a/im-service/im-router/README.md b/im-service/im-router/README.md new file mode 100644 index 0000000..602f79c --- /dev/null +++ b/im-service/im-router/README.md @@ -0,0 +1,5 @@ + +其他服务到 router 的服务，通过 uid 进行一致性 hash 负载均衡， + +router 到 entry 的服务，通过指定 user 所登录的 entry 节点 serverAddr 地址进行指向性路由负载均衡。 + diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/SoImUserCache.java b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/SoImUserCache.java new file mode 100644 index 0000000..f6f8b23 --- /dev/null +++ b/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 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 getRouterUserMap() { + return routerUserMap; + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/config/InvokeImEntryFilter.java b/im-service/im-router/src/main/java/net/sopod/soim/router/config/InvokeImEntryFilter.java new file mode 100644 index 0000000..215bb67 --- /dev/null +++ b/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); + } + +} 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 deleted file mode 100644 index d849ede..0000000 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterServiceRegistry.java +++ /dev/null @@ -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]; - } - -} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/service/OnlineUserServiceImpl.java b/im-service/im-router/src/main/java/net/sopod/soim/router/service/OnlineUserServiceImpl.java new file mode 100644 index 0000000..8568fba --- /dev/null +++ b/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(); + } + +} 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 9748d32..9a7b8b3 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 @@ -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 uidImEntryStore; - @DubboReference private UserDasService userDasService; @DubboReference private TextChatService textChatService; - public UserEntryRegistryServiceImpl() { - this.uidImEntryStore = new ConcurrentHashMap<>(); - } + @DubboReference + 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()); - this.uidImEntryStore.put(uid, routerUser); + .setOnlineTime(ImClock.millis()) + .setImEntryAddr(imEntryAddr); + SoImUserCache.getInstance().put(uid, routerUser); return CacheRes.success(0L); } - + @Override public List 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 stream = uidImEntryStore.values().stream(); + Stream 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 - RouterUser routerUser = uidImEntryStore.get(receiverUid); - if (routerUser == null) { - return Boolean.FALSE; + // 查询 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)) { // 未送到，消息存储，重发... diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/util/RpcContextUtil.java b/im-service/im-router/src/main/java/net/sopod/soim/router/util/RpcContextUtil.java new file mode 100644 index 0000000..d1d20dd --- /dev/null +++ b/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; + } + +} diff --git a/im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter b/im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.Filter new file mode 100644 index 0000000..2a4f817 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/im-service/im-router/src/main/resources/application.yml b/im-service/im-router/src/main/resources/application.yml index c8bc2f0..81a0116 100644 --- a/im-service/im-router/src/main/resources/application.yml +++ b/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