From da02a2e38c252235e2c2a507b3315e63ac05e2a0 Mon Sep 17 00:00:00 2001 From: tangmingyou <234767776@qq.com> Date: Wed, 4 May 2022 23:02:39 +0800 Subject: [PATCH] =?UTF-8?q?im-router,=20im-entry=20=E8=B4=9F=E8=BD=BD?= =?UTF-8?q?=E5=9D=87=E8=A1=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 3 +- .../net/sopod/soim/client/cmd/CmdStarter.java | 7 +- .../soim/client/config/ClientConfig.java | 2 +- .../soim/common/constant/AppConstant.java | 2 + .../soim/common/constant/DubboConstant.java | 4 + .../net/sopod/soim/common/util/Collects.java | 78 +++++++++-- .../sopod/soim/common/util/StringUtil.java | 4 + .../net/sopod/soim/core/session/Account.java | 16 ++- .../sopod/soim/entry/EntryApplication.java | 10 +- .../entry/config/DubboPreInvokeFilter.java | 4 +- ...r.java => ImEntryAPIExporterListener.java} | 17 ++- ...lder.java => ImEntryAppContextHolder.java} | 6 +- .../SpringApplicationContextInitialed.java | 2 +- .../handler/auth/ReqTokenAuthHandler.java | 9 +- .../entry/service/TextChatServiceImpl.java | 4 +- .../org.apache.dubbo.rpc.ExporterListener | 2 +- im-entry/src/main/resources/application.yml | 1 + .../ImEntryServerAddressLoadBalance.java | 45 ------- .../org.apache.dubbo.rpc.cluster.LoadBalance | 1 - .../logic/common/util/RpcContextUtil.java | 32 +++++ .../user/auth/service/UserAuthService.java | 2 +- .../soim/router/api/model/RegistryRes.java | 24 ++++ .../route/ImRouterConsistentHashRoute.java | 67 ++-------- .../api/route/ImRouterDirectLoadBalance.java | 44 ++++++ .../api/route/UidConsistentHashSelector.java | 62 +++++++++ .../router/api/service/AuthStoreService.java | 2 +- ...stryService.java => UserRouteService.java} | 7 +- .../org.apache.dubbo.rpc.cluster.LoadBalance | 3 +- .../logic/user/service/ChatServiceImpl.java | 12 +- .../user/service/UserAuthServiceImpl.java | 15 ++- .../user/service/UserBizServiceImpl.java | 6 +- im-service/im-router/pom.xml | 4 + .../net/sopod/soim/router/cache/DataSync.java | 26 ++++ .../router/cache/DataSyncProxyFactory.java | 66 +++++++++ .../sopod/soim/router/cache}/RouterUser.java | 4 +- .../soim/router/cache/SoImUserCache.java | 2 - .../cache/annotation/DataSyncIgnore.java | 12 ++ .../config/ImRouterAppContextHolder.java | 50 +++++++ .../router/config/ImRouterAppOnReady.java | 126 ++++++++++++++++++ .../{ => filter}/InvokeImEntryFilter.java | 15 ++- .../listener/ImRouterAPIExportListener.java | 40 ++++++ .../ImEntryServerAddressLoadBalance.java | 56 ++++++++ .../listener/ImRouterExportListener.java | 25 ---- .../router/service/OnlineUserServiceImpl.java | 2 +- ...iceImpl.java => UserRouteServiceImpl.java} | 51 +++---- .../soim/router/util/RpcContextUtil.java | 2 +- .../org.apache.dubbo.rpc.ExporterListener | 1 + .../dubbo/org.apache.dubbo.rpc.Filter | 2 +- .../org.apache.dubbo.rpc.cluster.LoadBalance | 1 + .../src/main/resources/application.yml | 8 +- pom.xml | 5 + 51 files changed, 750 insertions(+), 241 deletions(-) rename im-entry/src/main/java/net/sopod/soim/entry/config/{DubboImEntryExporterListener.java => ImEntryAPIExporterListener.java} (56%) rename im-entry/src/main/java/net/sopod/soim/entry/config/{ApplicationContextHolder.java => ImEntryAppContextHolder.java} (77%) delete mode 100644 im-service-api/im-entry-api/src/main/java/net/sopod/soim/entry/api/route/ImEntryServerAddressLoadBalance.java delete mode 100644 im-service-api/im-entry-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance create mode 100644 im-service-api/im-logic-common/src/main/java/net/sopod/soim/logic/common/util/RpcContextUtil.java create mode 100644 im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RegistryRes.java create mode 100644 im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterDirectLoadBalance.java create mode 100644 im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/UidConsistentHashSelector.java rename im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/{UserEntryRegistryService.java => UserRouteService.java} (75%) create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java rename {im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model => im-service/im-router/src/main/java/net/sopod/soim/router/cache}/RouterUser.java (80%) create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/DataSyncIgnore.java create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppContextHolder.java create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java rename im-service/im-router/src/main/java/net/sopod/soim/router/config/{ => filter}/InvokeImEntryFilter.java (76%) create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/config/listener/ImRouterAPIExportListener.java create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/config/loadbalance/ImEntryServerAddressLoadBalance.java delete mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java rename im-service/im-router/src/main/java/net/sopod/soim/router/service/{UserEntryRegistryServiceImpl.java => UserRouteServiceImpl.java} (63%) create mode 100644 im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.ExporterListener create mode 100644 im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance diff --git a/README.md b/README.md index 7eff52f..b2a922e 100644 --- a/README.md +++ b/README.md @@ -2,8 +2,9 @@ TODO - router -> entry 负载均衡 - router 新增节点顺时针相邻节点数据一致性哈希迁移 - router 冗余节点存储数据不提供服务 -- dubbo 服务异步处理吞吐量 +- dubbo 服务异步处理提升吞吐量 - das 消息队列异步写 +- entry 监控，http 查询 entry 地址返回接口 - 功能开发： - 消息群发(im-router 群消息路由,批量uid一致性哈希路由) - 好友列表(在线状态:批量uid一致性hash, router查询) diff --git a/im-client/src/main/java/net/sopod/soim/client/cmd/CmdStarter.java b/im-client/src/main/java/net/sopod/soim/client/cmd/CmdStarter.java index 158feaa..8083023 100644 --- a/im-client/src/main/java/net/sopod/soim/client/cmd/CmdStarter.java +++ b/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(); - this.cmdDispatcher.dispatchCmd(cmd); - + 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"); diff --git a/im-client/src/main/java/net/sopod/soim/client/config/ClientConfig.java b/im-client/src/main/java/net/sopod/soim/client/config/ClientConfig.java index f28f8d6..efbd92d 100644 --- a/im-client/src/main/java/net/sopod/soim/client/config/ClientConfig.java +++ b/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; } diff --git a/im-common/src/main/java/net/sopod/soim/common/constant/AppConstant.java b/im-common/src/main/java/net/sopod/soim/common/constant/AppConstant.java index 54c7699..37ee4a1 100644 --- a/im-common/src/main/java/net/sopod/soim/common/constant/AppConstant.java +++ b/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"; 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 c12c75f..0f1f9b4 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 @@ -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 服务接口前缀 diff --git a/im-common/src/main/java/net/sopod/soim/common/util/Collects.java b/im-common/src/main/java/net/sopod/soim/common/util/Collects.java index 24f748f..f5e131e 100644 --- a/im-common/src/main/java/net/sopod/soim/common/util/Collects.java +++ b/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,17 +12,51 @@ import java.util.function.Function; */ public class Collects { - /** - * 数组翻转 - */ - 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 isEmpty(@Nullable Collection collection) { + return collection == null || collection.isEmpty(); + } + + 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 Map collect2Map(Collection collect, + Function keyProvider) { + return collect2Map(collect, keyProvider, new HashMap<>()); + } + + public static Map collect2Map(Collection collect, + Function keyProvider, + Map resultMap) { + for (T item : collect) { + resultMap.put(keyProvider.apply(item), item); + } + return resultMap; + } + + public static Map collect2KvMap(Collection collect, + Function keyProvider, + Function valueProvider) { + return collect2KvMap(collect, keyProvider, valueProvider, new HashMap<>()); + } + + public static Map collect2KvMap(Collection collect, + Function keyProvider, + Function valueProvider, + Map resultMap) { + for (T item : collect) { + resultMap.put(keyProvider.apply(item), valueProvider.apply(item)); + } + return resultMap; + } public static Map> group(Collection collect, Function groupBy, @@ -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)); + } + } diff --git a/im-common/src/main/java/net/sopod/soim/common/util/StringUtil.java b/im-common/src/main/java/net/sopod/soim/common/util/StringUtil.java index 6d37521..c22bf6e 100644 --- a/im-common/src/main/java/net/sopod/soim/common/util/StringUtil.java +++ b/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(); + } + } 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 05e9b12..9890339 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 @@ -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); } } diff --git a/im-entry/src/main/java/net/sopod/soim/entry/EntryApplication.java b/im-entry/src/main/java/net/sopod/soim/entry/EntryApplication.java index 9656fe1..951bb1f 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/EntryApplication.java +++ b/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(); } } diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java b/im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java index 95f98ed..72ad192 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java +++ b/im-entry/src/main/java/net/sopod/soim/entry/config/DubboPreInvokeFilter.java @@ -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); } 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/ImEntryAPIExporterListener.java similarity index 56% rename from im-entry/src/main/java/net/sopod/soim/entry/config/DubboImEntryExporterListener.java rename to im-entry/src/main/java/net/sopod/soim/entry/config/ImEntryAPIExporterListener.java index 60c926e..f89a8b3 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/DubboImEntryExporterListener.java +++ b/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 (ApplicationContextHolder.getDubboAppServiceAddr() == null) { - logger.info("im-entry dubbo app serverAddr: {}", invokerUrl.getAddress()); - ApplicationContextHolder.setDubboAppServiceAddr(invokerUrl.getAddress()); + if (!InjvmProtocol.NAME.equals(invokerUrl.getProtocol())) { + if (ImEntryAppContextHolder.getDubboAppServiceAddr() == null) { + ImEntryAppContextHolder.setDubboAppServiceAddr(invokerUrl.getAddress()); + logger.info("im-entry registry serverAddr: {}", invokerUrl.getAddress()); } } } @Override public void unexported(Exporter exporter) { - + logger.info("unexported listener: {}", exporter.getInvoker().getInterface()); } } diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextHolder.java b/im-entry/src/main/java/net/sopod/soim/entry/config/ImEntryAppContextHolder.java similarity index 77% rename from im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextHolder.java rename to im-entry/src/main/java/net/sopod/soim/entry/config/ImEntryAppContextHolder.java index 65ec490..9619240 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/ApplicationContextHolder.java +++ b/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 getBean(Class beanType) { @@ -23,7 +23,7 @@ public class ApplicationContextHolder { } public static void setDubboAppServiceAddr(String dubboAppServiceAddr) { - ApplicationContextHolder.dubboAppServiceAddr = dubboAppServiceAddr; + ImEntryAppContextHolder.dubboAppServiceAddr = dubboAppServiceAddr; } public static String getDubboAppServiceAddr() { diff --git a/im-entry/src/main/java/net/sopod/soim/entry/config/SpringApplicationContextInitialed.java b/im-entry/src/main/java/net/sopod/soim/entry/config/SpringApplicationContextInitialed.java index 7a00db1..1fa5ef2 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/config/SpringApplicationContextInitialed.java +++ b/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); 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 28c7388..8f1d1eb 100644 --- a/im-entry/src/main/java/net/sopod/soim/entry/handler/auth/ReqTokenAuthHandler.java +++ b/im-entry/src/main/java/net/sopod/soim/entry/handler/auth/ReqTokenAuthHandler.java @@ -1,7 +1,7 @@ package net.sopod.soim.entry.handler.auth; import com.google.protobuf.MessageLite; -import net.sopod.soim.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 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/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 deleted file mode 100644 index 81277d1..0000000 --- a/im-service-api/im-entry-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance +++ /dev/null @@ -1 +0,0 @@ -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/src/main/java/net/sopod/soim/logic/common/util/RpcContextUtil.java b/im-service-api/im-logic-common/src/main/java/net/sopod/soim/logic/common/util/RpcContextUtil.java new file mode 100644 index 0000000..0dab951 --- /dev/null +++ b/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); + } + +} diff --git a/im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/auth/service/UserAuthService.java b/im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/auth/service/UserAuthService.java index 8d4c8c1..14cd394 100644 --- a/im-service-api/im-logic-user-api/src/main/java/net/sopod/soim/logic/user/auth/service/UserAuthService.java +++ b/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); } diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RegistryRes.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RegistryRes.java new file mode 100644 index 0000000..f5d01b7 --- /dev/null +++ b/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; + +} diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java index a53d592..18c1b90 100644 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java +++ b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterConsistentHashRoute.java @@ -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); - /** - * 一致性 hash 每个节点的虚拟节点数量 - */ - private static final int VIRTUAL_NODE_SIZE = 120; - - private volatile ConsistentHashSelector selector; + private volatile UidConsistentHashSelector selector; /** * 后续如有接口版本号，构建 map 每个方法版本，对应一个 Selector @@ -40,56 +35,22 @@ public class ImRouterConsistentHashRoute extends AbstractLoadBalance { @SuppressWarnings("unchecked") protected Invoker doSelect(List> invokers, URL url, Invocation invocation) { int invokersHash = getInvokersHash(invokers); - logger.info("invokers hash: {}", invokersHash); - if (selector == null || selector.identityHashCode != invokersHash) { - selector = new ConsistentHashSelector<>(invokers, invokersHash); + // logger.info("invokers hash: {}", invokersHash); + + if (selector == null || selector.getIdentityHashCode() != invokersHash) { + // 调用节点有变化，构建新的 hash 表 + Map> serverAddrInvokerMap = Collects.collect2Map(invokers, + invoker -> invoker.getUrl().getAddress(), + new HashMap<>(6)); + selector = new UidConsistentHashSelector<>(serverAddrInvokerMap, invokersHash); } - return ((ConsistentHashSelector)selector).select(invocation); + // 获取上下文 uid, 同 RpcContext + String uid = invocation.getAttachment(DubboConstant.CTX_UID); + return (Invoker) selector.select(uid); } private int getInvokersHash(List> invokers) { return invokers.hashCode(); } - static class ConsistentHashSelector { - - private final TreeMap> virtualInvokers; - - private final int identityHashCode; - - ConsistentHashSelector(List> invokers, int identityHashCode) { - this.identityHashCode = identityHashCode; - this.virtualInvokers = new TreeMap<>(); - - for (Invoker invoker : invokers) { - String address = invoker.getUrl().getAddress(); - for (int i = 0, len = VIRTUAL_NODE_SIZE / 4; i < len; i++) { - for (int h = 0; h < 4; h++) { - long hash = hash(address + i, h); - this.virtualInvokers.put(hash, invoker); - } - } - } - } - - private static long hash(String value, int number) { - return HashAlgorithms.md5Hash(value, number); - } - - public Invoker select(Invocation invocation) { - // 获取上下文 uid, 同 RpcContext - String uid = invocation.getAttachment(DubboConstant.CTX_UID); - if (uid == null) { - throw new IllegalStateException("im-router consistent hash route, ctx uid can not be null!"); - } - long hash = hash(uid, 0); - Map.Entry> entry = virtualInvokers.ceilingEntry(hash); - if (entry == null) { - entry = virtualInvokers.firstEntry(); - } - return entry.getValue(); - } - - } - } diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterDirectLoadBalance.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/ImRouterDirectLoadBalance.java new file mode 100644 index 0000000..cb52411 --- /dev/null +++ b/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 Invoker doSelect(List> 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 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服务"); + } + +} diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/UidConsistentHashSelector.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/route/UidConsistentHashSelector.java new file mode 100644 index 0000000..833b08e --- /dev/null +++ b/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 { + + /** + * 一致性 hash 每个节点的虚拟节点数量 + */ + private static final int VIRTUAL_NODE_SIZE = 120; + + private final TreeMap virtualNodeMap; + + private final int identityHashCode; + + public UidConsistentHashSelector(Map serverAddressMap, int identityHashCode) { + this.identityHashCode = identityHashCode; + this.virtualNodeMap = new TreeMap<>(); + + // 构建虚拟节点一致性 hash 表 + for (Map.Entry 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 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; + } + +} diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/AuthStoreService.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/AuthStoreService.java index 55c5754..deed50b 100644 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/AuthStoreService.java +++ b/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 diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserEntryRegistryService.java b/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java similarity index 75% rename from im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserEntryRegistryService.java rename to im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserRouteService.java index ef60f28..a0a3b7b 100644 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/service/UserEntryRegistryService.java +++ b/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); /** * 查询在线用户列表 diff --git a/im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance b/im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance index 4502309..93b9473 100644 --- a/im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance +++ b/im-service-api/im-router-api/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance @@ -1 +1,2 @@ -im_route_consistent_hash=net.sopod.soim.router.api.route.ImRouterConsistentHashRoute \ No newline at end of file +im_route_consistent_hash=net.sopod.soim.router.api.route.ImRouterConsistentHashRoute +im_router_direct=net.sopod.soim.router.api.route.ImRouterDirectLoadBalance \ No newline at end of file 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 1b9594e..6fc5f7a 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,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())); - return userEntryRegistryService.routeTextChat(textChat); + // 设置调用 router 为消息接受者地址 + RpcContextUtil.setContextUid(textChat.getReceiverUid()); + return userRouteService.routeTextChat(textChat); } } diff --git a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java index 85ac866..d2c898e 100644 --- a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java +++ b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserAuthServiceImpl.java @@ -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.service.UserEntryRegistryService; +import net.sopod.soim.router.api.model.RegistryRes; +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(); } } diff --git a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserBizServiceImpl.java b/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserBizServiceImpl.java index 73c78ef..89a0a0e 100644 --- a/im-service/im-logic-user/src/main/java/net/sopod/soim/logic/user/service/UserBizServiceImpl.java +++ b/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 sayHi(String name) { @@ -38,7 +38,7 @@ public class UserBizServiceImpl implements UserBizService { @Override public List onlineUserList(String keyword) { logger.info("service context uid: {}", RpcContext.getServiceContext().getAttachment("uid")); - return userEntryRegistryService.onlineUserList(keyword); + return userRouteService.onlineUserList(keyword); } } diff --git a/im-service/im-router/pom.xml b/im-service/im-router/pom.xml index 4795ffe..86d1699 100644 --- a/im-service/im-router/pom.xml +++ b/im-service/im-router/pom.xml @@ -81,6 +81,10 @@ com.github.ben-manes.caffeine caffeine + + cglib + cglib + \ No newline at end of file diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java new file mode 100644 index 0000000..b566fdf --- /dev/null +++ b/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; + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java new file mode 100644 index 0000000..271b483 --- /dev/null +++ b/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 newProxyInstance(Class 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()); + } + +} diff --git a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RouterUser.java b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/RouterUser.java similarity index 80% rename from im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RouterUser.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/cache/RouterUser.java index 0f7fe74..5aff674 100644 --- a/im-service-api/im-router-api/src/main/java/net/sopod/soim/router/api/model/RouterUser.java +++ b/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; 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 index f6f8b23..468e2a5 100644 --- 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 @@ -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; diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/DataSyncIgnore.java b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/DataSyncIgnore.java new file mode 100644 index 0000000..9dfa142 --- /dev/null +++ b/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 { + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppContextHolder.java b/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppContextHolder.java new file mode 100644 index 0000000..22450d2 --- /dev/null +++ b/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 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 getRegistryInvokerUrls() { + return registryInvokerUrls; + } + + public static void setAppServiceAddr(String appServiceAddr) { + ImRouterAppContextHolder.appServiceAddr = appServiceAddr; + } + + public static String getAppServiceAddr() { + return appServiceAddr; + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java b/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java new file mode 100644 index 0000000..fbc8880 --- /dev/null +++ b/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, 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 registries = registryManager.getRegistries(); + List 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 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 consistentHashNodeMap = new HashMap<>(); + Map stringInstanceMap = Collects.collect2Map(imRouterInstances, + Instance::toInetAddr, + new HashMap<>(Collects.mapCapacity(imRouterInstances.size())) + ); + Map 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 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(); + } + } + +} 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/filter/InvokeImEntryFilter.java similarity index 76% rename from im-service/im-router/src/main/java/net/sopod/soim/router/config/InvokeImEntryFilter.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/config/filter/InvokeImEntryFilter.java index 215bb67..3ce4572 100644 --- 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/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); - String uid3 = RpcContext.getServiceContext().getAttachment(DubboConstant.CTX_UID); - logger.info("ctxUid: {}, {}, {}", ctxUid, uid2, uid3); - // 设置 im-entry 服务地址 - invocation.setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr); + // 设置调用 im-entry 服务地址 + RpcContext.getServiceContext().setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr); + // invocation.setAttachment(DubboConstant.IM_ENTRY_ADDR, imEntryServerAddr); + logger.info("invoke im-entry: uid={}, entryAddr={}", ctxUid, imEntryServerAddr); } return invoker.invoke(invocation); } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/config/listener/ImRouterAPIExportListener.java b/im-service/im-router/src/main/java/net/sopod/soim/router/config/listener/ImRouterAPIExportListener.java new file mode 100644 index 0000000..81086d8 --- /dev/null +++ b/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()); + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/config/loadbalance/ImEntryServerAddressLoadBalance.java b/im-service/im-router/src/main/java/net/sopod/soim/router/config/loadbalance/ImEntryServerAddressLoadBalance.java new file mode 100644 index 0000000..6e78e82 --- /dev/null +++ b/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 Invoker doSelect(List> 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 invoker : invokers) { + if (entryAddr.equals(invoker.getUrl().getAddress())) { + return invoker; + } + } + // TODO 自定义异常，用户重新登录 + throw new IllegalCallerException("没有地址为 " + entryAddr + " im-entry 服务"); + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java b/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java deleted file mode 100644 index 85e0f21..0000000 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/listener/ImRouterExportListener.java +++ /dev/null @@ -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) { - - } - -} 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 index 8568fba..77a879a 100644 --- 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 @@ -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; 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/UserRouteServiceImpl.java similarity index 63% rename from im-service/im-router/src/main/java/net/sopod/soim/router/service/UserEntryRegistryServiceImpl.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/service/UserRouteServiceImpl.java index 9a7b8b3..0a1b070 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/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.util.ServerContext; +import net.sopod.soim.router.config.ImRouterAppContextHolder; 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 -public class UserEntryRegistryServiceImpl implements UserEntryRegistryService { +@DubboService() +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)) { // 未送到，消息存储，重发... 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 index d1d20dd..bc251bc 100644 --- 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 @@ -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; diff --git a/im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.ExporterListener b/im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.ExporterListener new file mode 100644 index 0000000..ba360e6 --- /dev/null +++ b/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 \ No newline at end of file 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 index 2a4f817..bcc1aae 100644 --- 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 @@ -1 +1 @@ -invoke_im_entry_filter=net.sopod.soim.router.config.InvokeImEntryFilter \ No newline at end of file +invoke_im_entry_filter=net.sopod.soim.router.config.filter.InvokeImEntryFilter \ No newline at end of file diff --git a/im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance b/im-service/im-router/src/main/resources/META-INF/dubbo/org.apache.dubbo.rpc.cluster.LoadBalance new file mode 100644 index 0000000..8dae8c9 --- /dev/null +++ b/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 \ 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 81a0116..c0e5565 100644 --- a/im-service/im-router/src/main/resources/application.yml +++ b/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 # 不自动注册，数据初始化后注册 diff --git a/pom.xml b/pom.xml index ee03c4d..5ea5c47 100644 --- a/pom.xml +++ b/pom.xml @@ -249,6 +249,11 @@ spring-boot-starter-amqp ${spring-boot.version} + + cglib + cglib + 3.3.0 +