ConcurrentHashMap结构
在JDK7中ConcurrentHashMap是基于细粒度分离锁实现的，其结果大致如下：
在JDK7中ConcurrentHashMap由多个Segment组成，每个Segment都继承了ReentrantLock，是一把重入锁。
JDK8 中完全重写了ConcurrentHashmap，ConcurrentHashmap实现上和原来的分段式存储有很大的区别，本文讲解的JDK8中ConcurrentHashMap与JDK8的HashMap有相通之处，底层依然由“数组”+链表+红黑树实现，利用全新的CAS算法保证线程安全。Segment虽保留，但已经简化属性，仅仅是为了兼容旧版本。
CAS算法：unsafe.compareAndSwapInt(this, valueOffset, expect, update); CAS(Compare And Swap)，意思是如果valueOffset位置包含的值与expect值相同，则更新valueOffset位置的值为update，并返回true，否则不更新，返回false。
CAS作为知名无锁算法，那ConcurrentHashMap就没用锁了么？当然不是，hash值相同的链表的头结点还是会synchronized上锁。
重要属性
我们重点关注几个出镜率比较高的属性
private transient volatile int sizeCtl;
sizeCtl是控制标识符，volatile关键字保证了其线程可见性，不同的值表示不同的意义。
负数代表正在进行初始化或扩容操作
-1代表正在初始化
-N 表示有N-1个线程正在进行扩容操作
正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小，类似于扩容阈值。它的值始终是当前ConcurrentHashMap容量的0.75倍，这与loadfactor是对应的。实际容量>=sizeCtl，则扩容。
transient volatile Node<K,V>[] table;
盛装Node元素的数组，在第一次插入时被初始化（延迟初始化，并没有在构造函数执行时进行初始化），它的大小总是2的整数次幂 。
//链表转树阈值，大于8时
static final int TREEIFY_THRESHOLD = 8;
//树转链表阈值，小于等于6时（仅在扩容transfer时才可能树转链表）
static final int UNTREEIFY_THRESHOLD = 6;
重要内部类
Node是最核心的内部类，它包装了key-value键值对，所有插入ConcurrentHashMap的数据都包装在这里面。
相关注意的点在源码中用注释说明：
static class Node implements Map.Entry {
final int hash;//final 修饰
final K key; //final 修饰
volatile V val; // volatile 保证线程可见性
volatile Node next; //volatile保证线程可见性
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; }
//不允许直接改变value的值
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
//增加find方法辅助get
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)));
}
TreeNode是另外一个核心的数据结构。当链表长度过长的时候，会转换为TreeNode。但是与HashMap不相同的是，它并不是直接转换为红黑树，而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。
/**
* Nodes for use in TreeBins 链表>8，才可能转为TreeNode
*/
static final class TreeNode extends Node {
TreeNode parent; // red-black tree links
TreeNode left;
TreeNode right;
TreeNode 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;
}
}
TreeNode继承自ConcurrentHashMap的Node类而并非HashMap中的继承自LinkedHashMap.Entry
TreeBin并不负责包装用户的key、value信息，而是包装的很多TreeNode节点。它代替了TreeNode的根节点，也就是说在实际的ConcurrentHashMap“数组”中，存放的是TreeBin对象，而不是TreeNode对象，这是与HashMap的区别。另外这个类还带有了读写锁。
static final class TreeBin extends Node {
TreeNode root;
volatile TreeNode first;
volatile Thread waiter;
volatile int lockState;
// values for lockState
static final int WRITER = 1; // set while holding write lock
static final int WAITER = 2; // set when waiting for write lock
static final int READER = 4; // increment value for setting read lock
......
/**
* Creates bin with initial set of nodes headed by b.
* 构造红黑树
*/
TreeBin(TreeNode b) {
super(TREEBIN, null, null, null);
this.first = b;
TreeNode r = null;
for (TreeNode x = b, next; x != null; x = next) {
next = (TreeNode)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 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 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);
}
......
}
ForwardingNode一个用于连接两个table的节点类，它包含一个nextTable指针，用于指向下一张表。而且这个节点的key value next指针全部为null，它的hash值为-1。生命周期：仅存活于扩容操作且bin不为null时，一定会出现在每个bin的首位。
/**
* A node inserted at head of bins during transfer operations.
*/
static final class ForwardingNode extends Node {
final Node[] nextTable;
ForwardingNode(Node[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}
//从nextTable里进行查询节点，而不是以自身为头结点进行查找
Node find(int h, Object k) {
// loop to avoid arbitrarily deep recursion on forwarding nodes
outer: for (Node[] tab = nextTable;;) {
Node 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)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
}
三个核心方法
ConcurrentHashMap定义了三个原子操作，用于对指定位置的节点进行操作。正是这些原子操作保证了ConcurrentHashMap的线程安全。
//获取i位置上的Node节点
static final Node tabAt(Node[] tab, int i) {
return (Node)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
//利用CAS算法设置i位置上的结点Node,之所以能实现并发是因为它指定了原来这个节点的值是多少
//在CAS算法中会比较内存中的值与你指定的值是否相等，如果相等才接受你的修改，否则拒绝你的修改
//因为当前线程中的值并不是最新的值，这种修改可能会覆盖其他线程的修改结果
static final boolean casTabAt(Node[] tab, int i,
Node c, Node v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
//利用volatile方法设置节点位置i的值，仅在上锁区被调用
static final void setTabAt(Node[] tab, int i, Node v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
初始化initTable
对于ConcurrentHashMap来说，调用它的构造方法仅仅是设置了一些参数而已。而整个table的初始化是在向ConcurrentHashMap中插入第一个元素的时候发生的。
/**
* Initializes table, using the size recorded in sizeCtl.
*/
private final Node[] initTable() {
Node[] 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[] nt = (Node[])new Node[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
初始化方法主要应用了关键属性sizeCtl 如果这个值小于0，表示其他线程正在进行初始化，就放弃这个操作。在这也可以看出ConcurrentHashMap的初始化只能由一个线程完成。如果获得了初始化权限，就用CAS方法将sizeCtl置为-1，防止其他线程进入。初始化数组后，将sizeCtl的值改为0.75*n。
扩容
当ConcurrentHashMap容量不足的时候，需要对table进行扩容。这个方法的基本思想跟HashMap是很像的，但是由于它是支持并发扩容的，所以要复杂的多。
原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足concurrent的要求，而是希望利用并发处理去减少扩容带来的时间影响。因为在扩容的时候，总是会涉及到从一个“数组”到另一个“数组”拷贝的操作，如果这个操作能够并发进行，那真真是极好的了。
整个扩容操作分两部分：
- 构建一个nextTable，它的容量是原来的两倍，这个线程是单线程操作的。单线程的保证是通过RESIZE_STAMP_SHIFT变量经过一次运算来保证的。
- 将原来的table数组复制到nextTable中，这里允许多线程操作。
/** The next table to use; non-null only while resizing.
* 一个过渡的table表，只有在扩容时才使用
*/
private transient volatile Node<K,V>[] nextTable;
我们首先看下单个线程是如何完成的，它的大致思路就是遍历、复制的过程。首先通过运算得到需要遍历的次数i,然后利用tabAt方法获取位置i处的元素。
（1）如果这个位置为空，就在原table的i位置放入ForwardingNode节点，这个也是触发并发扩容的关键点。
（2）如果这个位置是Node节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把它们分别放在nextTable的i和i+n位置。
（3）如果这个节点是TreeBin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi,把处理结果分别放在nextTable的i和i+n位置。
（4）遍历完所有节点后就完成了复制工作。这时让nextTable作为新的table,并且更新sizeCtl为新容量的0.75倍，完成扩容。
再看下多线程的情况
如果线程遍历到的节点是forward节点，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后遍历。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。这个方法设计很值得学习，数据复制的结构图如下：
/* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
*/
private final void transfer(Node[] tab, Node[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node[] nt = (Node[])new Node[n << 1];//构造一个nextTable表，它的容量是原来的两倍
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode fwd = new ForwardingNode(nextTab);//构造一个连节点指针，用于标识位
boolean advance = true;//并发扩容关键变量，如果为true说明这个节点已经处理过
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node f; int fh;
//while循环，依次遍历hash表中的节点
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
//如果所有节点都完成了复制工作，就把nextTable赋值给table,并清空临时对象nextTable
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);//扩容阈值设置为原来的1.5倍，依然相当于现在容量的0.75
return;
}
//CAS算法更新扩容阈值，在这里sizeCtl减1，说明新加入一个线程参与到扩容操作
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
//如果遍历到的节点为空，则放入ForwardingNode指针
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//如果遍历大ForwardingNode节点，说明这个节点节点已经被处理过，直接跳过，这是控制并发扩容的核心
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {//节点上锁
synchronized (f) {
if (tabAt(tab, i) == f) {
Node ln, hn;
if (fh >= 0) {//fh>=0 证明是Node节点
int runBit = fh & n;
//以下部分完成构建两个链表，一个是原链表，另一个是原链表的反序链表
Node lastRun = f;
for (Node p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (Node p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) == 0)
ln = new Node(ph, pk, pv, ln);
else
hn = new Node(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);//在nextTable的i位置插入链表
setTabAt(nextTab, i + n, hn);//在nextTable的i+n位置插入链表
setTabAt(tab, i, fwd);//在table的i位置插入ForwardingNode节点，表示已经处理过该节点
advance = true;//设置true，遍历下一个节点
}
else if (f instanceof TreeBin) {//对TreeBin对象进行处理，和上面相似
TreeBin t = (TreeBin)f;
TreeNode lo = null, loTail = null;
TreeNode hi = null, hiTail = null;
int lc = 0, hc = 0;
//构建正反两个链表
for (Node e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode p = new TreeNode
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
//如果扩容后不再需要Tree结构，反向转换为链表
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
put方法
前面的所有的介绍其实都为这个方法做铺垫。ConcurrentHashMap最常用的就是put和get两个方法。
现在来介绍put方法，这个put方法依然沿用HashMap的put方法的思想，根据hash值计算这个新插入的点在table中的位置i，如果i位置是空的，直接放进去，否则进行判断，如果i位置是树节点，按照树的方式插入新的节点，否则把i插入到链表的末尾。
ConcurrentHashMap中依然沿用这个思想，有一个最重要的不同点就是ConcurrentHashMap不允许key或value为null值。另外由于涉及到多线程，put方法就要复杂一点。 在多线程中可能出现下面情况：
- 如果一个或多个线程正在对ConcurrentHashMap进行扩容，当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到，是因为transfer方法中在空结点上插入forward节点，如果检测到需要插入的位置被forward节点占有，就帮助进行扩容；
- 如果检查到要插入的节点是非空且不说ForwardingNode节点，就对这个节点进行加锁，保证线程安全。尽管这有些影响性能，但还是比HashTable的synchronized要好很多。
整个流程首先定义不允许key或value为null的情况放入，对于每个放入的值，首先利用spread方法对key的hashCode进行一次hash计算，由此决定这个值在table中的位置。
如果这个位置的值为空，不需要加锁操作，利用 CAS 尝试写入，失败则自旋保证成功。
如果这个位置存在节点，说明发生了hash碰撞，首先判断这个节点的类型，如果是链表节点（fh>=0），则得到的节点就是hash值相同的节点组成的链表的头节点。如果遇到hash值与key值都与新加入节点是一致的情况，则只需要更新value值即可。否则依次向后遍历，直到链表尾插入这个结点。如果加入这个节点以后链表长度大于8，就把这个链表转换成红黑树。如果这个节点的类型已经是树节点的话，直接调用树节点的插入方法进行插入新的值。
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
//不允许key或value为null
if (key == null || value == null) throw new NullPointerException();
//计算hash值
int hash = spread(key.hashCode());
int binCount = 0;
//死循环 何时插入成功，何时跳出
for (Node[] tab = table;;) {
Node f; int n, i, fh;
//如果table为空，初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//根据hash值，计算出table里面的位置
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node(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) {//节点上锁，这个节点可以理解为hash值相同节点组成的链表头节点
if (tabAt(tab, i) == f) {
if (fh >= 0) {//说明是链表节点
binCount = 1;
for (Node e = f;; ++binCount) {//遍历链表所有节点
K ek;
//如果hash值，key值都相同，则修改对应的value
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node pred = e;
//如果遍历到最后的节点，那么就证明新的节点需要插入，，就把它插入到链表尾部
if ((e = e.next) == null) {
pred.next = new Node(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {//如果节点是树节点，就按照树 的方式进行插入
Node p;
binCount = 2;
if ((p = ((TreeBin)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {//如果链表节点大于8,就需要把链表结构转换为树结构
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);//当前ConcurrentHashMap的元素数量加1
return null;
}
get方法
get方法比较简单，给定一个key来确定value的时候，必须满足两个条件 key相同 hash值相同，对于节点可能在链表或树上的情况，需要分别去查找。
public V get(Object key) {
Node[] tab; Node e, p; int n, eh; K ek;
int h = spread(key.hashCode());//计算hash值
//根据hash值确定节点位置
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {//如果搜索的节点key与传入的key相同且不为null,直接返回这个节点
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;
}
本文由 feng 创作，采用 知识共享署名4.0
国际许可协议进行许可
本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名
最后编辑时间为: Jun 24,2020