ConcurrentHashMap解析（JDK8）

在JDK8中，ConcurrentHashMap实现机制较JDK7发生了很大变化，其摒弃了分段锁（Segment）的概念，而是利用CAS算法，与Hashtable一样，内部由“数组+链表+红黑树”的方式实现。同时又增加了许多辅助类，例如TreeBin，以实现并发性。

构造函数

构造器中有3个参数，分别是initialCapacity，loadFactor，concurrencyLevel。

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);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}

注意，在构造函数中有一个名为sizeCtl的成员变量，需要解释一下：sizeCtl是一个控制标识符，当为负数时表示正在进行初始化或扩容操作（-1代表正在初始化，-N代表有N-1个线程正在进行扩容操作）；当为正数时表示还未被初始化，此时这个数值代表初始化或下一次扩容的大小（类似于扩容阈值的概念）。

其它一些成员变量如下：

/**
* bins的数组，在第一次插入操作时延迟加载
* 数组的大小是2的整数次幂
*/
transient volatile Node<K,V>[] table;

/**
* 表的初始化和扩容的控制标识
*/
private transient volatile int sizeCtl;

/**
* The next table index (plus one) to split while resizing.
*/
private transient volatile int transferIndex;

/**
* Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
*/
private transient volatile int cellsBusy;

/**
* Table of counter cells. When non-null, size is a power of 2.
*/
private transient volatile CounterCell[] counterCells;

内部类

Node
Node是ConcurrentHashMap最核心的内部类，封装了key-value键值对。它不允许调用setValue方法直接修改Node的val域。源码如下：

static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;

Node(int hash, K key, V val, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}

public final K getKey()       { return key; }
public final V getValue()     { return val; }
public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
public final String toString(){ return key + "=" + val; }
public final V setValue(V value) {
throw new UnsupportedOperationException();<span style="white-space:pre">	</span>//不允许修改val域
}

public final boolean equals(Object o) {
Object k, v, u; Map.Entry<?,?> e;
return ((o instanceof Map.Entry) &&
(k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
(v = e.getValue()) != null &&
(k == key || k.equals(key)) &&
(v == (u = val) || v.equals(u)));
}

/**
* Virtualized support for map.get(); overridden in subclasses.
*/
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}
}

注意，Node类的val域和next域都为volatile。

TreeNode

当链表长度过长时，会转换成TreeNode类。注意不同于HashMap，它并不是直接转换为红黑树的，而是把这些结点封装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的封装。并且TreeNode继承于Node。源码如下：

static final class TreeNode<K,V> extends Node<K,V> {
TreeNode<K,V> parent;  // red-black tree links
TreeNode<K,V> left;
TreeNode<K,V> right;
TreeNode<K,V> prev;    // needed to unlink next upon deletion
boolean red;

TreeNode(int hash, K key, V val, Node<K,V> next,
TreeNode<K,V> parent) {
super(hash, key, val, next);
this.parent = parent;
}

Node<K,V> find(int h, Object k) {
return findTreeNode(h, k, null);
}

/**
* Returns the TreeNode (or null if not found) for the given key
* starting at given root.
*/
final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
if (k != null) {
TreeNode<K,V> p = this;
do  {
int ph, dir; K pk; TreeNode<K,V> q;
TreeNode<K,V> pl = p.left, pr = p.right;
if ((ph = p.hash) > h)
p = pl;
else if (ph < h)
p = pr;
else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
return p;
else if (pl == null)
p = pr;
else if (pr == null)
p = pl;
else if ((kc != null ||
(kc = comparableClassFor(k)) != null) &&
(dir = compareComparables(kc, k, pk)) != 0)
p = (dir < 0) ? pl : pr;
else if ((q = pr.findTreeNode(h, k, kc)) != null)
return q;
else
p = pl;
} while (p != null);
}
return null;
<span style="white-space:pre">	</span>}
}

TreeBin

TreeNode使用在bins的头部。TreeBins不持有用户的键值，但是相反持有指向TreeNodes的列表和它们的根。它们也持有了一个读写锁，用于在树重新构造之前，写入线程去等待读取线程完成。也就是说，在ConcurrentHashMap内部结构中存储的不是TreeNode对象，而是TreeBin对象。

其构造方法如下：

