線程安全集合類概述

重點(diǎn)介紹java.util.concurrent.* 下的線程安全集合類，可以發(fā)現(xiàn)它們有規(guī)律，里面包含三類關(guān)鍵詞：Blocking、CopyOnWrite、Concurrent

Blocking 大部分實(shí)現(xiàn)基于鎖，并提供用來(lái)阻塞的方法
CopyOnWrite 之類容器修改開(kāi)銷相對(duì)較重
Concurrent 類型的容器
? ?內(nèi)部很多操作使用 cas 優(yōu)化，一般可以提供較高吞吐量
? ?弱一致性
? ? ? ? 遍歷時(shí)弱一致性，例如，當(dāng)利用迭代器遍歷時(shí)，如果容器發(fā)生修改，迭代器仍然可以繼續(xù)進(jìn)行? 遍歷，這時(shí)內(nèi)容是舊的
? ? ? ? 求大小弱一致性，size 操作未必是 100% 準(zhǔn)確
? ? ? ? 讀取弱一致性

遍歷時(shí)如果發(fā)生了修改，對(duì)于非安全容器來(lái)講，使用 fail-fast 機(jī)制也就是讓遍歷立刻失敗，拋出
ConcurrentModificationException，不再繼續(xù)遍歷

ConcurrentHashMap原理

1. JDK 7 HashMap 并發(fā)死鏈

這得用jdk7才有效果，我沒(méi)有jdk7,就體會(huì)一下把

 public static void main(String[] args) {
// 測(cè)試 java 7 中哪些數(shù)字的 hash 結(jié)果相等System.out.println("長(zhǎng)度為16時(shí)，桶下標(biāo)為1的key");for (int i = 0; i < 64; i++) {if (hash(i) % 16 == 1) {System.out.println(i);}}System.out.println("長(zhǎng)度為32時(shí)，桶下標(biāo)為1的key");for (int i = 0; i < 64; i++) {if (hash(i) % 32 == 1) {System.out.println(i);}}
// 1, 35, 16, 50 當(dāng)大小為16時(shí)，它們?cè)谝粋€(gè)桶內(nèi)final HashMap<Integer, Integer> map = new HashMap<Integer, Integer>();
// 放 12 個(gè)元素map.put(2, null);map.put(3, null);map.put(4, null);map.put(5, null);map.put(6, null);map.put(7, null);map.put(8, null);map.put(9, null);map.put(10, null);map.put(16, null);map.put(35, null);map.put(1, null);System.out.println("擴(kuò)容前大小[main]:"+map.size());new Thread() {@Overridepublic void run() {
// 放第 13 個(gè)元素, 發(fā)生擴(kuò)容map.put(50, null);System.out.println("擴(kuò)容后大小[Thread-0]:"+map.size());}}.start();new Thread() {@Overridepublic void run() {
// 放第 13 個(gè)元素, 發(fā)生擴(kuò)容map.put(50, null);System.out.println("擴(kuò)容后大小[Thread-1]:"+map.size());}}.start();}final static int hash(Object k) {int h = 0;if (0 != h && k instanceof String) {return sun.misc.Hashing.stringHash32((String) k);}h ^= k.hashCode();h ^= (h >>> 20) ^ (h >>> 12);return h ^ (h >>> 7) ^ (h >>> 4);}

原始鏈表，格式：[下標(biāo)] (key,next)
[1] (1,35)->(35,16)->(16,null)
線程 a 執(zhí)行到 1 處 ，此時(shí)局部變量 e 為 (1,35)，而局部變量 next 為 (35,16) 線程 a 掛起
線程 b 開(kāi)始執(zhí)行
第一次循環(huán)
[1] (1,null)
第二次循環(huán)
[1] (35,1)->(1,null)
第三次循環(huán)
[1] (35,1)->(1,null)
[17] (16,null)
切換回線程 a，此時(shí)局部變量 e 和 next 被恢復(fù)，引用沒(méi)變但內(nèi)容變了：e 的內(nèi)容被改為 (1,null)，而 next 的內(nèi)
容被改為 (35,1) 并鏈向 (1,null)
第一次循環(huán)
[1] (1,null)
第二次循環(huán)，注意這時(shí) e 是 (35,1) 并鏈向 (1,null) 所以 next 又是 (1,null)
[1] (35,1)->(1,null)
第三次循環(huán)，e 是 (1,null)，而 next 是 null，但 e 被放入鏈表頭，這樣 e.next 變成了 35 （2 處）
[1] (1,35)->(35,1)->(1,35)
已經(jīng)是死鏈了
北

究其原因，是因?yàn)樵诙嗑€程環(huán)境下使用了非線程安全的 map 集合
JDK 8 雖然將擴(kuò)容算法做了調(diào)整，不再將元素加入鏈表頭（而是保持與擴(kuò)容前一樣的順序），但仍不意味著能夠在多線程環(huán)境下能夠安全擴(kuò)容，還會(huì)出現(xiàn)其它問(wèn)題（如擴(kuò)容丟數(shù)據(jù)）

