From 2b0ce755bdc579872c03bc09667d93bf65e05571 Mon Sep 17 00:00:00 2001 From: godotg Date: Sat, 25 Mar 2023 14:48:03 +0800 Subject: [PATCH] feat[map]: fast concurrent segment lock --- .../ConcurrentHashMapLongObject.java | 187 ++++++++++++++++++ .../protocol/collection/ConcurrentTest.java | 38 +++- 2 files changed, 224 insertions(+), 1 deletion(-) create mode 100644 protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashMapLongObject.java diff --git a/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashMapLongObject.java b/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashMapLongObject.java new file mode 100644 index 00000000..3ce66d8d --- /dev/null +++ b/protocol/src/main/java/com/zfoo/protocol/collection/concurrent/ConcurrentHashMapLongObject.java @@ -0,0 +1,187 @@ +/* + * 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.*; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; + + +/** + * @author godotg + * @version 3.0 + */ +public class ConcurrentHashMapLongObject implements Map { + + public static final int DEFAULT_BUCKET_SIZE = 16; + + // 分段锁 + private int buckets; + private ReadWriteLock[] locks; + // bucket对应的分段map + private List> maps; + + public ConcurrentHashMapLongObject(int buckets) { + this.buckets = buckets; + this.locks = new ReadWriteLock[buckets]; + this.maps = new ArrayList<>(buckets); + + for (var i = 0; i < buckets; i++) { + locks[i] = new ReentrantReadWriteLock(); + maps.add(new LongObjectHashMap<>()); + } + } + + public ConcurrentHashMapLongObject() { + this(DEFAULT_BUCKET_SIZE); + } + + private int getBucket(long key) { + return Math.abs((int) key) % buckets; + } + + @Override + public int size() { + var sum = 0; + for (var map : maps) { + sum += map.size(); + } + return sum; + } + + @Override + public boolean isEmpty() { + return size() == 0; + } + + @Override + public boolean containsKey(Object key) { + return containsKey(((Long) key).longValue()); + } + + public boolean containsKey(long key) { + var bucket = getBucket(key); + var readLock = locks[bucket].readLock(); + readLock.lock(); + try { + return maps.get(bucket).containsKey(key); + } finally { + readLock.unlock(); + } + } + + @Override + public boolean containsValue(Object value) { + for (var i = 0; i < buckets; i++) { + var readLock = locks[i].readLock(); + var map = maps.get(i); + readLock.lock(); + try { + if (map.containsValue(value)) { + return true; + } + } finally { + readLock.unlock(); + } + } + return false; + } + + @Override + public V get(Object key) { + return get(((Long) key).longValue()); + } + + public V get(long key) { + var bucket = getBucket(key); + var readLock = locks[bucket].readLock(); + readLock.lock(); + try { + return maps.get(bucket).get(key); + } finally { + readLock.unlock(); + } + } + + @Override + public V put(Long key, V value) { + return put(key.longValue(), value); + } + + public V put(long key, V value) { + var bucket = getBucket(key); + var writeLock = locks[bucket].writeLock(); + writeLock.lock(); + try { + return maps.get(bucket).put(key, value); + } finally { + writeLock.unlock(); + } + } + + @Override + public V remove(Object key) { + return remove(((Long) key).longValue()); + } + + public V remove(long key) { + var bucket = getBucket(key); + var writeLock = locks[bucket].writeLock(); + writeLock.lock(); + try { + return maps.get(bucket).remove(key); + } finally { + writeLock.unlock(); + } + } + + @Override + public void putAll(Map m) { + for (var entry : m.entrySet()) { + put(entry.getKey(), entry.getValue()); + } + } + + @Override + public void clear() { + for (var i = 0; i < buckets; i++) { + var writeLock = locks[i].writeLock(); + var map = maps.get(i); + writeLock.lock(); + try { + map.clear(); + } finally { + writeLock.unlock(); + } + } + } + + @Override + public Set keySet() { + throw new UnsupportedOperationException(); + } + + @Override + public Collection values() { + throw new UnsupportedOperationException(); + } + + @Override + public Set> entrySet() { + throw new UnsupportedOperationException(); + } + + +} diff --git a/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java b/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java index 2b6ba153..78b8207b 100644 --- a/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java +++ b/protocol/src/test/java/com/zfoo/protocol/collection/ConcurrentTest.java @@ -12,6 +12,7 @@ package com.zfoo.protocol.collection; +import com.zfoo.protocol.collection.concurrent.ConcurrentHashMapLongObject; import com.zfoo.protocol.collection.concurrent.CopyOnWriteHashMapLongObject; import org.junit.Assert; import org.junit.Ignore; @@ -29,7 +30,7 @@ public class ConcurrentTest { private static final int EXECUTOR_SIZE = Runtime.getRuntime().availableProcessors(); @Test - public void test() throws InterruptedException { + public void copyOnWriteTest() throws InterruptedException { var map = new CopyOnWriteHashMapLongObject(); var num = 1_0000; var countDownLatch = new CountDownLatch(EXECUTOR_SIZE); @@ -62,4 +63,39 @@ public class ConcurrentTest { countDownLatch2.await(); Assert.assertTrue(map.isEmpty()); } + + @Test + public void concurrentTest() throws InterruptedException { + var map = new ConcurrentHashMapLongObject(); + var num = 100_0000; + var countDownLatch = new CountDownLatch(EXECUTOR_SIZE); + for (var i = 0; i < EXECUTOR_SIZE; i++) { + new Thread(new Runnable() { + @Override + public void run() { + for (int j = 0; j < num; j++) { + map.put(j, j); + } + countDownLatch.countDown(); + } + }).start(); + } + countDownLatch.await(); + Assert.assertEquals(map.size(), num); + + var countDownLatch2 = new CountDownLatch(EXECUTOR_SIZE); + for (var i = 0; i < EXECUTOR_SIZE; 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()); + } }