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 dcb79d38..c5be0d33 100644 --- a/event/src/main/java/com/zfoo/event/manager/EventBus.java +++ b/event/src/main/java/com/zfoo/event/manager/EventBus.java @@ -16,6 +16,7 @@ package com.zfoo.event.manager; import com.zfoo.event.model.event.IEvent; import com.zfoo.event.model.vo.IEventReceiver; import com.zfoo.protocol.collection.CollectionUtils; +import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject; import com.zfoo.protocol.util.AssertionUtils; import com.zfoo.protocol.util.StringUtils; import com.zfoo.util.SafeRunnable; @@ -47,7 +48,7 @@ public abstract class EventBus { private static final ExecutorService[] executors = new ExecutorService[EXECUTORS_SIZE]; - private static final Map threadMap = new ConcurrentHashMap<>(); + private static final CopyOnWriteHashMapLongObject threadMap = new CopyOnWriteHashMapLongObject<>(EXECUTORS_SIZE); private static final Map, List> receiverMap = new HashMap<>(); @@ -151,7 +152,7 @@ public abstract class EventBus { } public static Executor threadExecutor(long currentThreadId) { - return threadMap.get(currentThreadId); + return threadMap.getPrimitive(currentThreadId); } } diff --git a/net/src/main/java/com/zfoo/net/task/TaskBus.java b/net/src/main/java/com/zfoo/net/task/TaskBus.java index 99301a5e..c4f88993 100644 --- a/net/src/main/java/com/zfoo/net/task/TaskBus.java +++ b/net/src/main/java/com/zfoo/net/task/TaskBus.java @@ -18,6 +18,7 @@ import com.zfoo.net.NetContext; import com.zfoo.net.task.dispatcher.AbstractTaskDispatch; import com.zfoo.net.task.dispatcher.ITaskDispatch; import com.zfoo.net.task.model.PacketReceiverTask; +import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject; import com.zfoo.protocol.util.AssertionUtils; import com.zfoo.protocol.util.StringUtils; import com.zfoo.scheduler.manager.SchedulerBus; @@ -42,7 +43,7 @@ public final class TaskBus { private static final Logger logger = LoggerFactory.getLogger(TaskBus.class); - // 线程池的大小 + // 线程池的大小，也可以通过provider thread配置指定 public static final int EXECUTOR_SIZE; private static final ITaskDispatch taskDispatch; @@ -53,8 +54,6 @@ public final class TaskBus { */ private static final ExecutorService[] executors; - private static final Map threadMap = new ConcurrentHashMap<>(); - static { var localConfig = NetContext.getConfigManager().getLocalConfig(); var providerConfig = localConfig.getProvider(); @@ -73,6 +72,8 @@ public final class TaskBus { } } + private static final CopyOnWriteHashMapLongObject threadMap = new CopyOnWriteHashMapLongObject<>(EXECUTOR_SIZE); + public static class TaskThreadFactory implements ThreadFactory { private final int poolNumber; private final AtomicInteger threadNumber = new AtomicInteger(1); @@ -131,7 +132,7 @@ public final class TaskBus { // 在task，event，scheduler线程执行的异步请求，请求成功过后依然在相同的线程执行回调任务 public static Executor currentThreadExecutor() { var threadId = Thread.currentThread().getId(); - var taskExecutor = threadMap.get(threadId); + var taskExecutor = threadMap.getPrimitive(threadId); if (taskExecutor != null) { return taskExecutor; } diff --git a/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashSet.java b/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashSet.java index f4a61deb..816f135a 100644 --- a/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashSet.java +++ b/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashSet.java @@ -23,15 +23,7 @@ import java.util.concurrent.ConcurrentHashMap; */ public class ConcurrentHashSet extends AbstractSet { - private final Map map; - - public ConcurrentHashSet() { - this.map = new ConcurrentHashMap<>(); - } - - public ConcurrentHashSet(int initialCapacity) { - this.map = new ConcurrentHashMap<>(initialCapacity); - } + private final Map map= new ConcurrentHashMap<>(); @Override public Iterator iterator() { diff --git a/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/CopyOnWriteHashMapLongObject.java b/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/CopyOnWriteHashMapLongObject.java new file mode 100644 index 00000000..f0f09cc7 --- /dev/null +++ b/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/CopyOnWriteHashMapLongObject.java @@ -0,0 +1,141 @@ +/* + * 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.protocol.collection.concurrent; + +import io.netty.util.collection.LongObjectHashMap; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.locks.ReentrantLock; + +/** + * @author godotg + * @version 3.0 + */ +public class CopyOnWriteHashMapLongObject implements Map { + + private final ReentrantLock lock = new ReentrantLock(); + private volatile LongObjectHashMap map; + + public CopyOnWriteHashMapLongObject() { + map = new LongObjectHashMap<>(); + } + + public CopyOnWriteHashMapLongObject(int initialCapacity) { + map = new LongObjectHashMap<>(initialCapacity); + } + + private LongObjectHashMap newCopyMap() { + var newMap = new LongObjectHashMap(); + newMap.putAll(map); + return newMap; + } + + private void setNewMap(LongObjectHashMap newMap) { + map = newMap; + } + + @Override + public int size() { + return map.size(); + } + + @Override + public boolean isEmpty() { + return map.isEmpty(); + } + + @Override + public boolean containsKey(Object key) { + return map.containsKey(key); + } + + @Override + public boolean containsValue(Object value) { + return map.containsValue(value); + } + + @Override + public V get(Object key) { + return map.get(key); + } + + public V getPrimitive(long key) { + return map.get(key); + } + + @Override + public V put(Long key, V value) { + lock.lock(); + try { + var newMap = newCopyMap(); + var oldValue = newMap.put(key, value); + setNewMap(newMap); + return oldValue; + } finally { + lock.unlock(); + } + } + + @Override + public V remove(Object key) { + lock.lock(); + try { + var newMap = newCopyMap(); + var oldValue = newMap.remove(key); + setNewMap(newMap); + return oldValue; + } finally { + lock.unlock(); + } + } + + @Override + public void putAll(Map m) { + lock.lock(); + try { + var newMap = newCopyMap(); + newMap.putAll(m); + setNewMap(newMap); + } finally { + lock.unlock(); + } + } + + @Override + public void clear() { + lock.lock(); + try { + var newMap = newCopyMap(); + setNewMap(newMap); + } finally { + lock.unlock(); + } + } + + @Override + public Set keySet() { + return map.keySet(); + } + + @Override + public Collection values() { + return map.values(); + } + + @Override + public Set> entrySet() { + return map.entrySet(); + } +} diff --git a/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java b/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java new file mode 100644 index 00000000..09d7db03 --- /dev/null +++ b/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java @@ -0,0 +1,64 @@ +/* + * 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.protocol.collection; + +import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject; +import org.junit.Assert; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; + +/** + * @author godotg + * @version 3.0 + */ +@Ignore +public class ConcurrentTest { + + @Test + public void test() throws InterruptedException { + var map = new CopyOnWriteHashMapLongObject(); + var num = 1_0000; + var executorSize = Runtime.getRuntime().availableProcessors(); + var countDownLatch = new CountDownLatch(executorSize); + for (var i = 0; i < executorSize; i++) { + new Thread(new Runnable() { + @Override + public void run() { + for (int j = 0; j < num; j++) { + map.put((long) j, j); + } + countDownLatch.countDown(); + } + }).start(); + } + countDownLatch.await(); + Assert.assertEquals(map.size(), num); + + var countDownLatch2 = new CountDownLatch(executorSize); + for (var i = 0; i < executorSize; i++) { + new Thread(new Runnable() { + @Override + public void run() { + for (int j = 0; j < num; j++) { + map.remove((long) j); + } + countDownLatch2.countDown(); + } + }).start(); + } + countDownLatch2.await(); + Assert.assertTrue(map.isEmpty()); + } +}