From ea911f130bdde294c2b73b4a72a59b2d99117a0b Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sun, 8 Aug 2021 11:25:37 +0800 Subject: [PATCH 1/4] =?UTF-8?q?perf[net]:=20=E4=BC=98=E5=8C=96=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E6=88=96=E5=BC=82=E6=AD=A5=E7=9A=84=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E6=8E=A7=E5=88=B6=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../net/dispatcher/manager/PacketSignal.java | 8 ++-- .../zfoo/net/dispatcher/PacketSignalTest.java | 40 +++++++------------ 2 files changed, 19 insertions(+), 29 deletions(-) diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java index d1e840e5..e6917449 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java @@ -51,12 +51,14 @@ public class PacketSignal { } public static SignalPacketAttachment removeSignalAttachment(SignalPacketAttachment packetAttachment) { - var packetId = packetAttachment.getPacketId(); + return removeSignalAttachment(packetAttachment.getPacketId()); + } + + public static SignalPacketAttachment removeSignalAttachment(int packetId) { var hash = packetId & SIGNAL_MASK; var attachment = signalPacketArray.get(hash); - if (attachment != null && attachment.getPacketId() == packetId) { - signalPacketArray.lazySet(hash, null); + if (attachment != null && attachment.getPacketId() == packetId && signalPacketArray.compareAndSet(hash, attachment, null)) { return attachment; } return signalPacketAttachmentMap.remove(packetId); diff --git a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java index 90d84da3..cc1e7164 100644 --- a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java +++ b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java @@ -16,7 +16,6 @@ import com.zfoo.event.manager.EventBus; import com.zfoo.net.dispatcher.manager.PacketSignal; import com.zfoo.net.packet.model.SignalPacketAttachment; import com.zfoo.scheduler.util.TimeUtils; -import com.zfoo.util.ThreadUtils; import org.junit.Ignore; import org.junit.Test; @@ -50,15 +49,6 @@ public class PacketSignalTest { }); } - for (int i = 0; i < 100; i++) { - new Thread(new Runnable() { - @Override - public void run() { - addAndRemoveWithSleep(); - } - }).start(); - } - countDownLatch.await(); System.out.println(TimeUtils.currentTimeMillis() - startTime); @@ -66,24 +56,22 @@ public class PacketSignalTest { } public void addAndRemove() { - for (int i = 0; i < 1_0000_0000; i++) { - var id = atomicInteger.incrementAndGet(); - var attachment = new SignalPacketAttachment(); - attachment.setPacketId(id); - PacketSignal.addSignalAttachment(attachment); - PacketSignal.removeSignalAttachment(attachment); + for (int count = 0; count < 1_0000; count++) { + var startIndex = atomicInteger.incrementAndGet(); + var endIndex = 0; + for (int i = 0; i < 1000; i++) { + endIndex = atomicInteger.incrementAndGet(); + } + for (int i = startIndex; i < endIndex; i++) { + var attachment = new SignalPacketAttachment(); + attachment.setPacketId(i); + PacketSignal.addSignalAttachment(attachment); + } + for (int i = startIndex; i < endIndex; i++) { + PacketSignal.removeSignalAttachment(i); + } } } - public void addAndRemoveWithSleep() { - for (int i = 0; i < 1000_0000; i++) { - var id = atomicInteger.incrementAndGet(); - var attachment = new SignalPacketAttachment(); - attachment.setPacketId(id); - PacketSignal.addSignalAttachment(attachment); - ThreadUtils.sleep(3000); - PacketSignal.removeSignalAttachment(attachment); - } - } } From 702bfe5fec72f09490766e4dedebb64f2d5c3769 Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sun, 8 Aug 2021 14:54:55 +0800 Subject: [PATCH 2/4] =?UTF-8?q?perf[net]:=20=E4=BC=98=E5=8C=96=E5=90=8C?= =?UTF-8?q?=E6=AD=A5=E6=88=96=E5=BC=82=E6=AD=A5=E7=9A=84=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E6=8E=A7=E5=88=B6=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../net/dispatcher/manager/PacketSignal.java | 10 ++- .../net/dispatcher/PacketSignalArray.java | 79 +++++++++++++++++ .../zfoo/net/dispatcher/PacketSignalMap.java | 49 +++++++++++ .../zfoo/net/dispatcher/PacketSignalTest.java | 86 ++++++++++++++----- 4 files changed, 199 insertions(+), 25 deletions(-) create mode 100644 net/src/test/java/com/zfoo/net/dispatcher/PacketSignalArray.java create mode 100644 net/src/test/java/com/zfoo/net/dispatcher/PacketSignalMap.java diff --git a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java index e6917449..c263a778 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java @@ -31,8 +31,8 @@ public class PacketSignal { private static final Logger logger = LoggerFactory.getLogger(PacketSignal.class); - // equal with 16383 - private static final int SIGNAL_MASK = 0B00000000_00000000_00111111_11111111; + // equal with 32767 + private static final int SIGNAL_MASK = 0B00000000_00000000_01111111_11111111; private static AtomicReferenceArray signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); @@ -45,9 +45,11 @@ public class PacketSignal { var packetId = packetAttachment.getPacketId(); var hash = packetId & SIGNAL_MASK; - if (!signalPacketArray.compareAndSet(hash, null, packetAttachment)) { - signalPacketAttachmentMap.put(packetId, packetAttachment); + if (signalPacketArray.compareAndSet(hash, null, packetAttachment)) { + return; } + + signalPacketAttachmentMap.put(packetId, packetAttachment); } public static SignalPacketAttachment removeSignalAttachment(SignalPacketAttachment packetAttachment) { diff --git a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalArray.java b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalArray.java new file mode 100644 index 00000000..8742bede --- /dev/null +++ b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalArray.java @@ -0,0 +1,79 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.dispatcher; + +import com.zfoo.protocol.util.JsonUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReferenceArray; + +/** + * 同步或异步的调用控制器 + * + * @author jaysunxiao + * @version 3.0 + */ +public class PacketSignalArray { + + private static final Logger logger = LoggerFactory.getLogger(PacketSignalArray.class); + + // equal with 16383 + private static final int SIGNAL_MASK = 0B00000000_00000000_01111111_11111111; + + private static AtomicReferenceArray signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); + + /** + * Session控制同步或异步的附加包，key：packetId + */ + private static Map signalPacketAttachmentMap = new ConcurrentHashMap<>(1000); + + public static void addSignalAttachment(int packetId) { + var hash = packetId & SIGNAL_MASK; + + if (signalPacketArray.compareAndSet(hash, null, packetId)) { + return; + } +// logger.info("add [packetId:{}] [oldPacketId:{}]", packetId, signalPacketArray.get(hash)); + signalPacketAttachmentMap.put(packetId, packetId); + } + + + public static void removeSignalAttachment(int packetId) { + var hash = packetId & SIGNAL_MASK; + var oldPacketId = signalPacketArray.get(hash); + + if (oldPacketId != null && oldPacketId == packetId && signalPacketArray.compareAndSet(hash, oldPacketId, null)) { + return; + } +// logger.info("remove [packetId:{}] [oldPacketId:{}]", packetId, oldPacketId); + signalPacketAttachmentMap.remove(packetId); + } + + public static void status() { + var count = 0; + for (int i = 0; i < SIGNAL_MASK + 1; i++) { + var value = signalPacketArray.get(i); + if (value != null) { + logger.info("signalPacketArray has attachment [index:{}][count:{}][value:{}]", i, ++count, JsonUtils.object2String(value)); + } + } + + signalPacketAttachmentMap.forEach((key, value) -> { + logger.info("signalPacketAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); + }); + } + +} diff --git a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalMap.java b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalMap.java new file mode 100644 index 00000000..b59c7e7a --- /dev/null +++ b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalMap.java @@ -0,0 +1,49 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.dispatcher; + +import com.zfoo.protocol.util.JsonUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * 同步或异步的调用控制器 + * + * @author jaysunxiao + * @version 3.0 + */ +public class PacketSignalMap { + + private static final Logger logger = LoggerFactory.getLogger(PacketSignalMap.class); + + private static Map signalPacketAttachmentMap = new ConcurrentHashMap<>(1_0000); + + public static void addSignalAttachment(int packetId) { + signalPacketAttachmentMap.put(packetId, packetId); + } + + + public static void removeSignalAttachment(int packetId) { + signalPacketAttachmentMap.remove(packetId); + } + + public static void status() { + signalPacketAttachmentMap.forEach((key, value) -> { + logger.info("signalPacketAttachmentMap has attachment [key:{}][value:{}]", key, JsonUtils.object2String(value)); + }); + } + +} diff --git a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java index cc1e7164..366dd90e 100644 --- a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java +++ b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java @@ -14,11 +14,12 @@ package com.zfoo.net.dispatcher; import com.zfoo.event.manager.EventBus; import com.zfoo.net.dispatcher.manager.PacketSignal; -import com.zfoo.net.packet.model.SignalPacketAttachment; import com.zfoo.scheduler.util.TimeUtils; +import com.zfoo.util.ThreadUtils; import org.junit.Ignore; import org.junit.Test; +import java.util.ArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; @@ -31,47 +32,90 @@ public class PacketSignalTest { private AtomicInteger atomicInteger = new AtomicInteger(0); + private int executorSize = EventBus.EXECUTORS_SIZE / 2; + private int count = 100_0000; + private int totalIndex = 10; + @Test public void test() throws InterruptedException { // 预热 - addAndRemove(); + arrayTest(); + mapTest(); + ThreadUtils.sleep(3000); + + arrayTest(); + mapTest(); + System.out.println(atomicInteger.get()); + } + + public void arrayTest() throws InterruptedException { var startTime = TimeUtils.currentTimeMillis(); - var countDownLatch = new CountDownLatch(EventBus.EXECUTORS_SIZE); - for (int i = 0; i < EventBus.EXECUTORS_SIZE; i++) { + var countDownLatch = new CountDownLatch(executorSize); + for (var i = 0; i < executorSize; i++) { EventBus.asyncExecute(i).execute(new Runnable() { @Override public void run() { - addAndRemove(); + addAndRemoveArray(); countDownLatch.countDown(); } }); } - countDownLatch.await(); - - System.out.println(TimeUtils.currentTimeMillis() - startTime); PacketSignal.status(); + System.out.println(TimeUtils.currentTimeMillis() - startTime); } - public void addAndRemove() { - for (int count = 0; count < 1_0000; count++) { - var startIndex = atomicInteger.incrementAndGet(); - var endIndex = 0; - for (int i = 0; i < 1000; i++) { - endIndex = atomicInteger.incrementAndGet(); + public void mapTest() throws InterruptedException { + var startTime = TimeUtils.currentTimeMillis(); + + var countDownLatch = new CountDownLatch(executorSize); + for (int i = 0; i < executorSize; i++) { + EventBus.asyncExecute(i).execute(new Runnable() { + @Override + public void run() { + addAndRemoveMap(); + countDownLatch.countDown(); + } + }); + } + countDownLatch.await(); + PacketSignalMap.status(); + System.out.println(TimeUtils.currentTimeMillis() - startTime); + } + + public void addAndRemoveArray() { + var list = new ArrayList(totalIndex); + for (var i = 0; i < count; i++) { + list.clear(); + for (var j = 0; j < totalIndex; j++) { + var index = atomicInteger.incrementAndGet(); + list.add(index); + PacketSignalArray.addSignalAttachment(index); } - for (int i = startIndex; i < endIndex; i++) { - var attachment = new SignalPacketAttachment(); - attachment.setPacketId(i); - PacketSignal.addSignalAttachment(attachment); - } - for (int i = startIndex; i < endIndex; i++) { - PacketSignal.removeSignalAttachment(i); + + for (var index : list) { + PacketSignalArray.removeSignalAttachment(index); } } } + public void addAndRemoveMap() { + var list = new ArrayList(totalIndex); + for (var i = 0; i < count; i++) { + list.clear(); + for (var j = 0; j < totalIndex; j++) { + var index = atomicInteger.incrementAndGet(); + list.add(index); + PacketSignalMap.addSignalAttachment(index); + } + + for (var index : list) { + PacketSignalMap.removeSignalAttachment(index); + } + } + } + } From e7067741a7de5386a959dfe7cfca7aa502e2e76f Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sun, 8 Aug 2021 16:00:32 +0800 Subject: [PATCH 3/4] =?UTF-8?q?perf[net]:=20=E4=BC=98=E5=8C=96=E5=8D=8F?= =?UTF-8?q?=E8=AE=AE=E6=8E=A5=E6=94=B6=E5=99=A8=EF=BC=8C=E5=85=B1=E7=94=A8?= =?UTF-8?q?ProtocolManager=E6=9D=A5=E5=87=8F=E5=B0=91=E5=88=9D=E5=A7=8B?= =?UTF-8?q?=E5=8C=96=E7=9A=84=E7=A9=BA=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../net/dispatcher/manager/PacketBus.java | 15 ++++++-- .../net/packet/service/PacketService.java | 8 ++++ .../zfoo/net/schema/NetDefinitionParser.java | 5 --- .../com/zfoo/net/schema/NetProcessor.java | 32 ---------------- .../protocol/registration/EnhanceUtils.java | 11 +++++- .../registration/IProtocolRegistration.java | 13 ++++++- .../registration/ProtocolRegistration.java | 37 +++++++++++-------- 7 files changed, 63 insertions(+), 58 deletions(-) delete mode 100644 net/src/main/java/com/zfoo/net/schema/NetProcessor.java 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 index dfb59197..d54dc0ae 100644 --- a/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java +++ b/net/src/main/java/com/zfoo/net/dispatcher/manager/PacketBus.java @@ -26,6 +26,7 @@ import com.zfoo.net.session.model.Session; import com.zfoo.protocol.IPacket; import com.zfoo.protocol.ProtocolManager; import com.zfoo.protocol.collection.ArrayUtils; +import com.zfoo.protocol.registration.IProtocolRegistration; import com.zfoo.protocol.util.AssertionUtils; import com.zfoo.protocol.util.ReflectionUtils; import com.zfoo.protocol.util.StringUtils; @@ -43,9 +44,10 @@ public abstract class PacketBus { private static final Logger logger = LoggerFactory.getLogger(PacketBus.class); /** - * 客户端和服务端都有接受packet的方法，packetReceiverList对应的就是包的接收方法 + * 客户端和服务端都有接受packet的方法，packetReceiverList对应的就是包的接收方法，将receiver注册到IProtocolRegistration */ - private static final IPacketReceiver[] packetReceiverList = new IPacketReceiver[ProtocolManager.MAX_PROTOCOL_NUM]; + public static final IProtocolRegistration[] packetReceiverList = ProtocolManager.protocols; + /** * 正常消息的接收 @@ -54,7 +56,7 @@ public abstract class PacketBus { * 接收者同时只能处理一个session的一个包，同一个发送者发送过来的包排队处理 */ public static void submit(Session session, IPacket packet, IPacketAttachment packetAttachment) { - var packetReceiver = packetReceiverList[packet.protocolId()]; + var packetReceiver = (IPacketReceiver) packetReceiverList[packet.protocolId()].receiver(); if (packetReceiver == null) { throw new RuntimeException(StringUtils.format("no any packetReceiverDefinition found for this [packet:{}]", packet.getClass().getName())); } @@ -124,7 +126,12 @@ public abstract class PacketBus { var protocolId = (short) protocolIdField.get(null); var receiverDefinition = new PacketReceiverDefinition(bean, method, packetClazz, attachmentClazz); var enhanceReceiverDefinition = EnhanceUtils.createPacketReceiver(receiverDefinition); - packetReceiverList[protocolId] = enhanceReceiverDefinition; + + // 将receiver注册到IProtocolRegistration + var protocolRegistration = packetReceiverList[protocolId]; + var receiverField = ReflectionUtils.getFieldByNameInPOJOClass(protocolRegistration.getClass(), "receiver"); + ReflectionUtils.makeAccessible(receiverField); + ReflectionUtils.setField(receiverField, protocolRegistration, enhanceReceiverDefinition); } catch (Throwable t) { throw new RuntimeException(t); } diff --git a/net/src/main/java/com/zfoo/net/packet/service/PacketService.java b/net/src/main/java/com/zfoo/net/packet/service/PacketService.java index 267e12e3..c67bba15 100644 --- a/net/src/main/java/com/zfoo/net/packet/service/PacketService.java +++ b/net/src/main/java/com/zfoo/net/packet/service/PacketService.java @@ -14,6 +14,7 @@ package com.zfoo.net.packet.service; import com.zfoo.net.NetContext; +import com.zfoo.net.dispatcher.manager.PacketBus; import com.zfoo.net.packet.model.DecodedPacketInfo; import com.zfoo.net.packet.model.IPacketAttachment; import com.zfoo.protocol.IPacket; @@ -104,6 +105,13 @@ public class PacketService implements IPacketService { logger.error(ExceptionUtils.getMessage(e)); throw new RuntimeException(e); } + + // 注册协议接收器 + var beanNames = applicationContext.getBeanDefinitionNames(); + for (var beanName : beanNames) { + var bean = applicationContext.getBean(beanName); + PacketBus.registerPacketReceiverDefinition(bean); + } } @Override diff --git a/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java b/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java index 2dfe5f3c..b4928d51 100644 --- a/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java +++ b/net/src/main/java/com/zfoo/net/schema/NetDefinitionParser.java @@ -60,11 +60,6 @@ public class NetDefinitionParser implements BeanDefinitionParser { builder.addPropertyReference("localConfig", NetConfig.class.getCanonicalName()); parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition()); - // 注册NetProcessor - clazz = NetProcessor.class; - builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); - parserContext.getRegistry().registerBeanDefinition(clazz.getCanonicalName(), builder.getBeanDefinition()); - // 注册ProtocolManager clazz = ProtocolManager.class; builder = BeanDefinitionBuilder.rootBeanDefinition(clazz); diff --git a/net/src/main/java/com/zfoo/net/schema/NetProcessor.java b/net/src/main/java/com/zfoo/net/schema/NetProcessor.java deleted file mode 100644 index fc73bb7a..00000000 --- a/net/src/main/java/com/zfoo/net/schema/NetProcessor.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * Copyright (C) 2020 The zfoo Authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except - * in compliance with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed - * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and limitations under the License. - */ - -package com.zfoo.net.schema; - -import com.zfoo.net.dispatcher.manager.PacketBus; -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.config.BeanPostProcessor; - -/** - * @author jaysunxiao - * @version 3.0 - */ -public class NetProcessor implements BeanPostProcessor { - - @Override - public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - PacketBus.registerPacketReceiverDefinition(bean); - return bean; - } - -} diff --git a/protocol/src/main/java/com/zfoo/protocol/registration/EnhanceUtils.java b/protocol/src/main/java/com/zfoo/protocol/registration/EnhanceUtils.java index 4cca32fb..b50d82c4 100644 --- a/protocol/src/main/java/com/zfoo/protocol/registration/EnhanceUtils.java +++ b/protocol/src/main/java/com/zfoo/protocol/registration/EnhanceUtils.java @@ -134,6 +134,10 @@ public abstract class EnhanceUtils { constructorFiled.setModifiers(Modifier.PRIVATE); enhanceClazz.addField(constructorFiled); + CtField receiverFiled = new CtField(classPool.get(Object.class.getCanonicalName()), "receiver", enhanceClazz); + receiverFiled.setModifiers(Modifier.PRIVATE); + enhanceClazz.addField(receiverFiled); + // 定义类所包含的所有子协议成员 var allSubProtocolIds = ProtocolAnalysis.getAllSubProtocolIds(protocolId) .stream() @@ -142,7 +146,7 @@ public abstract class EnhanceUtils { for (var subProtocolId : allSubProtocolIds) { var protocolRegistrationField = new CtField(classPool.get(IProtocolRegistration.class.getCanonicalName()), getProtocolRegistrationFieldNameByProtocolId(subProtocolId), enhanceClazz); - constructorFiled.setModifiers(Modifier.PRIVATE); + protocolRegistrationField.setModifiers(Modifier.PRIVATE); enhanceClazz.addField(protocolRegistrationField); } @@ -163,6 +167,11 @@ public abstract class EnhanceUtils { protocolConstructorMethod.setBody("{return this.constructor;}"); enhanceClazz.addMethod(protocolConstructorMethod); + CtMethod receiverMethod = new CtMethod(classPool.get(Object.class.getCanonicalName()), "receiver", null, enhanceClazz); + receiverMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL); + receiverMethod.setBody("{return this.receiver;}"); + enhanceClazz.addMethod(receiverMethod); + CtMethod moduleMethod = new CtMethod(classPool.get(byte.class.getCanonicalName()), "module", null, enhanceClazz); moduleMethod.setModifiers(Modifier.PUBLIC + Modifier.FINAL); moduleMethod.setBody("{return " + registration.module() + ";}"); diff --git a/protocol/src/main/java/com/zfoo/protocol/registration/IProtocolRegistration.java b/protocol/src/main/java/com/zfoo/protocol/registration/IProtocolRegistration.java index c455ef25..7f0fae90 100644 --- a/protocol/src/main/java/com/zfoo/protocol/registration/IProtocolRegistration.java +++ b/protocol/src/main/java/com/zfoo/protocol/registration/IProtocolRegistration.java @@ -30,8 +30,19 @@ public interface IProtocolRegistration { Constructor protocolConstructor(); - Object read(ByteBuf buffer); + /** + * 协议接收器，回调方法，主要是存放一些额外的参数 + */ + Object receiver(); + /** + * 序列化 + */ void write(ByteBuf buffer, IPacket packet); + /** + * 反序列化 + */ + Object read(ByteBuf buffer); + } diff --git a/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolRegistration.java b/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolRegistration.java index 36bc9544..dbec0db5 100644 --- a/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolRegistration.java +++ b/protocol/src/main/java/com/zfoo/protocol/registration/ProtocolRegistration.java @@ -37,6 +37,8 @@ public class ProtocolRegistration implements IProtocolRegistration { private byte module; private Constructor constructor; + private Object receiver; + private Field[] fields; @@ -64,24 +66,12 @@ public class ProtocolRegistration implements IProtocolRegistration { return constructor; } - @Override - public Object read(ByteBuf buffer) { - if (!ByteBufUtils.readBoolean(buffer)) { - return null; - } - Object object = ReflectionUtils.newInstance(constructor); - - for (int i = 0, length = fields.length; i < length; i++) { - Field field = fields[i]; - IFieldRegistration packetFieldRegistration = fieldRegistrations[i]; - ISerializer serializer = packetFieldRegistration.serializer(); - Object fieldValue = serializer.readObject(buffer, packetFieldRegistration); - ReflectionUtils.setField(field, object, fieldValue); - } - return object; + public Object receiver() { + return receiver; } + @Override public void write(ByteBuf buffer, IPacket packet) { if (packet == null) { @@ -100,6 +90,23 @@ public class ProtocolRegistration implements IProtocolRegistration { } } + @Override + public Object read(ByteBuf buffer) { + if (!ByteBufUtils.readBoolean(buffer)) { + return null; + } + Object object = ReflectionUtils.newInstance(constructor); + + for (int i = 0, length = fields.length; i < length; i++) { + Field field = fields[i]; + IFieldRegistration packetFieldRegistration = fieldRegistrations[i]; + ISerializer serializer = packetFieldRegistration.serializer(); + Object fieldValue = serializer.readObject(buffer, packetFieldRegistration); + ReflectionUtils.setField(field, object, fieldValue); + } + return object; + } + public short getId() { return id; From 805a1aa3fb86c7829e43ae2a70466e7e52293ca6 Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sun, 8 Aug 2021 18:00:27 +0800 Subject: [PATCH 4/4] =?UTF-8?q?perf[net]:=20=E6=94=AF=E6=8C=81http?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/zfoo/net/core/AbstractServer.java | 2 +- .../com/zfoo/net/core/http/HttpServer.java | 61 ++++++++++ .../java/com/zfoo/net/core/udp/UdpServer.java | 6 + .../net/core/websocket/WebsocketServer.java | 14 +-- .../dispatcher/manager/PacketDispatcher.java | 3 - .../handler/codec/http/HttpCodecHandler.java | 115 ++++++++++++++++++ .../packet/model/HttpPacketAttachment.java | 47 +++++++ .../net/packet/model/NoAnswerAttachment.java | 2 +- .../packet/model/PacketAttachmentType.java | 21 ++-- net/src/test/resources/protocol.xml | 3 +- 10 files changed, 250 insertions(+), 24 deletions(-) create mode 100644 net/src/main/java/com/zfoo/net/core/http/HttpServer.java create mode 100644 net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java create mode 100644 net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java diff --git a/net/src/main/java/com/zfoo/net/core/AbstractServer.java b/net/src/main/java/com/zfoo/net/core/AbstractServer.java index dd004724..35f894f7 100644 --- a/net/src/main/java/com/zfoo/net/core/AbstractServer.java +++ b/net/src/main/java/com/zfoo/net/core/AbstractServer.java @@ -94,7 +94,7 @@ public abstract class AbstractServer implements IServer { allServers.add(this); - logger.info("TcpServer started at [{}:{}]", hostAddress, port); + logger.info("{} started at [{}:{}]", this.getClass().getSimpleName(), hostAddress, port); } diff --git a/net/src/main/java/com/zfoo/net/core/http/HttpServer.java b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java new file mode 100644 index 00000000..36eeb442 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/core/http/HttpServer.java @@ -0,0 +1,61 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + * + */ + +package com.zfoo.net.core.http; + +import com.zfoo.net.core.AbstractServer; +import com.zfoo.net.handler.ServerDispatcherHandler; +import com.zfoo.net.handler.codec.http.HttpCodecHandler; +import com.zfoo.protocol.IPacket; +import com.zfoo.util.net.HostAndPort; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.socket.SocketChannel; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.stream.ChunkedWriteHandler; + +import java.util.function.Function; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class HttpServer extends AbstractServer { + + /** + * http的地址解析器 + */ + private Function uriResolver; + + public HttpServer(HostAndPort host, Function uriResolver) { + super(host); + this.uriResolver = uriResolver; + } + + @Override + public ChannelInitializer channelChannelInitializer() { + return new ChannelHandlerInitializer(); + } + + + private class ChannelHandlerInitializer extends ChannelInitializer { + @Override + protected void initChannel(SocketChannel channel) { + channel.pipeline().addLast(new HttpServerCodec()); + channel.pipeline().addLast(new ChunkedWriteHandler()); + channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024)); + channel.pipeline().addLast(new HttpCodecHandler(uriResolver)); + channel.pipeline().addLast(new ServerDispatcherHandler()); + } + } +} diff --git a/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java b/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java index 4b0a2506..52c105b7 100644 --- a/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java +++ b/net/src/main/java/com/zfoo/net/core/udp/UdpServer.java @@ -26,6 +26,8 @@ import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.util.concurrent.DefaultThreadFactory; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * @author jaysunxiao @@ -33,6 +35,8 @@ import io.netty.util.concurrent.DefaultThreadFactory; */ public class UdpServer extends AbstractServer { + private static final Logger logger = LoggerFactory.getLogger(UdpServer.class); + public UdpServer(HostAndPort host) { super(host); } @@ -58,6 +62,8 @@ public class UdpServer extends AbstractServer { channel = channelFuture.channel(); allServers.add(this); + + logger.info("{} started at [{}:{}]", this.getClass().getSimpleName(), hostAddress, port); } @Override diff --git a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java b/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java index ac59c94d..ac9684af 100644 --- a/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java +++ b/net/src/main/java/com/zfoo/net/core/websocket/WebsocketServer.java @@ -18,7 +18,6 @@ import com.zfoo.net.handler.ServerDispatcherHandler; import com.zfoo.net.handler.codec.websocket.WebSocketCodecHandler; import com.zfoo.util.net.HostAndPort; import io.netty.channel.ChannelInitializer; -import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.HttpServerCodec; @@ -45,19 +44,18 @@ public class WebsocketServer extends AbstractServer { @Override public void initChannel(SocketChannel channel) { - ChannelPipeline pipeline = channel.pipeline(); // 编解码 http 请求 - pipeline.addLast(new HttpServerCodec()); + channel.pipeline().addLast(new HttpServerCodec()); // 写文件内容，支持异步发送大的码流，一般用于发送文件流 - pipeline.addLast(new ChunkedWriteHandler()); + channel.pipeline().addLast(new ChunkedWriteHandler()); // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest // 保证接收的 Http 请求的完整性 - pipeline.addLast(new HttpObjectAggregator(64 * 1024)); + channel.pipeline().addLast(new HttpObjectAggregator(64 * 1024)); // 处理其他的 WebSocketFrame - pipeline.addLast(new WebSocketServerProtocolHandler("/websocket")); + channel.pipeline().addLast(new WebSocketServerProtocolHandler("/websocket")); // 编解码WebSocketFrame二进制协议 - pipeline.addLast(new WebSocketCodecHandler()); - pipeline.addLast(new ServerDispatcherHandler()); + channel.pipeline().addLast(new WebSocketCodecHandler()); + channel.pipeline().addLast(new ServerDispatcherHandler()); } } 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 4687ec84..0a842108 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 @@ -129,9 +129,6 @@ public class PacketDispatcher implements IPacketDispatcher { } return; } - break; - case NORMAL_PACKET: - break; default: break; diff --git a/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java new file mode 100644 index 00000000..7d06ca1f --- /dev/null +++ b/net/src/main/java/com/zfoo/net/handler/codec/http/HttpCodecHandler.java @@ -0,0 +1,115 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.handler.codec.http; + +import com.zfoo.net.NetContext; +import com.zfoo.net.packet.model.DecodedPacketInfo; +import com.zfoo.net.packet.model.EncodedPacketInfo; +import com.zfoo.net.packet.model.HttpPacketAttachment; +import com.zfoo.net.packet.service.PacketService; +import com.zfoo.net.util.SessionUtils; +import com.zfoo.protocol.IPacket; +import com.zfoo.protocol.util.JsonUtils; +import com.zfoo.protocol.util.StringUtils; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.handler.codec.MessageToMessageCodec; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpVersion; +import io.netty.util.ReferenceCountUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.function.Function; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class HttpCodecHandler extends MessageToMessageCodec { + + private static final Logger logger = LoggerFactory.getLogger(HttpCodecHandler.class); + + private Function uriResolver; + + public HttpCodecHandler(Function uriResolver) { + super(); + this.uriResolver = uriResolver; + } + + @Override + protected void decode(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest, List list) { + var uri = fullHttpRequest.uri(); + + ByteBuf in = fullHttpRequest.content(); + + // 不够读一个int + if (in.readableBytes() <= PacketService.PACKET_HEAD_LENGTH) { + return; + } + + in.markReaderIndex(); + var length = in.readInt(); + + // 如果长度非法，则抛出异常断开连接 + if (length < 0) { + throw new IllegalArgumentException(StringUtils.format("[session:{}]的包头长度[length:{}]非法" + , SessionUtils.sessionInfo(channelHandlerContext), length)); + } + + // ByteBuf里的数据太小 + if (in.readableBytes() < length) { + in.resetReaderIndex(); + return; + } + + ByteBuf tmpByteBuf = null; + try { + tmpByteBuf = in.readRetainedSlice(length); + DecodedPacketInfo packetInfo = NetContext.getPacketService().read(tmpByteBuf); + + packetInfo.setPacketAttachment(HttpPacketAttachment.valueOf()); + list.add(packetInfo); + } catch (Exception e) { + logger.error("exception异常", e); + throw e; + } catch (Throwable t) { + logger.error("throwable错误", t); + throw t; + } finally { + ReferenceCountUtil.release(tmpByteBuf); + } + } + + @Override + protected void encode(ChannelHandlerContext channelHandlerContext, EncodedPacketInfo out, List list) { + try { + var byteBuf = channelHandlerContext.alloc().ioBuffer(); + var httpPacketAttachment = (HttpPacketAttachment) out.getPacketAttachment(); + + var fullHttpResponse = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer("I am ok".getBytes())); + + list.add(fullHttpResponse); + } catch (Exception e) { + logger.error("[{}]编码exception异常", JsonUtils.object2String(out), e); + throw e; + } catch (Throwable t) { + logger.error("[{}]编码throwable错误", JsonUtils.object2String(out), t); + throw t; + } + } +} diff --git a/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java b/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java new file mode 100644 index 00000000..3c6feaa7 --- /dev/null +++ b/net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java @@ -0,0 +1,47 @@ +/* + * Copyright (C) 2020 The zfoo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except + * in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed + * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and limitations under the License. + */ + +package com.zfoo.net.packet.model; + +import com.zfoo.util.math.RandomUtils; + +/** + * @author jaysunxiao + * @version 3.0 + */ +public class HttpPacketAttachment implements IPacketAttachment { + + public static final transient short PROTOCOL_ID = 3; + + + public static HttpPacketAttachment valueOf() { + var attachment = new HttpPacketAttachment(); + return attachment; + } + + @Override + public PacketAttachmentType packetType() { + return PacketAttachmentType.HTTP_PACKET; + } + + @Override + public int executorConsistentHash() { + return RandomUtils.randomInt(); + } + + @Override + public short protocolId() { + return PROTOCOL_ID; + } + +} diff --git a/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java b/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java index de118d08..ef2f31fd 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java +++ b/net/src/main/java/com/zfoo/net/packet/model/NoAnswerAttachment.java @@ -21,7 +21,7 @@ package com.zfoo.net.packet.model; */ public class NoAnswerAttachment implements IPacketAttachment { - public static final transient short PROTOCOL_ID = 3; + public static final transient short PROTOCOL_ID = 4; /** * 用来在TaskManage中计算一致性hash的参数 diff --git a/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java b/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java index 95dd66d7..193f5328 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java +++ b/net/src/main/java/com/zfoo/net/packet/model/PacketAttachmentType.java @@ -22,32 +22,33 @@ import java.util.Map; */ public enum PacketAttachmentType { - /** - * 正常的附加包 - */ - NORMAL_PACKET((byte) 0, null), - /** * 带有同步或者异步信息的附加包 */ - SIGNAL_PACKET((byte) 1, SignalPacketAttachment.class), + SIGNAL_PACKET((byte) 0, SignalPacketAttachment.class), /** * 带有网关信息的附加包 */ - GATEWAY_PACKET((byte) 2, GatewayPacketAttachment.class), + GATEWAY_PACKET((byte) 1, GatewayPacketAttachment.class), /** * udp消息的附加包 */ - UDP_PACKET((byte) 3, UdpPacketAttachment.class), + UDP_PACKET((byte) 2, UdpPacketAttachment.class), + + + /** + * http消息的附加包 + */ + HTTP_PACKET((byte) 3, HttpPacketAttachment.class), /** * 无返回消息的附加包 */ NO_ANSWER_PACKET((byte) 4, NoAnswerAttachment.class), - + ; @@ -60,7 +61,7 @@ public enum PacketAttachmentType { } public static PacketAttachmentType getPacketType(byte packetType) { - return map.getOrDefault(packetType, PacketAttachmentType.NORMAL_PACKET); + return map.getOrDefault(packetType, PacketAttachmentType.NO_ANSWER_PACKET); } public byte getPacketType() { diff --git a/net/src/test/resources/protocol.xml b/net/src/test/resources/protocol.xml index b682aa6e..ecd3f484 100644 --- a/net/src/test/resources/protocol.xml +++ b/net/src/test/resources/protocol.xml @@ -7,7 +7,8 @@ - + +