2. JDK 8 ConcurrentHashMap

重要屬性和內(nèi)部類

// 默認(rèn)為 0
// 當(dāng)初始化時(shí), 為 -1
// 當(dāng)擴(kuò)容時(shí), 為 -(1 + 擴(kuò)容線程數(shù))
// 當(dāng)初始化或擴(kuò)容完成后，為 下一次的擴(kuò)容的閾值大小
private transient volatile int sizeCtl;
// 整個(gè) ConcurrentHashMap 就是一個(gè) Node[]
static class Node<K,V> implements Map.Entry<K,V> {}
// hash 表
transient volatile Node<K,V>[] table;
// 擴(kuò)容時(shí)的 新 hash 表
private transient volatile Node<K,V>[] nextTable;
// 擴(kuò)容時(shí)如果某個(gè) bin 遷移完畢, 用 ForwardingNode 作為舊 table bin 的頭結(jié)點(diǎn)
static final class ForwardingNode<K,V> extends Node<K,V> {}
// 用在 compute 以及 computeIfAbsent 時(shí), 用來(lái)占位, 計(jì)算完成后替換為普通 Node
static final class ReservationNode<K,V> extends Node<K,V> {}
// 作為 treebin 的頭節(jié)點(diǎn), 存儲(chǔ) root 和 first
static final class TreeBin<K,V> extends Node<K,V> {}
// 作為 treebin 的節(jié)點(diǎn), 存儲(chǔ) parent, left, right
static final class TreeNode<K,V> extends Node<K,V> {}

重要方法

// 獲取 Node[] 中第 i 個(gè) Node
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i)
// cas 修改 Node[] 中第 i 個(gè) Node 的值, c 為舊值, v 為新值
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v)
// 直接修改 Node[] 中第 i 個(gè) Node 的值, v 為新值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v)
北

構(gòu)造器分析
可以看到實(shí)現(xiàn)了懶惰初始化，在構(gòu)造方法中僅僅計(jì)算了 table 的大小，以后在第一次使用時(shí)才會(huì)真正創(chuàng)建

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
// tableSizeFor 仍然是保證計(jì)算的大小是 2^n, 即 16,32,64 ...
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}

get 流程（全程沒(méi)有加鎖）

public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// spread 方法能確保返回結(jié)果是正數(shù)
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 如果頭結(jié)點(diǎn)已經(jīng)是要查找的 key
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// hash 為負(fù)數(shù)表示該 bin 在擴(kuò)容中或是 treebin, 這時(shí)調(diào)用 find 方法來(lái)查找
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
// 正常遍歷鏈表, 用 equals 比較
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}

put 流程（真是令人頭禿）

   public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
