翻译自： https://www.baeldung.com/java-concurrent-map
1. 概要
Map是使用最广泛的Java集合之一。
其中，HashMap是线程不安全的；HashTable是线程安全的，基于synchronized。
虽然HashTable是线程安全的，但是它不是很高效。另一个完全同步的Map Collections.sychronizedMap 也不是很高效。如果我们想要线程安全并且在高并发的情况下保持高吞吐量，这些都不合适。
要解决这个问题，java1.5的集合框架里提供了ConcurrentMap接口。下面我们将基于java1.8来介绍一下ConcurrentMap。
2. ConcurrentMap
CourrentMap接口继承了Map接口。它提供了一种代码结构，用来指引解决数据读写时的线程安全问题。
ConcurrentMap重写了Map中的很多default方法，用来实现线程安全和内存一致性的原子操作。
很多的默认实现被重写，不允许使用null作为key/value。
getOrDefault
forEach
replaceAll
computeIfAbsent
computeIfPresent
compute
merge
下面的方法也将被重写，用来支持原子性，没有默认的接口实现、
putIfAbsent
remove
replace（key，oldValue，newValue）
repalce（key， value）
其他的方法是直接继承自Map的。
3. ConcurrentHashMap
ConcurrentHashMap是一个开箱即用的ConcurrentMap的实现类。
为了获得更好的性能，它将数据分成多个bucket，放在一个数组里，数组是一张哈希表，就是以前的段，在更新的时候使用CAS（compareAndSwap）。
bucket是懒加载的，在第一次插入的时候。每个bucket都会被独立的锁上，锁住桶中的第一个node。读操作是不会被阻塞，并且更新的资源争夺也被限制了。
段的数量是由访问表的线程决定的，所以段的更新，一次只有一个线程。
在java8之前，段的数量是由访问的线程数量决定的，一个线程一个段。
这就是为什么它的构造函数和HashMap比起来多了concurrencyLevel的参数，它就是估算的线程数量。
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
另外两个参数和HashMap相同。
但是，从java8开始，构造函数只是为了和之前的版本兼容，保留了下来，他们只能影响初始的map的大小。
3.1 Thead-safety
ConcurrenMap保证了在多线程的环境里操作key/value的内存一致性。
一个线程想要将对象当作key或者value放到map里 必须在 在另一个线程想要访问或者删除这个对象之前。
来确认下，让我们来看下内存不一致的例子。
import org.junit.Assert;
import org.junit.Test;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* Author: Neal Shan
* Date: 2019/4/22
*/
public class ConcurrentMapTest {
@Test
public void givenHashMap_whenSumParalllel_thenError() throws Exception {
Map<String, Integer> map = new HashMap<>();
List<Integer> sumList = parallelSum100(map, 1000);
Assert.assertNotEquals(1, sumList
.stream()
.distinct()
.count());
long wrongResultCount = sumList
.stream()
.filter(num -> num != 100)
.count();
Assert.assertTrue(wrongResultCount > 0);
}
@Test
public void givenConcurrentHashMap_whenSumParallel_thenCOrrect()
throws Exception {
Map<String, Integer> map = new ConcurrentHashMap<>();
List<Integer> sumList = parallelSum100(map, 1000);
Assert.assertEquals(1, sumList
.stream()
.distinct()
.count());
long wrongResultCount = sumList
.stream()
.filter(num -> num != 100)
.count();
Assert.assertEquals(0, wrongResultCount);
}
private List<Integer> parallelSum100(Map<String, Integer> map, int executionTimes)
throws InterruptedException {
List<Integer> sumList = new ArrayList<>(1000);
for (int i = 0; i < executionTimes; i++) {
map.put("test", 0);
ExecutorService executorService =
Executors.newFixedThreadPool(4);
for (int j = 0; j < 10; j++) {
executorService.execute(() -> {
for (int k = 0; k < 10; k++) {
map.computeIfPresent(
"test",
(key, value) -> value + 1
);
}
});
}
executorService.shutdown();
executorService.awaitTermination(5, TimeUnit.SECONDS);
sumList.add(map.get("test"));
}
return sumList;
}
}
对于每个map.computeIfPresent的并发操作，HashMap没有提供一个与当前的整数一致的最新的整数，导致不一致和不期望的结果。
但是ConcurrentHashMap可以得到一致的，正确的结果。
3.2 Null key/value
很多ConcurrentMap提供的API不允许null的key或者value。
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenPutWithinNullKey_thenThrowsNPE() {
ConcurrentMap concurrentMap = new ConcurrentHashMap();
concurrentMap.put(null, new Object());
}
@Test(expected = NullPointerException.class)
public void givenConcurrentHashMap_whenPutNullValueThrowsNPE() {
ConcurrentMap concurrentMap = new ConcurrentHashMap();
concurrentMap.put("test", null);
}
但是对于compute和merge的话，computed的value可以为null，用来表示key-value的映射被删除了。
@Test
public void givenConcurrentHashMap_whenComputeRemappingNull_thenMappingRemoved() {
Object oldValue = new Object();
ConcurrentMap concurrentMap = new ConcurrentHashMap();
concurrentMap.put("test", oldValue);
concurrentMap.compute("test", (s, o) -> null);
Assert.assertNull(concurrentMap.get("test"));
}
3.3 Stream Support
java8提供了Stream的支持。
不想其他的stream方法，批量的操作在并发条件下也是安全的。ConcurrentModificationException不会被抛出，同样也应用到它的iterators。相对于streams，很多forEach，serach，reduce方法也被加进来支持遍历，以及map-reduce操作。
3.4 Performance
在表面之下，ConcurrentHashMap和HashMap很相似，都是访问，更新数据，基于一个哈希表，虽然更复杂一些。
当然，ConcurrentHashMap在很多的并发读写数据的场景下可以获得更好的性能。
让我们来写一个简单的性能测试micro-benchmark，测试get，put的性能。并且和HashTable以及Collections.synchronizedMap比较，每个方法在4个线程中跑50,0000次。
@Test
public void givenMaps_whenGetPut500KTimes_thenConcurrentMapFaster() throws InterruptedException {
Map<String, Object> hashtable = new Hashtable<>();
Map<String, Object> synchronizedMap = Collections.synchronizedMap(new HashMap<>());
Map<String, Object> concurrnetMap = new ConcurrentHashMap<>();
long hashtableAvgRuntime = timeElapseForGetPut(hashtable);
long syncHashMapAvgRuntime = timeElapseForGetPut(synchronizedMap);
long concurrentHashMapAvgRuntime = timeElapseForGetPut(concurrnetMap);
Assert.assertTrue(hashtableAvgRuntime > concurrentHashMapAvgRuntime);
Assert.assertTrue(syncHashMapAvgRuntime > concurrentHashMapAvgRuntime);
}
private long timeElapseForGetPut(Map<String, Object> map) throws InterruptedException {
ExecutorService executorService =
Executors.newFixedThreadPool(4);
long startTime = System.nanoTime();
for (int i = 0; i < 4; i++) {
executorService.execute(() -> {
for (int j = 0; j < 500_000; j++) {
int value = ThreadLocalRandom
.current()
.nextInt(10000);
String key = String.valueOf(value);
map.put(key, value);
map.get(key);
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.SECONDS);
return (System.nanoTime() - startTime) / 500_000;
}
要注意的是，micro-benchmark只是一个简单的场景，不能反映真实场景。
在多线程的的环境中，多个线程访问同一个map，ConcurrentHashMap是最合适的。
但是当Map只被一个线程访问时，那就用HashMap比较合适了，简单，性能稳定。
3.5 Pitfalls
读操作不会阻塞concurrenthashmap，而且还可以和更新操作重叠。为了更好的性能，他们只反映最近完成更新的操作，见官方https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ConcurrentHashMap.html
有许多要记住的：
1. 计算map集合状态的方法包含size，isEmpty，containsValue，只有在map没有其他线程更新时有用。
@Test
public void givenConcurrentMap_whenUpdatingAndGetSize_thenError()
throws InterruptedException{
final int MAX_SIZE = 1000;
List<Integer> mapSizes = new ArrayList<>();
ConcurrentMap<String, Integer> concurrentMap = new ConcurrentHashMap<>();
Runnable collecitonMapSizes = () -> {
for (int i = 0; i < MAX_SIZE; i++) {
mapSizes.add(concurrentMap.size());
}
};
Runnable updateMapData = () -> {
for (int i = 0; i < MAX_SIZE; i++) {
concurrentMap.put(String.valueOf(i), i);
}
};
ExecutorService executorService =
Executors.newFixedThreadPool(4);
executorService.execute(updateMapData);
executorService.execute(collecitonMapSizes);
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.SECONDS);
Assert.assertNotEquals(MAX_SIZE, mapSizes.get(MAX_SIZE - 1).intValue());
Assert.assertEquals(MAX_SIZE, concurrentMap.size());
}
如果并发更新是在严格的控制下，聚合状态的的结果还是可靠的。虽然，这些聚合的方法不能保证实时的精准，但是他们也许适合用来做监控或者估算。
注意使用size（）的时候应该使用mappingCount（）代替，它返回一个long count（），虽然在底层上他们是基于相同的估算的。
hashcode是很重要的，记住使用相同的hashcode是会严重影响性能的。
如果key实现了Comparable接口，ConcurrentHashMap可能使用比较排序后的keys来帮助斩断联系，改善。但是我们还是要避免使用相同的hashcode。
iterators只是为单线程设计的，他们提供了弱的一致性，也不直接报错，他们永远不会抛出CocurrentModificationException。
默认的初始capacity是16，在会随着concurrencyLevel调整。
要注意remapping函数，虽然我们可以remapping，使用compute或者merge函数。我们应该让他们保持快，短，简单，并且专注在当前mapping上，来避免不可预期的阻塞。
keys在concurrenthashmap里是无序的，在某些情况下，当排序是需要的，我们推荐使用ConcurrentSkipListMap。
4. ConcurrentNavigableMap
在某些需要排序的情况下，我么可以使用ConcurrentSkipListMap，一个concurrent类型的TreeMap。
作为ConcurrentMap的一个补充。ConcurrentNavigableMap支持完全排序keys，默认是按升序排的。并且可以并发排序。返回map的方法都兼容了并发。
subMap
headMap
tailMap
descendingMap
keySet()的iterators和spliterators是一种弱内存一致性。
navigableKeySet
keySet
descendingKeySet
5. ConcurrentSkipListMap
前面，我们讲了NavigableMap 接口，还有它的实现类TreeMap。ConcurrentSkipListMap可以看成是并发版的TreeMap。
在实际中，java中没有红黑树的实现。ConcurrentSkipListMap是使用的SkipLists的一个并发版的变种，提供了一个时间复杂度为log（n）的containsKey， get， put，remove以及他们的变种。
另外，TreeMap的特性，key 插入，删除，更新，以及访问都是线程安全的。下面就来比较一下TreeMap在并发条件下的表现。
@Test
public void giventSkipListMap_whenNavConcurrently_thenCOuntCorrect()
throws InterruptedException {
NavigableMap<Integer, Integer> skipListMap =
new ConcurrentSkipListMap<>();
int count = countMapElementByPollingFirstEntry(skipListMap, 10000, 4);
Assert.assertEquals(10000 * 4, count);
}
@Test
public void giventTreeMap_whenNavCOncurrently_thenCOuntError()
throws InterruptedException {
NavigableMap<Integer, Integer> treeMap = new TreeMap<>();
int count = countMapElementByPollingFirstEntry(treeMap, 10000, 4);
Assert.assertNotEquals(10000 * 4, count);
}
private int countMapElementByPollingFirstEntry(
NavigableMap<Integer, Integer> navigableMap,
int elementCount,
int concurrencyLevel) throws InterruptedException {
for (int i = 0; i < elementCount * concurrencyLevel; i++) {
navigableMap.put(i, 1);
}
AtomicInteger counter = new AtomicInteger(0);
ExecutorService executorService =
Executors.newFixedThreadPool(concurrencyLevel);
for (int j = 0; j < concurrencyLevel; j++) {
executorService.execute(() -> {
for (int i = 0; i < elementCount; i++) {
if (navigableMap.pollFirstEntry() != null) {
counter.incrementAndGet();
}
}
});
}
executorService.shutdown();
executorService.awaitTermination(1, TimeUnit.SECONDS);
return counter.get();
}
这只是对他们的一些简单的介绍，详细请看javadoc。
6. 总结
在这篇文章里，我们介绍了ConcurrentMap接口，以及concurrenthashmap的特性，还有concurrentNavigableMap的key排序。
