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..f1d5f9dd 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)); } @@ -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); @@ -219,7 +221,9 @@ public class Router implements IRouter { } /** - * 注意:这个里面其实还是调用send发送的消息 + * 注意: + * 1.这个里面其实还是调用send发送的消息 + * 2.这个argument的参数,不仅是在asyncAsk回调时要到哪个线程用, 在provider执行方也用 * * @param session * @param packet @@ -328,7 +332,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); 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 @@