perf[net]: 同步或异步的调用控制器优先使用AtomicReferenceArray增强性能，而不是一直使用ConcurrentHashMap

event/src/main/java/com/zfoo/event/manager/EventBus.java

@@ -37,7 +37,7 @@ public abstract class EventBus {
     private static final Logger logger = LoggerFactory.getLogger(EventBus.class);
 
     // 线程池的大小
-    private static final int EXECUTORS_SIZE = Runtime.getRuntime().availableProcessors() * 2;
+    public static final int EXECUTORS_SIZE = Runtime.getRuntime().availableProcessors() * 2;
 
     private static final ExecutorService[] executors = new ExecutorService[EXECUTORS_SIZE];
 

net/src/main/java/com/zfoo/net/consumer/service/Consumer.java

@@ -15,6 +15,7 @@ package com.zfoo.net.consumer.service;
 
 import com.zfoo.net.NetContext;
 import com.zfoo.net.dispatcher.manager.PacketDispatcher;
+import com.zfoo.net.dispatcher.manager.PacketSignal;
 import com.zfoo.net.dispatcher.model.answer.AsyncAnswer;
 import com.zfoo.net.dispatcher.model.answer.SyncAnswer;
 import com.zfoo.net.dispatcher.model.exception.ErrorResponseException;
@@ -71,7 +72,7 @@ public class Consumer implements IConsumer {
         clientAttachment.setExecutorConsistentHash(executorConsistentHash);
 
         try {
-            session.addClientSignalAttachment(clientAttachment);
+            PacketSignal.addSignalAttachment(clientAttachment);
 
             // load balancer之前调用
             loadBalancer.beforeLoadBalancer(session, packet, clientAttachment);
@@ -96,7 +97,7 @@ public class Consumer implements IConsumer {
             throw new NetTimeOutException(StringUtils.format("syncRequest timeout exception, ask:[{}], attachment:[{}]"
                     , JsonUtils.object2String(packet), JsonUtils.object2String(clientAttachment)));
         } finally {
-            session.removeClientSignalAttachment(clientAttachment);
+            PacketSignal.removeSignalAttachment(clientAttachment);
         }
     }
 

net/src/main/java/com/zfoo/net/dispatcher/manager/PacketDispatcher.java

@@ -86,7 +86,7 @@ public class PacketDispatcher implements IPacketDispatcher {
 
                     } else {
                         // 客户端收到服务器应答，客户端发送的时候isClient为true，服务器收到的时候将其设置为false
-                        var attachment = (SignalPacketAttachment) session.removeClientSignalAttachment(signalPacketAttachment);
+                        var attachment = (SignalPacketAttachment) PacketSignal.removeSignalAttachment(signalPacketAttachment);
                         if (attachment != null) {
                             attachment.getResponseFuture().complete(packet);
                         } else {
@@ -185,7 +185,7 @@ public class PacketDispatcher implements IPacketDispatcher {
         clientAttachment.setExecutorConsistentHash(executorConsistentHash);
 
         try {
-            session.addClientSignalAttachment(clientAttachment);
+            PacketSignal.addSignalAttachment(clientAttachment);
             send(session, packet, clientAttachment);
 
             IPacket responsePacket = clientAttachment.getResponseFuture().get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
@@ -203,7 +203,7 @@ public class PacketDispatcher implements IPacketDispatcher {
             throw new NetTimeOutException(StringUtils.format("syncRequest timeout exception, ask:[{}], attachment:[{}]"
                     , JsonUtils.object2String(packet), JsonUtils.object2String(clientAttachment)));
         } finally {
-            session.removeClientSignalAttachment(clientAttachment);
+            PacketSignal.removeSignalAttachment(clientAttachment);
         }
 
     }
@@ -241,7 +241,7 @@ public class PacketDispatcher implements IPacketDispatcher {
                     })
                     .whenCompleteAsync((responsePacket, e) -> {
                         try {
-                            session.removeClientSignalAttachment(clientAttachment);
+                            PacketSignal.removeSignalAttachment(clientAttachment);
 
                             // 如果有异常的话，whenCompleteAsync的下一个thenAccept不会执行
                             if (e != null) {
@@ -269,13 +269,13 @@ public class PacketDispatcher implements IPacketDispatcher {
                     }, TaskManager.getInstance().getExecutorByConsistentHash(executorConsistentHash));
 
 
-            session.addClientSignalAttachment(clientAttachment);
+            PacketSignal.addSignalAttachment(clientAttachment);
 
             // 等到上层调用whenComplete才会发送消息
             asyncAnswer.setAskCallback(() -> send(session, packet, clientAttachment));
             return asyncAnswer;
         } catch (Exception e) {
-            session.removeClientSignalAttachment(clientAttachment);
+            PacketSignal.removeSignalAttachment(clientAttachment);
             throw e;
         }
     }

net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.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.manager;
+
+import com.zfoo.net.packet.model.SignalPacketAttachment;
+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 PacketSignal {
+
+    private static final Logger logger = LoggerFactory.getLogger(PacketSignal.class);
+
+    // equal with 16383
+    private static final int SIGNAL_MASK = 0B00000000_00000000_00111111_11111111;
+
+    private static AtomicReferenceArray<SignalPacketAttachment> signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1);
+
+    /**
+     * Session控制同步或异步的附加包，key：packetId
+     */
+    private static Map<Integer, SignalPacketAttachment> signalPacketAttachmentMap = new ConcurrentHashMap<>(1000);
+
+    public static void addSignalAttachment(SignalPacketAttachment packetAttachment) {
+        var packetId = packetAttachment.getPacketId();
+        var hash = packetId & SIGNAL_MASK;
+
+        if (!signalPacketArray.compareAndSet(hash, null, packetAttachment)) {
+            signalPacketAttachmentMap.put(packetId, packetAttachment);
+        }
+    }
+
+    public static SignalPacketAttachment removeSignalAttachment(SignalPacketAttachment packetAttachment) {
+        var packetId = packetAttachment.getPacketId();
+        var hash = packetId & SIGNAL_MASK;
+
+        var attachment = signalPacketArray.get(hash);
+        if (attachment != null && attachment.getPacketId() == packetId) {
+            signalPacketArray.lazySet(hash, null);
+            return attachment;
+        }
+        return 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/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java

@@ -15,10 +15,10 @@ package com.zfoo.net.packet.model;
 
 import com.zfoo.protocol.IPacket;
 import com.zfoo.scheduler.util.TimeUtils;
-import com.zfoo.util.security.IdUtils;
 
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * 附加包对业务层透明，禁止在业务层使用
@@ -30,10 +30,12 @@ public class SignalPacketAttachment implements IPacketAttachment {
 
     public static final transient short PROTOCOL_ID = 0;
 
+    private static final AtomicInteger ATOMIC_PACKET_ID = new AtomicInteger(0);
+
     /**
      * 唯一标识一个packet， 唯一表示一个PacketAttachment，hashcode() and equals() 也通过packetId计算
      */
-    private int packetId = IdUtils.getLocalIntId();
+    private int packetId = ATOMIC_PACKET_ID.incrementAndGet();
 
     /**
      * 用来在TaskManage中计算一致性hash的参数

net/src/main/java/com/zfoo/net/session/model/Session.java

@@ -13,16 +13,12 @@
 
 package com.zfoo.net.session.model;
 
-import com.zfoo.net.packet.model.IPacketAttachment;
-import com.zfoo.net.packet.model.SignalPacketAttachment;
 import com.zfoo.protocol.util.StringUtils;
 import io.netty.channel.Channel;
 
-import java.util.Collections;
 import java.util.EnumMap;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
@@ -45,11 +41,6 @@ public class Session {
      */
     private Map<AttributeType, Object> attributes = new EnumMap<>(AttributeType.class);
 
-    /**
-     * 客户端Session控制同步或异步的附加包，key：packetId
-     */
-    private Map<Integer, SignalPacketAttachment> clientSignalPacketAttachmentMap = new ConcurrentHashMap<>();
-
 
     public Session(Channel channel) {
         if (channel == null) {
@@ -60,15 +51,6 @@ public class Session {
     }
 
 
-    public void addClientSignalAttachment(SignalPacketAttachment packetAttachment) {
-        clientSignalPacketAttachmentMap.put(packetAttachment.getPacketId(), packetAttachment);
-    }
-
-    public IPacketAttachment removeClientSignalAttachment(SignalPacketAttachment packetAttachment) {
-        return clientSignalPacketAttachmentMap.remove(packetAttachment.getPacketId());
-    }
-
-
     @Override
     public String toString() {
         return StringUtils.format("[sid:{}] [channel:{}] [attributes:{}]", sid, channel, attributes);
@@ -112,16 +94,11 @@ public class Session {
         return attributes.get(key);
     }
 
-    public Map<Integer, IPacketAttachment> getClientSignalPacketAttachmentMap() {
-        return Collections.unmodifiableMap(clientSignalPacketAttachmentMap);
-    }
-
     public Channel getChannel() {
         return channel;
     }
 
     public void close() {
         channel.close();
-        clientSignalPacketAttachmentMap.clear();
     }
 }

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

@@ -0,0 +1,89 @@
+/*
+ * 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.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.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author jaysunxiao
+ * @version 3.0
+ */
+@Ignore
+public class PacketSignalTest {
+
+    private AtomicInteger atomicInteger = new AtomicInteger(0);
+
+    @Test
+    public void test() throws InterruptedException {
+        // 预热
+        addAndRemove();
+
+        var startTime = TimeUtils.currentTimeMillis();
+
+        var countDownLatch = new CountDownLatch(EventBus.EXECUTORS_SIZE);
+        for (int i = 0; i < EventBus.EXECUTORS_SIZE; i++) {
+            EventBus.asyncExecute(i).execute(new Runnable() {
+                @Override
+                public void run() {
+                    addAndRemove();
+                    countDownLatch.countDown();
+                }
+            });
+        }
+
+        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);
+        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 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);
+        }
+    }
+
+}

net/src/test/java/com/zfoo/net/session/SessionUtils.java

@@ -16,7 +16,6 @@ package com.zfoo.net.session;
 import com.zfoo.net.NetContext;
 import com.zfoo.net.session.model.Session;
 import com.zfoo.protocol.util.FileUtils;
-import com.zfoo.protocol.util.JsonUtils;
 import com.zfoo.protocol.util.StringUtils;
 import com.zfoo.util.ThreadUtils;
 
@@ -34,16 +33,14 @@ public abstract class SessionUtils {
                 builder.append(StringUtils.format("clientSession总数：[{}]", NetContext.getSessionManager().getClientSessionMap().size()));
                 builder.append(FileUtils.LS);
                 for (Session session : NetContext.getSessionManager().getClientSessionMap().values()) {
-                    builder.append(StringUtils.format("[session:{}],[attachment:{}]"
-                            , session.getChannel().remoteAddress(), JsonUtils.object2String(session.getClientSignalPacketAttachmentMap())));
+                    builder.append(StringUtils.format("[session:{}]", session.getChannel().remoteAddress()));
                     builder.append(FileUtils.LS);
                 }
 
                 builder.append(StringUtils.format("serverSession总数：[{}]", NetContext.getSessionManager().getServerSessionMap().size()));
                 builder.append(FileUtils.LS);
                 for (Session session : NetContext.getSessionManager().getServerSessionMap().values()) {
-                    builder.append(StringUtils.format("[session:{}],[attachment:{}]"
-                            , session.getChannel().remoteAddress(), JsonUtils.object2String(session.getClientSignalPacketAttachmentMap())));
+                    builder.append(StringUtils.format("[session:{}]", session.getChannel().remoteAddress()));
                     builder.append(FileUtils.LS);
                 }