From 96c5917e0316d5a4fff2a0286a53b3cd361e68d4 Mon Sep 17 00:00:00 2001 From: jianan <1072772483@qq.com> Date: Fri, 8 Jul 2022 15:42:23 +0800 Subject: [PATCH] =?UTF-8?q?perf[module]:=E7=BD=91=E5=85=B3=E8=BD=AC?= =?UTF-8?q?=E5=8F=91=E6=B6=88=E6=81=AF=20GatewayTest=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- net/src/main/java/com/zfoo/net/NetContext.java | 8 ++++++++ .../java/com/zfoo/net/handler/GatewayRouteHandler.java | 10 ++++++++-- .../net/core/gateway/GatewayProviderController.java | 7 +++++++ .../java/com/zfoo/net/core/gateway/GatewayTest.java | 5 ++++- 4 files changed, 27 insertions(+), 3 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/NetContext.java b/net/src/main/java/com/zfoo/net/NetContext.java index 5fdfa130..70633418 100644 --- a/net/src/main/java/com/zfoo/net/NetContext.java +++ b/net/src/main/java/com/zfoo/net/NetContext.java @@ -54,8 +54,16 @@ public class NetContext implements ApplicationListener, private ApplicationContext applicationContext; + /** + * 使用Router时，使用的send方法依然是通过Session发送消息 + * 如：创建了一个TcpClient，那明确得到了这个Session，则可以通过这个session发送消息 + */ private IRouter router; + /** + * 使用Consumer时，提供的接口是没有Session参数的，都是通过负载均衡策略从而拿到Session，然后再发送消息。 + * 如：通过web端向游戏服务器发送请求，web端压根没保存任何session，因此就要通过Consumer去请求 + */ private IConsumer consumer; private IConfigManager configManager; diff --git a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java index cbfdf3de..45f30e13 100644 --- a/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java +++ b/net/src/main/java/com/zfoo/net/handler/GatewayRouteHandler.java @@ -79,9 +79,12 @@ public class GatewayRouteHandler extends ServerRouteHandler { } var signalAttachment = (SignalAttachment) decodedPacketInfo.getAttachment(); + + // 把客户端信息包装为一个GatewayAttachment,因此通过这个网关附加包可以得到玩家的uid、sid之类的信息 var gatewayAttachment = new GatewayAttachment(session, signalAttachment); // 网关优先使用IGatewayLoadBalancer作为一致性hash的计算参数，然后才会使用客户端的session做参数 + // 例子：以聊天服务来说，玩家知道自己在哪个群组groupId中，那往这个群发送消息时，会在Packet中带上这个groupId做为一致性hash就可以了。 if (packet instanceof IGatewayLoadBalancer) { var loadBalancerConsistentHashObject = ((IGatewayLoadBalancer) packet).loadBalancerConsistentHashObject(); gatewayAttachment.useExecutorConsistentHash(loadBalancerConsistentHashObject); @@ -95,14 +98,15 @@ public class GatewayRouteHandler extends ServerRouteHandler { return; } } - // 再使用session的sid做一致性hash，因为每次客户端连接过来sid都会改变，所以客户端重写建立连接的话可能会被路由到其它的服务器 + // 再使用session的sid做一致性hash，因为每次客户端连接过来sid都会改变，所以客户端重新建立连接的话可能会被路由到其它的服务器 // 如果有特殊需求的话，可以考虑去重写网关的转发策略 + // 拿着玩家的sid做一致性hash，那肯定是：一旦重连sid就会一直变化。所以：一般情况下除非自己创建TcpClient，否则逻辑不应该走到这里。 而是走上面的通过UID做一致性hash var sid = session.getSid(); forwardingPacket(packet, gatewayAttachment, sid); } /** - * 转发网关收到的包 + * 转发网关收到的包到Provider */ private void forwardingPacket(IPacket packet, IAttachment attachment, Object argument) { try { @@ -124,6 +128,8 @@ public class GatewayRouteHandler extends ServerRouteHandler { var sid = session.getSid(); var uid = (Long) session.getAttribute(AttributeType.UID); + + // 连接到网关的客户端断开了连接 EventBus.asyncSubmit(GatewaySessionInactiveEvent.valueOf(sid, uid == null ? 0 : uid.longValue())); super.channelInactive(ctx); diff --git a/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java b/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java index 86e5f910..f8856f29 100644 --- a/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java +++ b/net/src/test/java/com/zfoo/net/core/gateway/GatewayProviderController.java @@ -33,6 +33,13 @@ public class GatewayProviderController { private static final Logger logger = LoggerFactory.getLogger(GatewayProviderController.class); + /** + * 注意：这里第2个请求参数以Request结尾，那么第3个参数必须是 GatewayAttachment类型(参加：PacketBus中扫描时的校验) + * + * @param session + * @param request + * @param gatewayAttachment + */ @PacketReceiver public void atGatewayToProviderRequest(Session session, GatewayToProviderRequest request, GatewayAttachment gatewayAttachment) { logger.info("provider receive [packet:{}] from client", JsonUtils.object2String(request)); diff --git a/net/src/test/java/com/zfoo/net/core/gateway/GatewayTest.java b/net/src/test/java/com/zfoo/net/core/gateway/GatewayTest.java index 881013fe..56795a1b 100644 --- a/net/src/test/java/com/zfoo/net/core/gateway/GatewayTest.java +++ b/net/src/test/java/com/zfoo/net/core/gateway/GatewayTest.java @@ -77,6 +77,8 @@ public class GatewayTest { public void startGateway() { var context = new ClassPathXmlApplicationContext("gateway/gateway_consistent_session_config.xml"); SessionUtils.printSessionInfo(); + + // 注意：这里创建的是GatewayServer里面是GatewayRouteHandler(而不是BaseRouteHandler),里面会通过ConsumerSession把消息转发到Provider var gatewayServer = new GatewayServer(HostAndPort.valueOf("127.0.0.1:9000"), null); gatewayServer.start(); ThreadUtils.sleep(Long.MAX_VALUE); @@ -105,7 +107,8 @@ public class GatewayTest { var thread = new Thread(() -> { for (int j = 0; j < 10000; j++) { try { - // 注意：这里的ask请求参数是 xxxRequest，不是xxxAsk + // 注意：这里的第2个请求参数是 xxxRequest，不是xxxAsk。 因为这里是网关要将数据转发给Provider的，因此当然不能是xxxAsk这种请求。 + // 第3个参数argument是null，这样子随机一个服务提供者进行消息处理 var response = NetContext.getRouter().syncAsk(session, request, GatewayToProviderResponse.class, null).packet(); logger.info("客户端请求[{}]收到消息[{}]", atomicInteger.incrementAndGet(), JsonUtils.object2String(response)); } catch (Exception e) {