Merge remote-tracking branch 'origin/main'

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);
     }
 
 

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<String, IPacket> uriResolver;
+
+    public HttpServer(HostAndPort host, Function<String, IPacket> uriResolver) {
+        super(host);
+        this.uriResolver = uriResolver;
+    }
+
+    @Override
+    public ChannelInitializer<SocketChannel> channelChannelInitializer() {
+        return new ChannelHandlerInitializer();
+    }
+
+
+    private class ChannelHandlerInitializer extends ChannelInitializer<SocketChannel> {
+        @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());
+        }
+    }
+}

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

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());
         }
     }
 

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);
             }

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;

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<SignalPacketAttachment> signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1);
 
@@ -45,18 +45,22 @@ 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) {
-        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);

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<FullHttpRequest, EncodedPacketInfo> {
+
+    private static final Logger logger = LoggerFactory.getLogger(HttpCodecHandler.class);
+
+    private Function<String, IPacket> uriResolver;
+
+    public HttpCodecHandler(Function<String, IPacket> uriResolver) {
+        super();
+        this.uriResolver = uriResolver;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest, List<Object> 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<Object> 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;
+        }
+    }
+}

net/src/main/java/com/zfoo/net/packet/model/HttpPacketAttachment.java

@@ -11,22 +11,37 @@
  * See the License for the specific language governing permissions and limitations under the License.
  */
 
-package com.zfoo.net.schema;
+package com.zfoo.net.packet.model;
 
-import com.zfoo.net.dispatcher.manager.PacketBus;
-import org.springframework.beans.BeansException;
-import org.springframework.beans.factory.config.BeanPostProcessor;
+import com.zfoo.util.math.RandomUtils;
 
 /**
  * @author jaysunxiao
  * @version 3.0
  */
-public class NetProcessor implements BeanPostProcessor {
+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 Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
-        PacketBus.registerPacketReceiverDefinition(bean);
-        return bean;
+    public PacketAttachmentType packetType() {
+        return PacketAttachmentType.HTTP_PACKET;
+    }
+
+    @Override
+    public int executorConsistentHash() {
+        return RandomUtils.randomInt();
+    }
+
+    @Override
+    public short protocolId() {
+        return PROTOCOL_ID;
     }
 
 }

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的参数

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() {

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

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);

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<Integer> signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1);
+
+    /**
+     * Session控制同步或异步的附加包，key：packetId
+     */
+    private static Map<Integer, Integer> 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));
+        });
+    }
+
+}

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<Integer, Integer> 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));
+        });
+    }
+
+}

net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java

@@ -14,12 +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;
 
@@ -32,57 +32,89 @@ 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();
+        PacketSignal.status();
+        System.out.println(TimeUtils.currentTimeMillis() - startTime);
+    }
 
-        for (int i = 0; i < 100; i++) {
-            new Thread(new Runnable() {
+    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() {
-                    addAndRemoveWithSleep();
+                    addAndRemoveMap();
+                    countDownLatch.countDown();
                 }
-            }).start();
+            });
         }
-
         countDownLatch.await();
-
+        PacketSignalMap.status();
         System.out.println(TimeUtils.currentTimeMillis() - startTime);
-        PacketSignal.status();
     }
 
-    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);
+    public void addAndRemoveArray() {
+        var list = new ArrayList<Integer>(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 (var index : list) {
+                PacketSignalArray.removeSignalAttachment(index);
+            }
         }
     }
 
-    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);
+
+    public void addAndRemoveMap() {
+        var list = new ArrayList<Integer>(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);
+            }
         }
     }
 

net/src/test/resources/protocol.xml

@@ -7,7 +7,8 @@
         <protocol id="0" location="com.zfoo.net.packet.model.SignalPacketAttachment"/>
         <protocol id="1" location="com.zfoo.net.packet.model.GatewayPacketAttachment"/>
         <protocol id="2" location="com.zfoo.net.packet.model.UdpPacketAttachment"/>
-        <protocol id="3" location="com.zfoo.net.packet.model.NoAnswerAttachment"/>
+        <protocol id="3" location="com.zfoo.net.packet.model.HttpPacketAttachment"/>
+        <protocol id="4" location="com.zfoo.net.packet.model.NoAnswerAttachment"/>
 
 
         <protocol id="20" location="com.zfoo.net.core.gateway.model.AuthUidToGatewayCheck"/>

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() + ";}");

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);
+
 }

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;