tiennm99/zfoo
1
0
Fork 0
mirror of https://github.com/tiennm99/zfoo.git synced 2026-08-21 08:27:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf[net]: 为了简化代码结构，消息接收提取出PacketBus包接收总线

Browse Source
This commit is contained in:
jaysunxiao
2021-06-26 11:54:16 +08:00
parent 174400e791
commit 3ff1e7de0a
6 changed files with 133 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
net/src/main/java/com/zfoo/net/NetContext.java
+7 -14
View File
@@ -20,7 +20,6 @@ import com.zfoo.net.core.AbstractServer;
import com.zfoo.net.core.tcp.TcpClient;
import com.zfoo.net.dispatcher.manager.IPacketDispatcher;
import com.zfoo.net.packet.service.IPacketService;
import com.zfoo.net.schema.NetProcessor;
import com.zfoo.net.session.manager.ISessionManager;
import com.zfoo.net.task.TaskManager;
import com.zfoo.protocol.exception.ExceptionUtils;
@@ -104,14 +103,8 @@ public class NetContext implements ApplicationListener<ApplicationContextEvent>,
instance.consumer = applicationContext.getBean(IConsumer.class);
instance.sessionManager = applicationContext.getBean(ISessionManager.class);
var beanNames = applicationContext.getBeanDefinitionNames();
var processor = applicationContext.getBean(NetProcessor.class);
for (var beanName : beanNames) {
processor.postProcessAfterInitialization(applicationContext.getBean(beanName), beanName);
}
NetContext.getPacketService().init();
NetContext.getConfigManager().initRegistry();
instance.packetService.init();
instance.configManager.initRegistry();
} else if (event instanceof ContextClosedEvent) {
shutdownBefore();
@@ -120,11 +113,6 @@ public class NetContext implements ApplicationListener<ApplicationContextEvent>,
}
@Override
public int getOrder() {
return 0;
}
public synchronized static void shutdownBefore() {
SchedulerContext.shutdown();
}
@@ -158,4 +146,9 @@ public class NetContext implements ApplicationListener<ApplicationContextEvent>,
logger.info("Net shutdown gracefully.");
}
@Override
public int getOrder() {
return 0;
}
}
net/src/main/java/com/zfoo/net/dispatcher/manager/IPacketDispatcher.java
+1 -3
View File
@@ -36,7 +36,7 @@ public interface IPacketDispatcher {
void receive(Session session, IPacket packet, @Nullable IPacketAttachment packetAttachment);
void doReceive(Session session, IPacket packet, @Nullable IPacketAttachment packetAttachment);
void atReceiver(Session session, IPacket packet, @Nullable IPacketAttachment packetAttachment);
/**
* attention：syncRequest和asyncRequest只能客户端调用
@@ -61,6 +61,4 @@ public interface IPacketDispatcher {
<T extends IPacket> AsyncAnswer<T> asyncAsk(Session session, IPacket packet, @Nullable Class<T> answerClass, @Nullable Object argument);
void registerPacketReceiverDefinition(Object bean);
}
net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java
+120
View File
@@ -0,0 +1,120 @@
package com.zfoo.net.dispatcher.manager;
import com.zfoo.event.model.event.IEvent;
import com.zfoo.net.dispatcher.model.anno.PacketReceiver;
import com.zfoo.net.dispatcher.model.vo.EnhanceUtils;
import com.zfoo.net.dispatcher.model.vo.IPacketReceiver;
import com.zfoo.net.dispatcher.model.vo.PacketReceiverDefinition;
import com.zfoo.net.packet.model.GatewayPacketAttachment;
import com.zfoo.net.packet.model.IPacketAttachment;
import com.zfoo.net.packet.service.PacketService;
import com.zfoo.net.session.model.Session;
import com.zfoo.protocol.IPacket;
import com.zfoo.protocol.ProtocolManager;
import com.zfoo.protocol.collection.CollectionUtils;
import com.zfoo.protocol.util.AssertionUtils;
import com.zfoo.protocol.util.ReflectionUtils;
import com.zfoo.protocol.util.StringUtils;
import javassist.CannotCompileException;
import javassist.NotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
/**
* @author jaysunxiao
* @version 3.0
*/
public abstract class PacketBus {
private static final Logger logger = LoggerFactory.getLogger(PacketBus.class);
/**
* 客户端和服务端都有接受packet的方法，packetReceiverList对应的就是包的接收方法
*/
private static final IPacketReceiver[] packetReceiverList = new IPacketReceiver[ProtocolManager.MAX_PROTOCOL_NUM];
/**
* 正常消息的接收
* <p>
* 发送者同时能发送多个包
* 接收者同时只能处理一个session的一个包，同一个发送者发送过来的包排队处理
*/
public static void submit(Session session, IPacket packet, IPacketAttachment packetAttachment) {
var packetReceiver = packetReceiverList[packet.protocolId()];
if (packetReceiver == null) {
throw new RuntimeException(StringUtils.format("no any packetReceiverDefinition found for this [packet:{}]", packet.getClass().getName()));
}
// 调用PacketReceiver
packetReceiver.invoke(session, packet, packetAttachment);
}
public static void registerPacketReceiverDefinition(Object bean) {
var clazz = bean.getClass();
var methods = ReflectionUtils.getMethodsByAnnoInPOJOClass(clazz, PacketReceiver.class);
if (CollectionUtils.isNotEmpty(methods) && !ReflectionUtils.isPojoClass(clazz)) {
logger.warn("消息注册类不是POJO类，父类的不会被扫描到");
}
for (var method : methods) {
var paramClazzs = method.getParameterTypes();
AssertionUtils.isTrue(paramClazzs.length == 2 || paramClazzs.length == 3
, "[class:{}] [method:{}] must have two or three parameter!", bean.getClass().getName(), method.getName());
AssertionUtils.isTrue(Session.class.isAssignableFrom(paramClazzs[0])
, "[class:{}] [method:{}],the first parameter must be Session type parameter Exception.", bean.getClass().getName(), method.getName());
AssertionUtils.isTrue(IPacket.class.isAssignableFrom(paramClazzs[1])
, "[class:{}] [method:{}],the second parameter must be IPacket type parameter Exception.", bean.getClass().getName(), method.getName());
AssertionUtils.isTrue(paramClazzs.length != 3 || IPacketAttachment.class.isAssignableFrom(paramClazzs[2])
, "[class:{}] [method:{}],the third parameter must be IPacketAttachment type parameter Exception.", bean.getClass().getName(), method.getName());
var packetClazz = (Class<? extends IEvent>) paramClazzs[1];
var attachmentClazz = paramClazzs.length == 3 ? paramClazzs[2] : null;
var packetName = packetClazz.getCanonicalName();
var methodName = method.getName();
AssertionUtils.isTrue(Modifier.isPublic(method.getModifiers())
, "[class:{}] [method:{}] [packet:{}] must use 'public' as modifier!", bean.getClass().getName(), methodName, packetName);
AssertionUtils.isTrue(!Modifier.isStatic(method.getModifiers())
, "[class:{}] [method:{}] [packet:{}] can not use 'static' as modifier!", bean.getClass().getName(), methodName, packetName);
var expectedMethodName = StringUtils.format("at{}", packetClazz.getSimpleName());
AssertionUtils.isTrue(methodName.equals(expectedMethodName)
, "[class:{}] [method:{}] [packet:{}] expects '{}' as method name!", bean.getClass().getName(), methodName, packetName, expectedMethodName);
// 如果以Request结尾的请求，那么attachment应该为GatewayAttachment
// 如果以Ask结尾的请求，那么attachment不能为GatewayAttachment
if (attachmentClazz != null) {
if (packetName.endsWith(PacketService.NET_REQUEST_SUFFIX)) {
AssertionUtils.isTrue(attachmentClazz.equals(GatewayPacketAttachment.class)
, "[class:{}] [method:{}] [packet:{}] must use [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayPacketAttachment.class.getCanonicalName());
} else if (packetName.endsWith(PacketService.NET_ASK_SUFFIX)) {
AssertionUtils.isTrue(!attachmentClazz.equals(GatewayPacketAttachment.class)
, "[class:{}] [method:{}] [packet:{}] can not match with [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayPacketAttachment.class.getCanonicalName());
}
}
try {
var protocolIdField = packetClazz.getDeclaredField(ProtocolManager.PROTOCOL_ID);
ReflectionUtils.makeAccessible(protocolIdField);
var protocolId = (short) protocolIdField.get(null);
var receiverDefinition = new PacketReceiverDefinition(bean, method, packetClazz, attachmentClazz);
var enhanceReceiverDefinition = EnhanceUtils.createPacketReceiver(receiverDefinition);
packetReceiverList[protocolId] = enhanceReceiverDefinition;
} catch (NoSuchFieldException | IllegalAccessException | InstantiationException | InvocationTargetException | NoSuchMethodException | CannotCompileException | NotFoundException e) {
throw new RuntimeException(e);
}
}
}
}
net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java
+2 -94
View File
@@ -14,50 +14,36 @@
package com.zfoo.net.dispatcher.manager;
import com.zfoo.event.manager.EventBus;
import com.zfoo.event.model.event.IEvent;
import com.zfoo.net.NetContext;
import com.zfoo.net.core.gateway.model.AuthUidToGatewayCheck;
import com.zfoo.net.core.gateway.model.AuthUidToGatewayConfirm;
import com.zfoo.net.core.gateway.model.AuthUidToGatewayEvent;
import com.zfoo.net.dispatcher.model.anno.PacketReceiver;
import com.zfoo.net.dispatcher.model.answer.AsyncAnswer;
import com.zfoo.net.dispatcher.model.answer.SyncAnswer;
import com.zfoo.net.dispatcher.model.exception.ErrorResponseException;
import com.zfoo.net.dispatcher.model.exception.NetTimeOutException;
import com.zfoo.net.dispatcher.model.exception.UnexpectedProtocolException;
import com.zfoo.net.dispatcher.model.vo.EnhanceUtils;
import com.zfoo.net.dispatcher.model.vo.IPacketReceiver;
import com.zfoo.net.dispatcher.model.vo.PacketReceiverDefinition;
import com.zfoo.net.packet.common.Error;
import com.zfoo.net.packet.common.Heartbeat;
import com.zfoo.net.packet.model.EncodedPacketInfo;
import com.zfoo.net.packet.model.GatewayPacketAttachment;
import com.zfoo.net.packet.model.IPacketAttachment;
import com.zfoo.net.packet.model.SignalPacketAttachment;
import com.zfoo.net.packet.service.PacketService;
import com.zfoo.net.session.model.AttributeType;
import com.zfoo.net.session.model.Session;
import com.zfoo.net.task.TaskManager;
import com.zfoo.net.task.model.ReceiveTask;
import com.zfoo.protocol.IPacket;
import com.zfoo.protocol.ProtocolManager;
import com.zfoo.protocol.collection.CollectionUtils;
import com.zfoo.protocol.exception.ExceptionUtils;
import com.zfoo.protocol.util.AssertionUtils;
import com.zfoo.protocol.util.JsonUtils;
import com.zfoo.protocol.util.ReflectionUtils;
import com.zfoo.protocol.util.StringUtils;
import com.zfoo.util.math.HashUtils;
import com.zfoo.util.math.RandomUtils;
import io.netty.util.concurrent.FastThreadLocal;
import javassist.CannotCompileException;
import javassist.NotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.lang.Nullable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -73,11 +59,6 @@ public class PacketDispatcher implements IPacketDispatcher {
public static final long DEFAULT_TIMEOUT = 3000;
/**
* 客户端和服务端都有接受packet的方法，packetReceiverList对应的就是包的接收方法
*/
private final IPacketReceiver[] packetReceiverList = new IPacketReceiver[ProtocolManager.MAX_PROTOCOL_NUM];
/**
* 会把receive收到的attachment存储在这个地方，只针对task线程。
* doWithReceivePacket会设置receivePacketAttachment，但是在方法调用完成会取消，不需要过多关注。
@@ -307,14 +288,8 @@ public class PacketDispatcher implements IPacketDispatcher {
* 接收者同时只能处理一个session的一个包，同一个发送者发送过来的包排队处理
*/
@Override
public void doReceive(Session session, IPacket packet, IPacketAttachment packetAttachment) {
public void atReceiver(Session session, IPacket packet, IPacketAttachment packetAttachment) {
try {
var packetReceiver = packetReceiverList[packet.protocolId()];
if (packetReceiver == null) {
throw new RuntimeException(StringUtils.format("no any packetReceiverDefinition found for this [packet:{}]", packet.getClass().getName()));
}
// 接收者（服务器）同步和异步消息的接收
if (packetAttachment != null) {
switch (packetAttachment.packetType()) {
@@ -327,8 +302,7 @@ public class PacketDispatcher implements IPacketDispatcher {
}
// 调用PacketReceiver
packetReceiver.invoke(session, packet, packetAttachment);
PacketBus.submit(session, packet, packetAttachment);
} catch (Exception e) {
logger.error(StringUtils.format("e[{}][{}]未知exception异常[e:{}]", session.getAttribute(AttributeType.UID), session.getSid(), e.getMessage()), e);
} catch (Throwable t) {
@@ -347,70 +321,4 @@ public class PacketDispatcher implements IPacketDispatcher {
}
}
@Override
public void registerPacketReceiverDefinition(Object bean) {
var clazz = bean.getClass();
var methods = ReflectionUtils.getMethodsByAnnoInPOJOClass(clazz, PacketReceiver.class);
if (CollectionUtils.isNotEmpty(methods) && !ReflectionUtils.isPojoClass(clazz)) {
logger.warn("消息注册类不是POJO类，父类的不会被扫描到");
}
for (var method : methods) {
var paramClazzs = method.getParameterTypes();
AssertionUtils.isTrue(paramClazzs.length == 2 || paramClazzs.length == 3
, "[class:{}] [method:{}] must have two or three parameter!", bean.getClass().getName(), method.getName());
AssertionUtils.isTrue(Session.class.isAssignableFrom(paramClazzs[0])
, "[class:{}] [method:{}],the first parameter must be Session type parameter Exception.", bean.getClass().getName(), method.getName());
AssertionUtils.isTrue(IPacket.class.isAssignableFrom(paramClazzs[1])
, "[class:{}] [method:{}],the second parameter must be IPacket type parameter Exception.", bean.getClass().getName(), method.getName());
AssertionUtils.isTrue(paramClazzs.length != 3 || IPacketAttachment.class.isAssignableFrom(paramClazzs[2])
, "[class:{}] [method:{}],the third parameter must be IPacketAttachment type parameter Exception.", bean.getClass().getName(), method.getName());
var packetClazz = (Class<? extends IEvent>) paramClazzs[1];
var attachmentClazz = paramClazzs.length == 3 ? paramClazzs[2] : null;
var packetName = packetClazz.getCanonicalName();
var methodName = method.getName();
AssertionUtils.isTrue(Modifier.isPublic(method.getModifiers())
, "[class:{}] [method:{}] [packet:{}] must use 'public' as modifier!", bean.getClass().getName(), methodName, packetName);
AssertionUtils.isTrue(!Modifier.isStatic(method.getModifiers())
, "[class:{}] [method:{}] [packet:{}] can not use 'static' as modifier!", bean.getClass().getName(), methodName, packetName);
var expectedMethodName = StringUtils.format("at{}", packetClazz.getSimpleName());
AssertionUtils.isTrue(methodName.equals(expectedMethodName)
, "[class:{}] [method:{}] [packet:{}] expects '{}' as method name!", bean.getClass().getName(), methodName, packetName, expectedMethodName);
// 如果以Request结尾的请求，那么attachment应该为GatewayAttachment
// 如果以Ask结尾的请求，那么attachment不能为GatewayAttachment
if (attachmentClazz != null) {
if (packetName.endsWith(PacketService.NET_REQUEST_SUFFIX)) {
AssertionUtils.isTrue(attachmentClazz.equals(GatewayPacketAttachment.class)
, "[class:{}] [method:{}] [packet:{}] must use [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayPacketAttachment.class.getCanonicalName());
} else if (packetName.endsWith(PacketService.NET_ASK_SUFFIX)) {
AssertionUtils.isTrue(!attachmentClazz.equals(GatewayPacketAttachment.class)
, "[class:{}] [method:{}] [packet:{}] can not match with [attachment:{}]!", bean.getClass().getName(), methodName, packetName, GatewayPacketAttachment.class.getCanonicalName());
}
}
try {
var protocolIdField = packetClazz.getDeclaredField(ProtocolManager.PROTOCOL_ID);
ReflectionUtils.makeAccessible(protocolIdField);
var protocolId = (short) protocolIdField.get(null);
var receiverDefinition = new PacketReceiverDefinition(bean, method, packetClazz, attachmentClazz);
var enhanceReceiverDefinition = EnhanceUtils.createPacketReceiver(receiverDefinition);
packetReceiverList[protocolId] = enhanceReceiverDefinition;
} catch (NoSuchFieldException | IllegalAccessException | InstantiationException | InvocationTargetException | NoSuchMethodException | CannotCompileException | NotFoundException e) {
throw new RuntimeException(e);
}
}
}
}
net/src/main/java/com/zfoo/net/schema/NetProcessor.java
+2 -5
View File
@@ -13,7 +13,7 @@
package com.zfoo.net.schema;
import com.zfoo.net.NetContext;
import com.zfoo.net.dispatcher.manager.PacketBus;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
@@ -25,10 +25,7 @@ public class NetProcessor implements BeanPostProcessor {
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (NetContext.getNetContext() == null) {
return bean;
}
NetContext.getDispatcher().registerPacketReceiverDefinition(bean);
PacketBus.registerPacketReceiverDefinition(bean);
return bean;
}
net/src/main/java/com/zfoo/net/task/model/ReceiveTask.java
+1 -1
View File
@@ -36,7 +36,7 @@ public final class ReceiveTask implements Runnable {
@Override
public void run() {
NetContext.getDispatcher().doReceive(session, packet, packetAttachment);
NetContext.getDispatcher().atReceiver(session, packet, packetAttachment);
}
public Session getSession() {