From f303555da05e485732a81cfc099960e891289aa1 Mon Sep 17 00:00:00 2001 From: jaysunxiao Date: Sat, 7 Aug 2021 23:06:16 +0800 Subject: [PATCH] =?UTF-8?q?perf[net]:=20=E5=90=8C=E6=AD=A5=E6=88=96?= =?UTF-8?q?=E5=BC=82=E6=AD=A5=E7=9A=84=E8=B0=83=E7=94=A8=E6=8E=A7=E5=88=B6?= =?UTF-8?q?=E5=99=A8=E4=BC=98=E5=85=88=E4=BD=BF=E7=94=A8AtomicReferenceArr?= =?UTF-8?q?ay=E5=A2=9E=E5=BC=BA=E6=80=A7=E8=83=BD=EF=BC=8C=E8=80=8C?= =?UTF-8?q?=E4=B8=8D=E6=98=AF=E4=B8=80=E7=9B=B4=E4=BD=BF=E7=94=A8Concurren?= =?UTF-8?q?tHashMap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/com/zfoo/event/manager/EventBus.java | 2 +- .../zfoo/net/consumer/service/Consumer.java | 5 +- .../dispatcher/manager/PacketDispatcher.java | 12 +-- .../net/dispatcher/manager/PacketSignal.java | 79 ++++++++++++++++ .../packet/model/SignalPacketAttachment.java | 6 +- .../com/zfoo/net/session/model/Session.java | 23 ----- .../zfoo/net/dispatcher/PacketSignalTest.java | 89 +++++++++++++++++++ .../com/zfoo/net/session/SessionUtils.java | 7 +- 8 files changed, 184 insertions(+), 39 deletions(-) create mode 100644 net/src/main/java/com/zfoo/net/dispatcher/manager/PacketSignal.java create mode 100644 net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java diff --git a/event/src/main/java/com/zfoo/event/manager/EventBus.java b/event/src/main/java/com/zfoo/event/manager/EventBus.java index 92bf0b85..1c1783e5 100644 --- a/event/src/main/java/com/zfoo/event/manager/EventBus.java +++ b/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]; diff --git a/net/src/main/java/com/zfoo/net/consumer/service/Consumer.java b/net/src/main/java/com/zfoo/net/consumer/service/Consumer.java index 9f696505..99963774 100644 --- a/net/src/main/java/com/zfoo/net/consumer/service/Consumer.java +++ b/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); } } 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 51ef1873..4687ec84 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 @@ -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; } } 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 new file mode 100644 index 00000000..d1e840e5 --- /dev/null +++ b/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 signalPacketArray = new AtomicReferenceArray<>(SIGNAL_MASK + 1); + + /** + * Session控制同步或异步的附加包，key：packetId + */ + private static Map 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)); + }); + } + +} diff --git a/net/src/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java b/net/src/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java index 16758473..642311c3 100644 --- a/net/src/main/java/com/zfoo/net/packet/model/SignalPacketAttachment.java +++ b/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的参数 diff --git a/net/src/main/java/com/zfoo/net/session/model/Session.java b/net/src/main/java/com/zfoo/net/session/model/Session.java index aa531a30..397cc62e 100644 --- a/net/src/main/java/com/zfoo/net/session/model/Session.java +++ b/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 attributes = new EnumMap<>(AttributeType.class); - /** - * 客户端Session控制同步或异步的附加包，key：packetId - */ - private Map 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 getClientSignalPacketAttachmentMap() { - return Collections.unmodifiableMap(clientSignalPacketAttachmentMap); - } - public Channel getChannel() { return channel; } public void close() { channel.close(); - clientSignalPacketAttachmentMap.clear(); } } diff --git a/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java b/net/src/test/java/com/zfoo/net/dispatcher/PacketSignalTest.java new file mode 100644 index 00000000..90d84da3 --- /dev/null +++ b/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); + } + } + +} diff --git a/net/src/test/java/com/zfoo/net/session/SessionUtils.java b/net/src/test/java/com/zfoo/net/session/SessionUtils.java index 0080158f..675b686f 100644 --- a/net/src/test/java/com/zfoo/net/session/SessionUtils.java +++ b/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); }