TreeBin(TreeNode<K,V> b) {
super(TREEBIN, null, null, null);
this.first = b;
TreeNode<K,V> r = null;
for (TreeNode<K,V> x = b, next; x != null; x = next) {
next = (TreeNode<K,V>)x.next;
x.left = x.right = null;
if (r == null) {
x.parent = null;
x.red = false;
r = x;
}
else {
K k = x.key;
int h = x.hash;
Class<?> kc = null;
for (TreeNode<K,V> p = r;;) {
int dir, ph;
K pk = p.key;
if ((ph = p.hash) > h)
dir = -1;
else if (ph < h)
dir = 1;
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) ||
(dir = compareComparables(kc, k, pk)) == 0)
dir = tieBreakOrder(k, pk);
TreeNode<K,V> xp = p;
if ((p = (dir <= 0) ? p.left : p.right) == null) {
x.parent = xp;
if (dir <= 0)
xp.left = x;
else
xp.right = x;
r = balanceInsertion(r, x);
break;
}
}
}
}
this.root = r;
assert checkInvariants(root);
}

在构造方法中，仅仅指定它的hash作为TREEBIN常量。同时也可以看到红黑树的构造。

ForwardingNode

ForwardingNode在转换期间插入到bins的头部的结点。 源码如下：

static final class For
e143
wardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}

Node<K,V> find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
outer: for (Node<K,V>[] tab = nextTable;;) {
Node<K,V> e; int n;
if (k == null || tab == null || (n = tab.length) == 0 ||
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
for (;;) {
int eh; K ek;
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
if (eh < 0) {
if (e instanceof ForwardingNode) {
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
}

它包含一个nextTable指针，用于指向下一个table。

在ConcurrentHashMap中，大量使用了Unsafe，其利用CAS算法。具体参见：http://blog.csdn.net/tian_ex/article/details/50492711

基本操作

Put操作

映射指定的键和值到表中，注意键和值都不能为Null。源码如下：

public V put(K key, V value) {
<span style="white-space:pre">	</span>return putVal(key, value, false);
}

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();<span style="white-space:pre">	</span>//不允许key或value为空
int hash = spread(key.hashCode());<span style="white-space:pre">	</span>//计算hash值
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)<span style="white-space:pre">	</span>
tab = initTable();
else if ((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
}
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
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;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}

首先根据key计算出hash值，然后调用tabAt方法传入hash值获取对应的Node。如果table的null或者其长度为0，则调用initTable方法初始化table，源码如下：

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(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>
;
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}

此时需要注意的是，对于ConcurrentHashMap而言，构造方法只是设置一些参数，真正table的初始化是在第一次插入操作时发生的。当sizeCtl小于0，之前讲过，此时表示有其它线程正在初始化table，故而调用Thread的yield方法，表示向线程调度器建议挂起自己。这样确保了始终只有一个线程进行初始化工作。否则利用CAS算法将sizeCtl置为-1，表示本线程正在初始化table。

在putVal方法中接着根据hash值获取在table中的索引。如果索引对应的值为空，则新建一个Node并放入。否则，在之前获取到的结点f加锁，如果fh大于0，表示此结点是一个链表的结点，然后遍历链表，如果遍历到的结点的hash值和key值都相等，则修改遍历到结点的值，如果遍历到最后一个结点，那就新建一个Node并把它插入链表的尾部；如果fh小于0且是TreeNode的子类，则表示此结点是树的结点，之后按照树的方式去遍历结点并进行修改。注意此时如果链表的长度达到阈值8的话，就需要把链表结构转换为树结构。

最后调用addCount方法将ConcurrentHashMap中的元素个数加1，其源码如下：

private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
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;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}

这个方法有两个参数：x和check。check用于确定是否检查扩容，x表示增加的数量。

get操作

get方法通过给定一个key获得value。源码如下：

public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}

此时需要考虑在链表和树结构去查找。

size操作

在ConcurrentHashMap获取map中的元素是一个估计值，因为在计算元素的个数的时候可能有其它线程对map进行增删操作。Java中提供了两种获取元素个数的方法：size方法以及mappingCount方法。其中mappingCount方法是JDK8增加的，根据Java API文档，这个方法应该代替size方法使用，因为ConcurrentHashMap可能包含映射的数量超过int所能表示的最大数量，其返回值是long类型：

Returns the number of mappings. This method should be used instead ofsize()because
a ConcurrentHashMap may contain more mappings than can be represented as an int. The value returned is an estimate; the actual count may differ if there are concurrent insertions or removals.

源码如下：

public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}

public long mappingCount() {
long n = sumCount();
return (n < 0L) ? 0L : n; // ignore transient negative values
}

final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}

由上可知，俩方法都是基于调用sumCount方法实现的。

未完待续！