mirror of
https://github.com/tiennm99/zfoo.git
synced 2026-08-21 00:25:23 +00:00
perf[module]:网关转发消息 GatewayTest注释
This commit is contained in:
@@ -54,8 +54,16 @@ public class NetContext implements ApplicationListener<ApplicationContextEvent>,
|
||||
|
||||
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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user