From e8fb6fa33dbdf65cabdd974924400abd3ed3e424 Mon Sep 17 00:00:00 2001 From: jianan <1072772483@qq.com> Date: Fri, 8 Jul 2022 12:10:36 +0800 Subject: [PATCH 1/2] =?UTF-8?q?perf[module]:=20=E5=86=8D=E6=AC=A1=E7=90=86?= =?UTF-8?q?=E8=A7=A3asyncAsk=E5=BC=82=E6=AD=A5=E5=9B=9E=E8=B0=83=E7=9A=84a?= =?UTF-8?q?rgument=E5=8F=82=E6=95=B0=E7=9A=84=E4=BD=9C=E7=94=A8:=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E6=96=B9=20=E5=92=8C=20=E6=89=A7=E8=A1=8C=E6=96=B9?= =?UTF-8?q?=E9=83=BD=E7=94=A8=E5=88=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- net/src/main/java/com/zfoo/net/router/Router.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/router/Router.java b/net/src/main/java/com/zfoo/net/router/Router.java index 0c3a7843..df29d7d8 100644 --- a/net/src/main/java/com/zfoo/net/router/Router.java +++ b/net/src/main/java/com/zfoo/net/router/Router.java @@ -148,7 +148,7 @@ public class Router implements IRouter { } // 正常发送消息的接收,把客户端的业务请求包装下到路由策略指定的线程进行业务处理 - // 注意：像客户端以asyncAsk发送请求，在服务器处理完后返回结果，也是进入这个receive方法，但是attachment不为空，会提前return掉不会走到这 + // 注意：像客户端以asyncAsk发送请求，在服务器处理完后返回结果，在请求方也是进入这个receive方法，但是attachment不为空，会提前return掉不会走到这 TaskBus.submit(new PacketReceiverTask(session, packet, attachment)); } @@ -219,7 +219,9 @@ public class Router implements IRouter { } /** - * 注意：这个里面其实还是调用send发送的消息 + * 注意： + * 1.这个里面其实还是调用send发送的消息 + * 2.这个argument的参数，不仅是在asyncAsk回调时要到哪个线程用， 在provider执行方也用 * * @param session * @param packet @@ -328,7 +330,8 @@ public class Router implements IRouter { } } - // 调用PacketReceiver,进行真正的业务处理 + // 调用PacketReceiver,进行真正的业务处理,这个submit只是根据packet找到protocolId，然后进行反射调用 + // 这个在哪个线程处理取决于：这个上层的PacketReceiverTask被丢到了哪个线程中 PacketBus.submit(session, packet, attachment); } catch (Exception e) { logger.error(StringUtils.format("e[uid:{}][sid:{}]未知exception异常", session.getAttribute(AttributeType.UID), session.getSid(), e.getMessage()), e); From 8216698c4dc09a23ed6a31495c6b5bf98d64260c Mon Sep 17 00:00:00 2001 From: jianan <1072772483@qq.com> Date: Fri, 8 Jul 2022 14:46:04 +0800 Subject: [PATCH 2/2] =?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=E5=88=B0Provider=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/java/com/zfoo/net/router/Router.java | 2 ++ .../com/zfoo/net/core/gateway/GatewayTest.java | 17 ++++++++++++++++- .../gateway_consistent_session_config.xml | 7 ++----- .../test/resources/provider/provider_config.xml | 1 + 4 files changed, 21 insertions(+), 6 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/router/Router.java b/net/src/main/java/com/zfoo/net/router/Router.java index df29d7d8..f1d5f9dd 100644 --- a/net/src/main/java/com/zfoo/net/router/Router.java +++ b/net/src/main/java/com/zfoo/net/router/Router.java @@ -196,6 +196,8 @@ public class Router implements IRouter { try { SignalBridge.addSignalAttachment(clientSignalAttachment); + + // 里面调用的依然是：send方法发送消息 send(session, packet, clientSignalAttachment); IPacket responsePacket = clientSignalAttachment.getResponseFuture().get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS); 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 d87b6e0e..881013fe 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 @@ -31,6 +31,8 @@ import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; /** + * 这是客户端连接网关，网关转发到服务提供者的测试用例 + * * @author jaysunxiao * @version 3.0 */ @@ -40,7 +42,12 @@ public class GatewayTest { private static final Logger logger = LoggerFactory.getLogger(GatewayTest.class); /** - * 启动zookeeper，依次运行下面的测试方法启动服务提供者，网关，然后运行clientTest，消息会通过网关转发到服务提供者 + * 启动zookeeper，依次运行下面的测试方法启动： + * 1.服务提供者 + * 2.网关 + * 3.然后运行clientTest + *

+ * 消息会通过网关转发到服务提供者 */ @Test public void startProvider0() { @@ -63,6 +70,9 @@ public class GatewayTest { ThreadUtils.sleep(Long.MAX_VALUE); } + /** + * 这是网关 + */ @Test public void startGateway() { var context = new ClassPathXmlApplicationContext("gateway/gateway_consistent_session_config.xml"); @@ -72,11 +82,15 @@ public class GatewayTest { ThreadUtils.sleep(Long.MAX_VALUE); } + /** + * 这里是客户端，客户端先请求数据到到网关(毕竟自己连接的就是网关) + */ @Test public void clientTest() { var context = new ClassPathXmlApplicationContext("gateway/gateway_client_config.xml"); SessionUtils.printSessionInfo(); + // 这里的地址是网关的地址 var client = new TcpClient(HostAndPort.valueOf("127.0.0.1:9000")); var session = client.start(); @@ -91,6 +105,7 @@ public class GatewayTest { var thread = new Thread(() -> { for (int j = 0; j < 10000; j++) { try { + // 注意：这里的ask请求参数是 xxxRequest，不是xxxAsk var response = NetContext.getRouter().syncAsk(session, request, GatewayToProviderResponse.class, null).packet(); logger.info("客户端请求[{}]收到消息[{}]", atomicInteger.incrementAndGet(), JsonUtils.object2String(response)); } catch (Exception e) { diff --git a/net/src/test/resources/gateway/gateway_consistent_session_config.xml b/net/src/test/resources/gateway/gateway_consistent_session_config.xml index 3c2f1fd6..85f0f8b7 100644 --- a/net/src/test/resources/gateway/gateway_consistent_session_config.xml +++ b/net/src/test/resources/gateway/gateway_consistent_session_config.xml @@ -24,12 +24,9 @@ - - - - + - + diff --git a/net/src/test/resources/provider/provider_config.xml b/net/src/test/resources/provider/provider_config.xml index e54749f9..62ca309b 100644 --- a/net/src/test/resources/provider/provider_config.xml +++ b/net/src/test/resources/provider/provider_config.xml @@ -24,6 +24,7 @@ +