From 3ff1e7de0aab8e4c314466da9769471438b9075b Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sat, 26 Jun 2021 11:54:16 +0800 Subject: [PATCH] =?UTF-8?q?perf[net]:=20=E4=B8=BA=E4=BA=86=E7=AE=80?= =?UTF-8?q?=E5=8C=96=E4=BB=A3=E7=A0=81=E7=BB=93=E6=9E=84=EF=BC=8C=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E6=8E=A5=E6=94=B6=E6=8F=90=E5=8F=96=E5=87=BAPacketBus?= =?UTF-8?q?=E5=8C=85=E6=8E=A5=E6=94=B6=E6=80=BB=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/java/com/zfoo/net/NetContext.java | 21 +-- .../dispatcher/manager/IPacketDispatcher.java | 4 +- .../net/dispatcher/manager/PacketBus.java | 120 ++++++++++++++++++ .../dispatcher/manager/PacketDispatcher.java | 96 +------------- .../com/zfoo/net/schema/NetProcessor.java | 7 +- .../com/zfoo/net/task/model/ReceiveTask.java | 2 +- 6 files changed, 133 insertions(+), 117 deletions(-) create mode 100644 net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java diff --git a/net/src/main/java/com/zfoo/net/NetContext.java b/net/src/main/java/com/zfoo/net/NetContext.java index 2077b7ce..5476d60e 100644 --- a/net/src/main/java/com/zfoo/net/NetContext.java +++ b/net/src/main/java/com/zfoo/net/NetContext.java @@ -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, 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, } - @Override - public int getOrder() { - return 0; - } - public synchronized static void shutdownBefore() { SchedulerContext.shutdown(); } @@ -158,4 +146,9 @@ public class NetContext implements ApplicationListener, logger.info("Net shutdown gracefully."); } + @Override + public int getOrder() { + return 0; + } + } diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/IPacketDispatcher.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/IPacketDispatcher.java index 47f26bb0..d1bfadbd 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/IPacketDispatcher.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/IPacketDispatcher.java @@ -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 { AsyncAnswer asyncAsk(Session session, IPacket packet, @Nullable Class answerClass, @Nullable Object argument); - void registerPacketReceiverDefinition(Object bean); - } diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java new file mode 100644 index 00000000..2a0b6d0f --- /dev/null +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java @@ -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]; + + /** + * 正常消息的接收 + *

+ * 发送者同时能发送多个包 + * 接收者同时只能处理一个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) 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); + } + } + } + +} diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java index 78f68094..95487824 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java @@ -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) 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); - } - } - } - } diff --git a/net/src/main/java/com/zfoo/net/schema/NetProcessor.java b/net/src/main/java/com/zfoo/net/schema/NetProcessor.java index f82d1b5c..fc73bb7a 100644 --- a/net/src/main/java/com/zfoo/net/schema/NetProcessor.java +++ b/net/src/main/java/com/zfoo/net/schema/NetProcessor.java @@ -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; } diff --git a/net/src/main/java/com/zfoo/net/task/model/ReceiveTask.java b/net/src/main/java/com/zfoo/net/task/model/ReceiveTask.java index 0145d967..ed4a62de 100644 --- a/net/src/main/java/com/zfoo/net/task/model/ReceiveTask.java +++ b/net/src/main/java/com/zfoo/net/task/model/ReceiveTask.java @@ -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() {