Java 并發容器總結
作者：碼農SharkChili
這篇文章我們著重探討一下Java中容器的并發管理容器的底層實現和使用技巧，希望對你有幫助。
一、詳解并發場景下的Map容器
1. 詳解JDK7版本HashMap
(1) jdk7版本下的HashMap數據結構
jdk7版本的hashMap底層采用數組加鏈表的形式存儲元素，假如需要存儲的鍵值對經過計算發現存放的位置已經存在鍵值對了，那么就是用頭插法將新節點插入到這個位置。
圖片
對應的我們也給出JDK7版本下的put方法，該版本進行元素插入時會通過hash散列計算得元素對應的索引位置，也就是我們常說的bucket，然后遍歷查看是否存在重復的key，若存在則直接將value覆蓋。反之，則會在循環結束后調用addEntry采用頭插法將元素插入：
public V put(K key, V value) {
//......
//計算key的散列值
int hash = hash(key);
int i = indexFor(hash, table.length);
//定位到對應桶的位置，查看是否存在重復的key，如果有則直接覆蓋
for (Entry<K,V> e = table[i]; e != null; e = e.next) {
Object k;
if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {
V oldValue = e.value;
e.value = value;
e.recordAccess(this);
return oldValue;
}
}
//走到這里說明要在一個空的位置添加節點，將modCount自增，并調用addEntry采用頭插法完成節點插入
modCount++;
addEntry(hash, key, value, i);
returnnull;
}對應我們也給出addEntry的邏輯實現，它會判斷數組是否需要擴容，然后調用createEntry執行頭插法的三步驟：
- 定位到對應bucket的頭節點
- 將新插入節點封裝為Entry，后繼節點指向bucket的頭節點，構成以我們節點為頭節點的鏈表
- 當前bucket指向我們新插入的頭節點
對應源碼如下所示，讀者可結合筆者說明和注釋了解一下過程：
void addEntry(int hash, K key, V value, int bucketIndex) {
//查看數組是否達到閾值，若達到則進行擴容操作
if ((size >= threshold) && (null != table[bucketIndex])) {
resize(2 * table.length);
hash = (null != key) ? hash(key) : 0;
bucketIndex = indexFor(hash, table.length);
}
//使用頭插法將節點插入
createEntry(hash, key, value, bucketIndex);
}
void createEntry(int hash, K key, V value, int bucketIndex) {
//定位bucket的第一個節點
Entry<K,V> e = table[bucketIndex];
//采用頭插法將bucket對應的節點作為新插入節點的后繼節點，再讓table[bucketIndex] 指向我們插入的新節點
table[bucketIndex] = new Entry<>(hash, key, value, e);
size++;
}(2) jdk7版本下的HashMap的擴容
還記得我們上文說明HashMap的put操作時提到的擴容方法resize嘛？它的具體實現如下，可以看到它會根據newCapacity創建一個新的容器newTable ，然后將原數組的元素通過transfer方法轉移到新的容器newTable中。
void resize(int newCapacity) {
Entry[] oldTable = table;
int oldCapacity = oldTable.length;
if (oldCapacity == MAXIMUM_CAPACITY) {
threshold = Integer.MAX_VALUE;
return;
}
//創建新的容器
Entry[] newTable = new Entry[newCapacity];
//將舊的容器的元素轉移到新數組中
transfer(newTable, initHashSeedAsNeeded(newCapacity));
table = newTable;
threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1);
}關于transfer的邏輯，這里涉及到鏈表元素的轉移操作，這里我們也直接以圖文的方式進行說明，執行擴容前會記錄帶轉移元素的e及其后繼節點next：
然后計算該節點e擴容后要存放到新空間的索引位置i，我們假設為4，此時節點e就會指向新空間索引4的頭節點元素，因為我們是entry-0是第一個執行遷移的元素，此時新bucket索引4空間為空，所以我們的entry-0指向空：
待遷移節點entry-0指向 newTable的頭節點后，對應newTable直接指向這個遷移節點，由此完成一個元素entry-0的遷移，同時e指針指向entry-0的后繼節點entry-1：
同理，假設entry-1通過計算后也是要遷移到索引4上，entry-1依然按照：指向newTable索引4位置的頭節點，也就是entry-0作為后繼節點、newTable[4]指向entry-1等步驟不斷循環完成邏輯元素遷移：
有了上述圖解的基礎，我們就可以很好的理解transfer這個元素遷移的源碼邏輯了：
void transfer(Entry[] newTable, boolean rehash) {
int newCapacity = newTable.length;
for (Entry<K,V> e : table) {
while(null != e) {
//獲取遷移節點的后繼節點
Entry<K,V> next = e.next;
//......
//計算遷移節點到新空間的索引位置
int i = indexFor(e.hash, newCapacity);
//節點e的next指向指向插入位置的頭節點，構成一個以自己為頭節點的鏈表
e.next = newTable[i];
//newTable[i]位置指向我們的節點e，完成一個元素遷移
newTable[i] = e;
//e指向第一步記錄的next指針，執行下一輪的元素遷移
e = next;
}
}
}(3) jdk7版本下的HashMap并發擴容問題
當我們了解了JDK7版本的hashMap擴容過程之后，我們就從多線程角度看看什么時候會出現問題，我們不妨想象有兩個線程同時在執行多線程操作。
我們假設線程0和線程1并發執行擴容，單位時間內二者所維護的e和next如下圖所示：
假設線程0先執行，按照擴容的代碼邏輯完成頭插法將entry-0和entry-1都遷移到索引4上，如下圖所示：
重點來了，此時線程1再次獲得CPU時間片指向代碼邏輯，此時：
- e還是指向entry-0，而next還是指向entry-1
- 執行e.next = newTable[i];就會拿到已遷移的entry-1
- 執行 newTable[i] = e;再次指向entry-0，由此關系構成下圖所示的環路
e = next;再次獲得entry-1，兩個元素不斷循環導致CPU100%問題：
通過圖解我們得知CPU100%原因之后，我們不妨通過代碼來重現這個問題。
首先我們將項目JDK版本設置為JDK7。然后定義一個大小為2的map，閾值為1.5，這也就以為著插入時看到size為3的時候會觸發擴容。
/**
* 這個map 桶的長度為2，當元素個數達到 2 * 1.5 = 3 的時候才會觸發擴容
*/
private static HashMap<Integer,String> map = new HashMap<Integer,String>(2,1.5f);所以我們的工作代碼如下，先插入3個元素，然后兩個線程分別插入第4個元素。需要補充一句，這幾個元素的key值是筆者經過調試后確定存放位置都在同一個索引上，所以這段代碼會觸發擴容的邏輯，讀者自定義數據樣本時，最好和讀者保持一致。
try{
map.put(5,"5");
map.put(7,"7");
map.put(3,"3");
System.out.println("此時元素已經達到3了，再往里面添加就會產生擴容操作：" + map);
new Thread("T1") {
public void run() {
map.put(11, "11");
System.out.println(Thread.currentThread().getName() + "擴容完畢 " );
};
}.start();
new Thread("T2") {
public void run() {
map.put(15, "15");
System.out.println(Thread.currentThread().getName() + "擴容完畢 " + map);
};
}.start();
Thread.sleep(60_000);//時間根據debug時間調整
//死循環后打印直接OOM，思考一下為什么？
//因為打印的時候回調用toString回遍歷鏈表，但此時鏈表已經成環狀了
//那么就會無限拼接字符串
// System.out.println(map);
System.out.println(map.get(5));
System.out.println(map.get(7));
System.out.println(map.get(3));
System.out.println(map.get(11));
System.out.println(map.get(15));
System.out.println(map.size());
}catch (Exception e){
}我們在擴容的核心方法插個斷點，斷點條件設置為：
Thread.currentThread().getName().equals("T1")||Thread.currentThread().getName().equals("T2")并且斷點的調試方式改成thread：
我們首先將線程1斷點調試到記錄next引用這一步，然后將線程切換為線程2，模擬線程1被掛起。
我們直接將線程2走完，模擬線程2完成擴容這一步，然后IDEA會自動切回線程1，我們也將線程1直接走完。
從控制臺輸出結果來看，控制臺遲遲無法結束，說明擴容的操作遲遲無法完成，很明顯線程1的擴容操作進入死循環，CPU100%問題由此印證。
2. 詳解JDK8版本的HashMap
(1) 基本數據結構
jdk8對HashMap底層數據結構做了調整，從原本的數組+鏈表轉為數組+鏈表/紅黑樹的形式，即保證在數組長度大于64且當前節點鏈表長度達到8的情況下，為避免元素哈希定位退化為O(n)級別的遍歷，通過鏈表樹化為紅黑樹來保證查詢效率：
對此我們也給出該版本的HashMap源碼，因為作者的風格比較經典，筆者這里就按照核心的4條主線進行說明：
- 經過哈希運算后，對應bucket不存在元素，直接基于key和value生成Node插入。
- 如果定位到的元素key一樣，默認情況下直接將元素值覆蓋并返回舊元素。
- 如果定位到的key對應bucket非空且為樹節點TreeNode則到樹節點中找到重復元素覆蓋或者將新節點插入。
- 如果key對應的bucket為鏈表，則遍歷找到重復節點覆蓋或者找到后繼節點插入。
對應我們put方法對應的核心源碼如下，讀者可以結合注釋了解一下：
final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
boolean evict) {
Node<K,V>[] tab; Node<K,V> p; int n, i;
if ((tab = table) == null || (n = tab.length) == 0)
n = (tab = resize()).length;
//用p記錄哈希定位后的bucket，若為空則直接創建節點存入該bucket中
if ((p = tab[i = (n - 1) & hash]) == null)
tab[i] = newNode(hash, key, value, null);
else {
//若定位到的元素key和當前key一致則將該引用存到e中，后續進行覆蓋處理
Node<K,V> e; K k;
if (p.hash == hash &&
((k = p.key) == key || (key != null && key.equals(k))))
e = p;
elseif (p instanceof TreeNode)//說明定位到的bucket
e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);
else {
//定位到鏈表中的最后一個節點
for (int binCount = 0; ; ++binCount) {
if ((e = p.next) == null) {
p.next = newNode(hash, key, value, null);
if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
treeifyBin(tab, hash);
break;
}
if (e.hash == hash &&
((k = e.key) == key || (key != null && key.equals(k))))
break;
p = e;
}
}
//定位到key相等的元素，如果onlyIfAbsent 設置為false即允許存在時覆蓋，則直接將元素覆蓋，返回就有值
if (e != null) { // existing mapping for key
V oldValue = e.value;
if (!onlyIfAbsent || oldValue == null)
e.value = value;
afterNodeAccess(e);
return oldValue;
}
}
//......
returnnull;
}(2) 多線程操作下的鍵值對覆蓋問題
筆者截取上述片段中的某個代碼段，即哈希定位桶為空的節點添加操作：
//如果數組對應的索引里面沒有元素，則直接插入
if ((p = tab[i = (n - 1) & hash]) == null)
tab[i] = newNode(hash, key, value, null);這段代碼，在并發操作下是存在多線程都判斷到空，然后后者將前者鍵值對覆蓋的情況，如下圖：
所以我們不妨寫個代碼印證這個問題，我們創建一個長度為2的map，用兩個線程往map底層數組的同一個位置中插入鍵值對。兩個線程分別起名為t1、t2，這樣方便后續debug調試。
為了驗證這個問題，筆者使用countDownLatch阻塞一下流程，只有兩個線程都完成工作之后，才能執行后續輸出邏輯。
private static HashMap<String, Long> map = new HashMap<>(2, 1.5f);
public static void main(String[] args) throws InterruptedException {
CountDownLatch countDownLatch = new CountDownLatch(2);
new Thread(() -> {
map.put("3", 3L);
countDownLatch.countDown();
}, "t1").start();
new Thread(() -> {
map.put("5", 5L);
countDownLatch.countDown();
}, "t2").start();
//等待上述線程執行完，繼續執行后續輸出邏輯
countDownLatch.await();
System.out.println(map.get("3"));
System.out.println(map.get("5"));
}然后在插入新節點的地方打個斷點，debug模式設置為thread，條件設置為:
"t1".equals(Thread.currentThread().getName())||"t2".equals(Thread.currentThread().getName())
啟動程序，我們在t1完成判斷，正準備執行創建節點的操作時將線程切換為t2：
可以看到t2準備將(5,5)這個鍵值對插入到數組中，我們直接放行這個邏輯：
此時線程自動切回t1，我們放行斷點，將(3,3)節點插入到數組中。此時，我們已經順利將線程2的鍵值對覆蓋了。
可以看到輸出結果key為5的value為null，hashMap在多線程情況下的索引覆蓋問題得以印證。
(3) 如何解決Map的線程安全問題
解決map線程安全問題有兩種手段，一種是JDK自帶的collections工具，另一種則是并發容器ConcurrentHashMap
為了演示沖突情況下的性能，我們使用不同的map執行100_0000次循環。
@Slf4j
publicclass MapTest {
@Test
public void mapTest() {
StopWatch stopWatch = new StopWatch();
stopWatch.start("synchronizedMap put");
Map<Object, Object> synchronizedMap = Collections.synchronizedMap(new HashMap<>());
IntStream.rangeClosed(0, 100_0000).parallel().forEach(i -> {
synchronizedMap.put(i, i);
});
stopWatch.stop();
stopWatch.start("concurrentHashMap put");
Map<Object, Object> concurrentHashMap = new ConcurrentHashMap<>();
IntStream.rangeClosed(0, 100_0000).parallel().forEach(i -> {
concurrentHashMap.put(i, i);
});
stopWatch.stop();
log.info(stopWatch.prettyPrint());
}
}從輸出結果來看concurrentHashMap 在沖突頻繁的情況下性能更加優異。
2023-03-14 20:29:25,669 INFO MapTest:37 - StopWatch '': running time (millis) = 1422
-----------------------------------------
ms % Task name
-----------------------------------------
00930 065% synchronizedMap put
00492 035% concurrentHashMap put原因很簡單synchronizedMap的put方法，每次操作都會上鎖，這意味著無論要插入的鍵值對在數組哪個位置，執行插入操作前都必須先得到操作map的鎖，鎖的粒度非常大：
public V put(K key, V value) {
synchronized (mutex) {return m.put(key, value);}
}反觀concurrentHashMap 它本質的設計是利用了一種鎖升級的思想，即先通過CAS完成節點插入，失敗后才利用synchronized關鍵字進行鎖定操作，同時鎖的僅僅只是數組中某個索引對應的bucket即利用了鎖分段的思想，分散了鎖的粒度和競爭的壓力：
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) thrownew NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//獲取當前鍵值對要存放的位置f
elseif ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
elseif ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//鎖定的范圍是對應的某個bucket
synchronized (f) {
//......
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
returnnull;
}二、詳解ConcurrentHashMap中的操作注意事項
1. 非原子化操作
使用ConcurrentHashMap存放鍵值對，并不一定意味著所有存的操作都是線程安全的。對于非原子化操作仍然是存在線程安全問題。
如下所示，我們的代碼首先會得到一個含有900的元素的ConcurrentHashMap，然后開10個線程去查看map中還差多少個鍵值對夠1000個，缺多少補多少。
//線程數
privatestaticint THREAD_COUNT = 10;
//數據項的大小
privatestaticint ITEM_COUNT = 1000;
//返回一個size大小的ConcurrentHashMap
private ConcurrentHashMap<String, Object> getData(int size) {
return LongStream.rangeClosed(1, size)
.parallel()
.boxed()
.collect(Collectors.toConcurrentMap(i -> UUID.randomUUID().toString(),
Function.identity(),
(o1, o2) -> o1,
ConcurrentHashMap::new));
}
@GetMapping("wrong")
public String wrong() throws InterruptedException {
//900個元素的ConcurrentHashMap
ConcurrentHashMap<String, Object> map = getData(ITEM_COUNT - 100);
log.info("init size:{}", map.size());
ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
forkJoinPool.execute(() -> {
IntStream.rangeClosed(1, 10).parallel().forEach(i -> {
//判斷當前map缺多少個元素就夠1000個，缺多少補多少
int gap = ITEM_COUNT - map.size();
log.info("{} the gap:{}",Thread.currentThread().getName(), gap);
map.putAll(getData(gap));
});
});
forkJoinPool.shutdown();
forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
log.info("finish size:{}", map.size());
return"ok";
}從輸出結果可以看出，ConcurrentHashMap只能保存put的時候是線程安全，但無法保證put意外的操作線程安全，這段代碼計算ConcurrentHashMap還缺多少鍵值對的操作很可能出現多個線程得到相同的差值，結果補入相同大小的元素，導致ConcurrentHashMap多存放鍵值對的情況。
2023-03-1420:52:52,471 INFO ConcurrentHashMapMisuseController:44 - init size:900
2023-03-1420:52:52,473 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-9 the gap:100
2023-03-1420:52:52,473 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-2 the gap:100
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-6 the gap:100
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-4 the gap:100
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-13 the gap:100
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-11 the gap:100
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-9 the gap:0
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-15 the gap:0
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-10 the gap:-100
2023-03-1420:52:52,474 INFO ConcurrentHashMapMisuseController:51 - ForkJoinPool-1-worker-9 the gap:0
2023-03-1420:52:52,476 INFO ConcurrentHashMapMisuseController:60 - finish size:1500
解決方式也很簡單，將查詢缺少個數和put操作原子化，說的通俗一點就是對查和插兩個操作上一把鎖確保多線程互斥即可。
@GetMapping("right")
public String right() throws InterruptedException {
ConcurrentHashMap<String, Object> map = getData(ITEM_COUNT - 100);
log.info("init size:{}", map.size());
ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
forkJoinPool.execute(() -> {
IntStream.rangeClosed(1, 10).parallel().forEach(i -> {
synchronized (map){
int gap = ITEM_COUNT - map.size();
log.info("{} the gap:{}",Thread.currentThread().getName(), gap);
map.putAll(getData(gap));
}
});
});
forkJoinPool.shutdown();
forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
log.info("finish size:{}", map.size());
return"ok";
}可以看到輸出結果正常了：
2023-03-1420:59:56,730 INFO ConcurrentHashMapMisuseController:69 - init size:900
2023-03-1420:59:56,732 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-9 the gap:100
2023-03-1420:59:56,733 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-4 the gap:0
2023-03-1420:59:56,734 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-8 the gap:0
2023-03-1420:59:56,734 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-9 the gap:0
2023-03-1420:59:56,734 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-1 the gap:0
2023-03-1420:59:56,734 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-15 the gap:0
2023-03-1420:59:56,734 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-2 the gap:0
2023-03-1420:59:56,734 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-6 the gap:0
2023-03-1420:59:56,735 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-11 the gap:0
2023-03-1420:59:56,735 INFO ConcurrentHashMapMisuseController:76 - ForkJoinPool-2-worker-13 the gap:0
2023-03-1420:59:56,737 INFO ConcurrentHashMapMisuseController:87 - finish size:10002. 合理使用API發揮ConcurrentHashMap最大性能
我們會循環1000w次，在這1000w次隨機生成10以內的數字，以10以內數字為key，出現次數為value存放到ConcurrentHashMap中。
你可能會寫出這樣一段代碼：
//map中的項數
privatestaticint ITEM_COUNT = 10;
//線程數
privatestaticint THREAD_COUNT = 10;
//循環次數
privatestaticint LOOP_COUNT = 1000_0000;
private Map<String, Long> normaluse() throws InterruptedException {
Map<String, Long> map = new ConcurrentHashMap<>(ITEM_COUNT);
ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
LongStream.rangeClosed(1, LOOP_COUNT).parallel().forEach(i -> {
String key = "item" + ThreadLocalRandom.current().nextInt(ITEM_COUNT);
synchronized (map) {
if (map.containsKey(key)) {
map.put(key, map.get(key) + 1);
} else {
map.put(key, 1L);
}
}
});
forkJoinPool.shutdown();
forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
return map;
}實際上判斷key是否存在，若不存在則初始化這個key的操作，在ConcurrentHashMap中已經提供好了這樣的API。 我們通過computeIfAbsent進行判斷key是否存在，若不存在則初始化的原子操作，注意此時的value是一個Long類型的累加器，這個LongAdder是一個線程安全的累加器，通過LongAdder的increment方法確保多線程情況下，這一點我們可以在LongAdder的注釋中得知。
LongAdders can be used with a {@link
* java.util.concurrent.ConcurrentHashMap} to maintain a scalable
* frequency map (a form of histogram or multiset). For example, to
* add a count to a {@code ConcurrentHashMap<String,LongAdder> freqs},
* initializing if not already present, you can use {@code
* freqs.computeIfAbsent(k -> new LongAdder()).increment();}大概意思是說LongAdder可以用于統計頻率等場景，所以我們的代碼就直接簡化為下面這段代碼，基于computeIfAbsent和LongAdder的良好設計，這段代碼的語義非常豐富，大體是執行這樣一段操作：
- computeIfAbsent執行k插入，如果k不存在則插入k，value為LongAdder，若存在執行步驟2。
- 不覆蓋原有k，直接返回容器中k對應的LongAdder的引用
- 基于LongAdder的increment完成計數累加
由此也就實現了我們并發詞頻統計的需求了：
ConcurrentHashMap<String,LongAdder> freqs
freqs.computeIfAbsent(k -> new LongAdder()).increment();所以我們改進后的代碼如下：
private Map<String, Long> gooduse() throws InterruptedException {
Map<String, LongAdder> map = new ConcurrentHashMap<>(ITEM_COUNT);
ForkJoinPool forkJoinPool = new ForkJoinPool(THREAD_COUNT);
LongStream.rangeClosed(1, LOOP_COUNT).parallel().forEach(i -> {
String key = "item" + ThreadLocalRandom.current().nextInt(ITEM_COUNT);
map.computeIfAbsent(key, k -> new LongAdder()).increment();
});
forkJoinPool.shutdown();
forkJoinPool.awaitTermination(1, TimeUnit.HOURS);
return map.entrySet().stream()
.collect(Collectors.toMap(e -> e.getKey()
, e -> e.getValue().longValue()));
}完成后我們不妨對這段代碼進行性能壓測：
@GetMapping("good")
public String good() throws InterruptedException {
StopWatch stopWatch = new StopWatch();
stopWatch.start("normaluse");
Map<String, Long> normaluse = normaluse();
stopWatch.stop();
Assert.isTrue(normaluse.size() == ITEM_COUNT, "normaluse size error");
Assert.isTrue(normaluse.entrySet()
.stream()
.mapToLong(i -> i.getValue().longValue())
.reduce(0, Long::sum)
== LOOP_COUNT, "normaluse count error");
stopWatch.start("gooduse");
Map<String, Long> gooduse = gooduse();
stopWatch.stop();
Assert.isTrue(gooduse.size() == ITEM_COUNT, "gooduse size error");
Assert.isTrue(gooduse.entrySet()
.stream()
.mapToLong(i -> i.getValue().longValue())
.reduce(0, Long::sum)
== LOOP_COUNT, "gooduse count error");
log.info(stopWatch.prettyPrint());
return"ok";
}很明顯后者的性能要優于前者，那么原因是什么呢？
-----------------------------------------
ms % Task name
-----------------------------------------
03458 080% normaluse
00871 020% gooduse從ConcurrentHashMap的computeIfAbsent中不難看出，其底層實現"若key不存在則初始化"是通過ReservationNode+CAS實現的，相比于上一段代碼那種非原子化的操作性能自然高出不少。
三、詳解ArrayList線程安全問題
1. 問題重現以原因
我們使用并行流在多線程情況下往list中插入100w個元素。
@Test
public void listTest() {
StopWatch stopWatch = new StopWatch();
List<Object> list=new ArrayList<>();
IntStream.rangeClosed(1, 100_0000).parallel().forEach(i -> {
list.add(i);
});
Assert.assertEquals(100_0000,list.size());
}從輸出結果來看，list確實發生了線程安全問題。
java.lang.AssertionError:
Expected :1000000
Actual :377628我們不妨看看arrayList的add方法，它的邏輯為：
- 判斷當前數組空間是否可以容納新元素，若不夠則創建一個新數組，并將舊數組的元素全部轉移到新數組中
- 將元素e追加到數組末尾
public boolean add(E e) {
//確定當前數組空間是否足夠，若不足則擴容
ensureCapacityInternal(size + 1); // Increments modCount!!
//將元素添加到末尾
elementData[size++] = e;
return true;
}所以如果我們兩個線程同時得到線程空間足夠，然后兩個線程分別執行插入邏輯，如下圖所示，因為各自明確加上自己的元素數組空間2是足夠的，所以執行elementData[size++] = e;時，線程2定位到的索引位置為2出現索引越界：
我們同樣可以寫一段簡單的代碼就能輕易重現這個問題：
@Test
public void listTest() throws InterruptedException {
ArrayList<Object> list = new ArrayList<>(2);
CountDownLatch countDownLatch = new CountDownLatch(2);
list.add(0);
new Thread(() -> {
list.add(1);
countDownLatch.countDown();
}, "t1").start();
new Thread(() -> {
list.add(2);
countDownLatch.countDown();
}, "t2").start();
countDownLatch.await();
System.out.println(list.toString());
}我們的add方法上打一個斷點，并設置條件為t1和t2兩個線程：
在t1線程正準備插入元素時，切換線程到t2：
然后直接將t2線程放行，回到t1線程放行后續操作。問題得以重現：
2. 解決ArrayList線程安全問題的兩個思路
在此回到這段代碼，解決這段代碼線程安全問題的方式有兩種：
@Test
public void listTest() {
StopWatch stopWatch = new StopWatch();
List<Object> list=new ArrayList<>();
IntStream.rangeClosed(1, 100_0000).parallel().forEach(i -> {
list.add(i);
});
Assert.assertEquals(100_0000,list.size());
}- 第一種是使用synchronizedList這個api將容器包裝為線程安全容器：
@Test
public void listTest() {
List<Object> list=Collections.synchronizedList(new ArrayList<>());
IntStream.rangeClosed(1, 100_0000).parallel().forEach(i -> {
list.add(i);
});
Assert.assertEquals(100_0000,list.size());
}- 第二種則是使用CopyOnWriteArrayList這個基于COW思想即寫時復制的并發容器：
@Test
public void listTest() {
List<Object> list=new CopyOnWriteArrayList<>();
IntStream.rangeClosed(1, 100_0000).parallel().forEach(i -> {
list.add(i);
});
Assert.assertEquals(100_0000,list.size());
}3. synchronizedList和CopyOnWriteArrayList區別
雖然兩者都可以保證并發操作的線程安全，但我們還是需要注意兩者使用場景上的區別：
synchronizedList保證多線程操作安全的原理很簡單，每次執行插入或者讀取操作前上鎖。
public E get(int index) {
synchronized (mutex) {return list.get(index);}
}
public void add(int index, E element) {
synchronized (mutex) {list.add(index, element);}
}CopyOnWriteArrayList意味寫時復制，從源碼中不難看出它保證線程安全的方式開銷非常大:
- 獲得寫鎖。
- 復制一個新數組newElements 。
- 在newElements 添加元素。
- 將數組修改為newElements。
對應的我們也給出相應的add源碼的實現邏輯：
public boolean add(E e) {
final ReentrantLock lock = this.lock;
//上鎖
lock.lock();
try {
//復制數組
Object[] elements = getArray();
int len = elements.length;
Object[] newElements = Arrays.copyOf(elements, len + 1);
//添加元素
newElements[len] = e;
//原子覆蓋
setArray(newElements);
returntrue;
} finally {
lock.unlock();
}
}而對于讀CopyOnWriteArrayList則非常簡單，直接返回原數組的值，所以CopyOnWriteArrayList更適合與讀多寫少的場景：
private E get(Object[] a, int index) {
return (E) a[index];
}對此我們對兩者讀寫性能進行了一次壓測，首先是寫性能壓測：
@GetMapping("testWrite")
public Map testWrite() {
int loopCount = 10_0000;
CopyOnWriteArrayList<Integer> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
List<Integer> synchronizedList = Collections.synchronizedList(new ArrayList<>());
//使用copyOnWriteArrayList添加10w個數據
StopWatch stopWatch = new StopWatch();
stopWatch.start("copyOnWriteArrayList add");
IntStream.rangeClosed(1, loopCount)
.parallel()
.forEach(__ -> copyOnWriteArrayList.add(ThreadLocalRandom.current().nextInt(loopCount)));
stopWatch.stop();
//使用synchronizedList添加10w個數據
stopWatch.start("synchronizedList add");
IntStream.rangeClosed(1, loopCount)
.parallel()
.forEach(__ -> synchronizedList.add(ThreadLocalRandom.current().nextInt(loopCount)));
stopWatch.stop();
log.info(stopWatch.prettyPrint());
Map<String, Integer> result = new HashMap<>();
result.put("copyOnWriteArrayList", copyOnWriteArrayList.size());
result.put("synchronizedList", synchronizedList.size());
return result;
}可以看出，高并發寫的情況下synchronizedList 性能更佳。
2023-03-15 00:16:14,532 INFO CopyOnWriteListMisuseController:39 - StopWatch '': running time (millis) = 5556
-----------------------------------------
ms % Task name
-----------------------------------------
05527 099% copyOnWriteArrayList add
00029 001% synchronizedList add讀取性能壓測代碼：
@GetMapping("testRead")
public Map testRead() {
int loopCount = 100_0000;
CopyOnWriteArrayList<Integer> copyOnWriteArrayList = new CopyOnWriteArrayList<>();
List<Integer> synchronizedList = Collections.synchronizedList(new ArrayList<>());
//為兩個list設置100_0000個元素
addAll(copyOnWriteArrayList);
addAll(synchronizedList);
//隨機讀取copyOnWriteArrayList中的元素
StopWatch stopWatch = new StopWatch();
stopWatch.start("copyOnWriteArrayList read");
IntStream.rangeClosed(0, loopCount)
.parallel()
.forEach(__ -> copyOnWriteArrayList.get(ThreadLocalRandom.current().nextInt(loopCount)));
stopWatch.stop();
//隨機讀取synchronizedList中的元素
stopWatch.start("synchronizedList read");
IntStream.rangeClosed(0, loopCount)
.parallel()
.forEach(__ -> synchronizedList.get(ThreadLocalRandom.current().nextInt(loopCount)));
stopWatch.stop();
log.info(stopWatch.prettyPrint());
Map<String, Integer> result = new HashMap<>();
result.put("copyOnWriteArrayList", copyOnWriteArrayList.size());
result.put("synchronizedList", synchronizedList.size());
return result;
}
private void addAll(List<Integer> list) {
list.addAll(IntStream.rangeClosed(1, 100_0000)
.parallel()
.boxed()
.collect(Collectors.toList()));
}而在高并發讀的情況下synchronizedList 性能更加：
2023-03-15 00:16:54,335 INFO CopyOnWriteListMisuseController:74 - StopWatch '': running time (millis) = 310
-----------------------------------------
ms % Task name
-----------------------------------------
00037 012% copyOnWriteArrayList read
00273 088% synchronizedList read四、阻塞隊列ArrayBlockingQueue和延遲隊列DelayQueue
筆者近期已經將阻塞隊列和延遲隊列的文章提交給了開源項目JavaGuide，關于阻塞的隊列讀者可以參考這篇文章:
- https://github.com/Snailclimb/JavaGuide/blob/main/docs/java/collection/arrayblockingqueue-source-code.md
- https://github.com/Snailclimb/JavaGuide/blob/main/docs/java/collection/delayqueue-source-code.md
五、小結
以上筆者對高并發容器的個人理解，總的來說讀者必須掌握以下幾點:
- 通過閱讀源碼了解容器工作機制，代入多線程繪圖推算出可能存在的線程安全問題，并學會使用IDEA加以實踐落地推算結果。
- 了解并發容器工作原理和所有API，確定在指定的場景可以正確使用并發容器保證線程安全和性能。
責任編輯：趙寧寧
來源：
寫代碼的SharkChili