// 其中 spread 方法會(huì)綜合高位低位, 具有更好的 hash 性
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
// f 是鏈表頭節(jié)點(diǎn)
// fh 是鏈表頭結(jié)點(diǎn)的 hash
// i 是鏈表在 table 中的下標(biāo)
Node<K,V> f; int n, i, fh;
// 要?jiǎng)?chuàng)建 table
if (tab == null || (n = tab.length) == 0)
// 初始化 table 使用了 cas, 無(wú)需 synchronized 創(chuàng)建成功, 進(jìn)入下一輪循環(huán)
tab = initTable();
// 要?jiǎng)?chuàng)建鏈表頭節(jié)點(diǎn)
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 添加鏈表頭使用了 cas, 無(wú)需 synchronized
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break;
}
// 幫忙擴(kuò)容
else if ((fh = f.hash) == MOVED)
// 幫忙之后, 進(jìn)入下一輪循環(huán)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
// 鎖住鏈表頭節(jié)點(diǎn)
synchronized (f) {
// 再次確認(rèn)鏈表頭節(jié)點(diǎn)沒(méi)有被移動(dòng)
if (tabAt(tab, i) == f) {
// 鏈表
if (fh >= 0) {
binCount = 1;
// 遍歷鏈表
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 找到相同的 key
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
// 更新
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
// 已經(jīng)是最后的節(jié)點(diǎn)了, 新增 Node, 追加至鏈表尾
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
// 紅黑樹(shù)
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
// putTreeVal 會(huì)看 key 是否已經(jīng)在樹(shù)中, 是, 則返回對(duì)應(yīng)的 TreeNode
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
// 釋放鏈表頭節(jié)點(diǎn)的鎖
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
// 如果鏈表長(zhǎng)度 >= 樹(shù)化閾值(8), 進(jìn)行鏈表轉(zhuǎn)為紅黑樹(shù)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// 增加 size 計(jì)數(shù)
addCount(1L, binCount);
return null;
}
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
Thread.yield();
// 嘗試將 sizeCtl 設(shè)置為 -1（表示初始化 table）
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 獲得鎖, 創(chuàng)建 table, 這時(shí)其它線程會(huì)在 while() 循環(huán)中 yield 直至 table 創(chuàng)建
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
// check 是之前 binCount 的個(gè)數(shù)
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
if (
// 已經(jīng)有了 counterCells, 向 cell 累加
(as = counterCells) != null ||
// 還沒(méi)有, 向 baseCount 累加
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)
) {
CounterCell a; long v; int m;
boolean uncontended = true;
if (
// 還沒(méi)有 counterCells
as == null || (m = as.length - 1) < 0 ||
// 還沒(méi)有 cell
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
// cell cas 增加計(jì)數(shù)失敗
!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
) {
// 創(chuàng)建累加單元數(shù)組和cell, 累加重試
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
// 獲取元素個(gè)數(shù)
s = sumCount();
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// newtable 已經(jīng)創(chuàng)建了，幫忙擴(kuò)容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
// 需要擴(kuò)容，這時(shí) newtable 未創(chuàng)建
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}

size 計(jì)算流程
size 計(jì)算實(shí)際發(fā)生在 put，remove 改變集合元素的操作之中
? ? ? 沒(méi)有競(jìng)爭(zhēng)發(fā)生，向 baseCount 累加計(jì)數(shù)
? ? ? 有競(jìng)爭(zhēng)發(fā)生，新建 counterCells，向其中的一個(gè) cell 累加計(jì)數(shù)
? ? ? ? ? counterCells 初始有兩個(gè) cell
? ? ? ? ? 如果計(jì)數(shù)競(jìng)爭(zhēng)比較激烈，會(huì)創(chuàng)建新的 cell 來(lái)累加計(jì)數(shù)

public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
// 將 baseCount 計(jì)數(shù)與所有 cell 計(jì)數(shù)累加
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}

Java 8 數(shù)組（Node） +（ 鏈表 Node | 紅黑樹(shù) TreeNode ） 以下數(shù)組簡(jiǎn)稱（table），鏈表簡(jiǎn)稱（bin）
初始化，使用 cas 來(lái)保證并發(fā)安全，懶惰初始化 table
樹(shù)化，當(dāng) table.length < 64 時(shí)，先嘗試擴(kuò)容，超過(guò) 64 時(shí)，并且 bin.length > 8 時(shí)，會(huì)將鏈表樹(shù)化，樹(shù)化過(guò)程會(huì)用 synchronized 鎖住鏈表頭
put，如果該 bin 尚未創(chuàng)建，只需要使用 cas 創(chuàng)建 bin；如果已經(jīng)有了，鎖住鏈表頭進(jìn)行后續(xù) put 操作，元素添加至 bin 的尾部
get，無(wú)鎖操作僅需要保證可見(jiàn)性，擴(kuò)容過(guò)程中 get 操作拿到的是 ForwardingNode 它會(huì)讓 get 操作在新table 進(jìn)行搜索
擴(kuò)容，擴(kuò)容時(shí)以 bin 為單位進(jìn)行，需要對(duì) bin 進(jìn)行 synchronized，但這時(shí)妙的是其它競(jìng)爭(zhēng)線程也不是無(wú)事可做，它們會(huì)幫助把其它 bin 進(jìn)行擴(kuò)容，擴(kuò)容時(shí)平均只有 1/6 的節(jié)點(diǎn)會(huì)把復(fù)制到新 table 中
size，元素個(gè)數(shù)保存在 baseCount 中，并發(fā)時(shí)的個(gè)數(shù)變動(dòng)保存在 CounterCell[] 當(dāng)中。最后統(tǒng)計(jì)數(shù)量時(shí)累加即可

LinkedBlockingQueue 原理

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
static class Node<E> {
E item;
/**
* 下列三種情況之一
* - 真正的后繼節(jié)點(diǎn)
* - 自己, 發(fā)生在出隊(duì)時(shí)
* - null, 表示是沒(méi)有后繼節(jié)點(diǎn), 是最后了
*/
Node<E> next;
Node(E x) { item = x; }
}
}

