mirror of
https://github.com/tiennm99/zfoo.git
synced 2026-08-06 08:23:55 +00:00
feat[map]: fast concurrent segment lock
This commit is contained in:
godotg
+187
@@ -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<V> implements Map<Long, V> {
|
||||
|
||||
public static final int DEFAULT_BUCKET_SIZE = 16;
|
||||
|
||||
// 分段锁
|
||||
private int buckets;
|
||||
private ReadWriteLock[] locks;
|
||||
// bucket对应的分段map
|
||||
private List<LongObjectHashMap<V>> 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<? extends Long, ? extends V> 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<Long> keySet() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Collection<V> values() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Set<Entry<Long, V>> entrySet() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -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<Integer>();
|
||||
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<Integer>();
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user