前言
上文【从入门到放弃-Java】并发编程-锁-synchronized中，我们介绍了可以使用内置锁synchronized同步类或代码块儿，到达线程安全的目的。
jdk帮我们把常用的一些模块封装成同步容器，如Vector、Hashtable、Collections.synchronizedXxx等。实现方式主要是将常用的容器类加了Synchronized同步。但我们知道，synchronized的频繁使用及竞争较为激烈时，对性能的影响比较大。
jdk1.5之后为我们提供了多种并发容器类，来提升同步容器的性能，这些类主要在java.util.concurrent包（简称juc，包内还有很多其它的并发工具类）中。我们本文先来学习下最常用的并发容器-ConcurrentHashMap。
ConcurrentHashMap
put
/**
* Maps the specified key to the specified value in this table.
* Neither the key nor the value can be null.
*
* <p>The value can be retrieved by calling the {@code get} method
* with a key that is equal to the original key.
*
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key
* @return the previous value associated with {@code key}, or
* {@code null} if there was no mapping for {@code key}
* @throws NullPointerException if the specified key or value is null
*/
// key和value都不能是null
public V put(K key, V value) {
return putVal(key, value, false);
}
/** 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值，将哈希的高位扩展到低位，并将高位强制为0。主要是为了减少hash冲突。
int hash = spread(key.hashCode());
int binCount = 0;
//Node是Map.Entry的实现类，存放key、value。但key、value都不能是null。table的个数是2的n次方
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh; K fk; V fv;
//Node会延迟初始化、即在第一次插入数据的时候进行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//以原子的方式获取Node数组n-1位置的node，如果未null，尝试插入新值
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//使用cas的方式设置新node的key、value值
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; // no lock when adding to empty bin
}
//如果Node是一个ForwardingNode，即有其它线程在扩容，则一起进行扩容操作
else if ((fh = f.hash) == MOVED)
//如果当前正在扩容，则当前线程加入一起帮助扩容。
tab = helpTransfer(tab, f);
//当使用putIfAbsent时，如果map中存在key，则返回对应的value
else if (onlyIfAbsent // check first node without acquiring lock
&& fh == hash
&& ((fk = f.key) == key || (fk != null && key.equals(fk)))
&& (fv = f.val) != null)
return fv;
else {
V oldVal = null;
/**
* currentHashMap在JDK1.8中使用synchronized对需要修改的Node加锁同步，替代了JDK1.7及之前版本采用分段锁的方式。两种方式对比：
* 1、1.7采用数组+Segment+分段锁的方式实现，分段锁及将几个map分为多个类似hashmap的结构，内部是多个Entry链表数组。加锁时，使用ReentrantLock对访问的Segment加锁，其它Segment可以正常操作。缺点是寻找节点需要两次hash，一次找到Segment，一次找到Entry链表的头部。
* 2、1.8采用数组+链表或红黑树的方式实现。使用Node替代了Segment，采用了CAS及synchronized进行同步。当Node链表的长度大于阙值（默认为8）时，会将链表转化为红黑树，提升查找性能。
*
*/
//通过synchronized的方式，对当前Node进行加锁操作。
synchronized (f) {
//判断f节点是否已被其它线程修改
if (tabAt(tab, i) == f) {
//如果当前Node还是链表结构时
if (fh >= 0) {
binCount = 1;
//遍历Node链表，设置value
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果当前节点的key与我们要设置的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<K,V> pred = e;
//设置e为Node链表中的下一个元素，继续判断key是否相等，直到找到相等的key设置值。但如果链表中没有相等的key时，则在链表尾部新增一个元素，并设置值。
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key, value);
break;
}
}
}
//如果当前Node为红黑树结构时
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;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
//如果Node链表的长度大于8时，判断是链表结构扩容，或者转为红黑树结构
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
/**
* Copies all of the mappings from the specified map to this one.
* These mappings replace any mappings that this map had for any of the
* keys currently in the specified map.
*
* @param m mappings to be stored in this map
*/
public void putAll(Map<? extends K, ? extends V> m) {
tryPresize(m.size());
for (Map.Entry<? extends K, ? extends V> e : m.entrySet())
putVal(e.getKey(), e.getValue(), false);
}get
/**
* Returns the value to which the specified key is mapped,
* or {@code null} if this map contains no mapping for the key.
*
* <p>More formally, if this map contains a mapping from a key
* {@code k} to a value {@code v} such that {@code key.equals(k)},
* then this method returns {@code v}; otherwise it returns
* {@code null}. (There can be at most one such mapping.)
*
* @throws NullPointerException if the specified key is null
*/
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//获取hash值
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//如果获取到的Node的hash值和key的相等，则说明是链表。
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//如果获取Node的hash值小于0则说明是非链式结构
else if (eh < 0)
//不断查找Node的下一个节点，知道找到为止
return (p = e.find(h, key)) != null ? p.val : null;
//不断查找Node的下一个节点，直到找到为止（感觉和find重复了。最外层的if中只需要一个Node::find方法就能搞定。知道原因的大神请指正）
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}treeifyBin
扩容或将结构转为红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n;
if (tab != null) {
//如果当前Node数组小于64则扩容，大于64时则转换为红黑树结构
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
//rehash：resize
tryPresize(n << 1);
//如果是链表结构则转换为红黑树结构
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for (Node<K,V> e = b; e != null; e = e.next) {
//创建树节点，加入红黑树中
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}tryPresize
扩容操作
/**
* Tries to presize table to accommodate the given number of elements.
*
* @param size number of elements (doesn't need to be perfectly accurate)
*/
private final void tryPresize(int size) {
//size在传入前已经翻倍，这里会再次调整，变为为：大于(1.5 * oldSize + 1)的2的幂，且小于MAXIMUM_CAPACITY的大小
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
//当sizeCtl小于等于0时。说明已有线程在初始化或者rehash了
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
如果table是空，即未初始化的话，进行初始化。
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
//sc = n - n / 4 = 0.75，在final中，将sizeCtl设置为当前大小的0.75倍。大于这个阙值时，会再次进行扩容。
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
//如果还未开始迁移
else if (tab == table) {
int rs = resizeStamp(n);
if (U.compareAndSetInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
// 开始迁移
transfer(tab, null);
}
}
}transfer
将Node迁移至新的table中
/**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//设置线程迁移数据的步长，单核步长为n，多核为(n >>> 3) / NCPU， 最小为16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//如果要迁移的table还未初始化，则进行初始化动作
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
//开始迁移，为要迁移的Node创建一个ForwardingNode节点。key和value都是null，hashcode为MOVED，nextTable指向新的table
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//表示一个节点已被迁移完毕，可以迁移下一个了。
boolean advance = true;
//迁移过程是否完毕。
boolean finishing = false; // to ensure sweep before committing nextTab
//i是迁移的起始位置，bound是迁移的末尾。
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
//其实位置大于结束位置，说明已经迁移完毕
if (--i >= bound || finishing)
advance = false;
//如果transferIndex小于等于0，则说明节点都已有线程在迁移了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSetInt
(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;
//迁移完毕后，将新的table赋值给table成员变量，修改sizeCtl完成迁移
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSetInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//不相等说明还有线程没迁移完毕
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
//所有线程迁移完毕后，设置finishing为完成。
finishing = advance = true;
i = n; // recheck before commit
}
}
//如果tab[i] = null，设置为fwd
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//如果当前节点已经迁移，则处理下一个节点
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//加锁同步处理
synchronized (f) {
//验证下是否已经被其它线程处理
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
//如果是链表结构
if (fh >= 0) {
//按照Node中元素hash值的第log(2)(n)位，记为runBit，是0或1将Node链表分为两个新的链表。
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//将runBit位为0的链表记为ln，为1的设为hn。这里是标记最后一个不一致的节点，lastRun后节点的runBit都一样，因此不用新修改节点，减少消耗
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
//将runBit位为0的链表记为ln，为1的设为hn。
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//将分开的两个节点设置为table的i和i+n位。
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
//如果是红黑树结构
else if (f instanceof TreeBin) {
TreeBin<K,V> t = (TreeBin<K,V>)f;
//按照Node中元素hash值的第log(2)(n)位，记为runBit，是0或1将Node红黑树分为两颗树。
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(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;
}
}
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
}
}
}总结
本文分析了currentHashMap是如何实现线程安全并提升性能的、如何扩容、JDK1.7和1.8实现方式的区别等
- 分Node加synchronize锁，不影响其它node的读写
- Node节点hash冲突的元素数量少于8时，使用链表结构，大于等于8时，转换为红黑树结构提升查找性能
- 扩容时，会将table的长度扩大为大于(1.5 * oldSize + 1)的2的幂大小，并将每个Node根据log(2)(n)位是0或1，分为两个Node，放在新table的i和i+n的位置
- JDK1.8将原currentHashMap使用数组+segment+ReentrantLock的方式改为数组+Node+CAS+synchronized的方式。减少了hash次数并采用cas和红黑树等多种优化提升性能
更多文章
见我的博客：https://nc2era.com