初始化鏈表 last = head = new Node<E>(null); Dummy 節(jié)點(diǎn)用來(lái)占位，item 為 null

當(dāng)一個(gè)節(jié)點(diǎn)入隊(duì) last = last.next = node;

?再來(lái)一個(gè)節(jié)點(diǎn)入隊(duì) last = last.next = node;

出隊(duì)

Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GC
head = first;
E x = first.item;
first.item = null;
return x;

h = head

first = h.next

h.next = h

head = first

E x = first.item;
first.item = null;
return x;

加鎖分析

==高明之處==在于用了兩把鎖和 dummy 節(jié)點(diǎn)
? ? 用一把鎖，同一時(shí)刻，最多只允許有一個(gè)線程（生產(chǎn)者或消費(fèi)者，二選一）執(zhí)行
? ? 用兩把鎖，同一時(shí)刻，可以允許兩個(gè)線程同時(shí)（一個(gè)生產(chǎn)者與一個(gè)消費(fèi)者）執(zhí)行
? ? ? ? 消費(fèi)者與消費(fèi)者線程仍然串行
? ? ? ? 生產(chǎn)者與生產(chǎn)者線程仍然串行

線程安全分析
當(dāng)節(jié)點(diǎn)總數(shù)大于 2 時(shí)（包括 dummy 節(jié)點(diǎn)），putLock 保證的是 last 節(jié)點(diǎn)的線程安全，takeLock 保證的是head 節(jié)點(diǎn)的線程安全。兩把鎖保證了入隊(duì)和出隊(duì)沒(méi)有競(jìng)爭(zhēng)
當(dāng)節(jié)點(diǎn)總數(shù)等于 2 時(shí)（即一個(gè) dummy 節(jié)點(diǎn)，一個(gè)正常節(jié)點(diǎn)）這時(shí)候，仍然是兩把鎖鎖兩個(gè)對(duì)象，不會(huì)競(jìng)爭(zhēng)
當(dāng)節(jié)點(diǎn)總數(shù)等于 1 時(shí)（就一個(gè) dummy 節(jié)點(diǎn)）這時(shí) take 線程會(huì)被 notEmpty 條件阻塞，有競(jìng)爭(zhēng)，會(huì)阻塞

put 操作

public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
int c = -1;
Node<E> node = new Node<E>(e);
final ReentrantLock putLock = this.putLock;
// count 用來(lái)維護(hù)元素計(jì)數(shù)
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
// 滿了等待
while (count.get() == capacity) {
// 倒過(guò)來(lái)讀就好: 等待 notFull
notFull.await();
}
// 有空位, 入隊(duì)且計(jì)數(shù)加一
enqueue(node);
c = count.getAndIncrement();
// 除了自己 put 以外, 隊(duì)列還有空位, 由自己叫醒其他 put 線程
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
// 如果隊(duì)列中有一個(gè)元素, 叫醒 take 線程
if (c == 0)
// 這里調(diào)用的是 notEmpty.signal() 而不是 notEmpty.signalAll() 是為了減少競(jìng)爭(zhēng)
signalNotEmpty();
}

take 操作

public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
// 如果隊(duì)列中只有一個(gè)空位時(shí), 叫醒 put 線程
// 如果有多個(gè)線程進(jìn)行出隊(duì), 第一個(gè)線程滿足 c == capacity, 但后續(xù)線程 c < capacity
if (c == capacity)
// 這里調(diào)用的是 notFull.signal() 而不是 notFull.signalAll() 是為了減少競(jìng)爭(zhēng)
signalNotFull()
return x;
}

LinkedBlockingQueue 與 ArrayBlockingQueue 的性能比較

Linked 支持有界，Array 強(qiáng)制有界
Linked 實(shí)現(xiàn)是鏈表，Array 實(shí)現(xiàn)是數(shù)組
Linked 是懶惰的，而 Array 需要提前初始化 Node 數(shù)組
Linked 每次入隊(duì)會(huì)生成新 Node，而 Array 的 Node 是提前創(chuàng)建好的
Linked 兩把鎖，Array 一把鎖