做网站客户会问什么问题,昆明百度搜索排名优化,服务外包平台,哪些网站是做婴童辅食招商的JUC第十五讲#xff1a;JUC集合-ConcurrentHashMap详解 本文是JUC第十五讲#xff1a;JUC集合-ConcurrentHashMap详解。JDK1.7之前的ConcurrentHashMap使用分段锁机制实现#xff0c;JDK1.8则使用数组链表红黑树数据结构和CAS原子操作实现ConcurrentHashMap#xff1b;本文…JUC第十五讲JUC集合-ConcurrentHashMap详解 本文是JUC第十五讲JUC集合-ConcurrentHashMap详解。JDK1.7之前的ConcurrentHashMap使用分段锁机制实现JDK1.8则使用数组链表红黑树数据结构和CAS原子操作实现ConcurrentHashMap本文将分别介绍这两种方式的实现方案及其区别。 文章目录 JUC第十五讲JUC集合-ConcurrentHashMap详解1、带着BAT大厂的面试问题去理解2、为什么HashTable慢3、ConcurrentHashMap - JDK 1.73.1、数据结构3.2、初始化3.3、put 过程分析3.4、初始化槽: ensureSegment3.5、获取写入锁: scanAndLockForPut3.6、扩容: rehash3.7、get 过程分析3.8、并发问题分析 4、ConcurrentHashMap - JDK 1.84.1、数据结构4.2、初始化4.3、put 过程分析4.4、初始化数组: initTable4.5、链表转红黑树: treeifyBin4.6、扩容: tryPresize4.7、数据迁移: transfer4.8、get 过程分析 5、对比总结6、ConcurrentHashMap使用案例Action1讲讲为什么ConcurrentHashMap是并发安全的吧既然有锁怎么去统计size呢Action2ConcurrentHashMap为啥线程安全呢(10分)参考文章 1、带着BAT大厂的面试问题去理解
请带着这些问题继续后文会很大程度上帮助你更好的理解相关知识点。
为什么HashTable慢? 它的并发度是什么? 那么ConcurrentHashMap并发度是什么? 并发度是segment的数量ConcurrentHashMap在JDK1.7和JDK1.8中实现有什么差别? JDK1.8解決了JDK1.7中什么问题 1.7是segment 1.8是数组链表红黑树ConcurrentHashMap JDK1.7实现的原理是什么? 分段锁机制ConcurrentHashMap JDK1.8实现的原理是什么? 数组链表红黑树CASConcurrentHashMap JDK1.7中Segment数(concurrencyLevel)默认值是多少? 为何一旦初始化就不可再扩容? 默认为16ConcurrentHashMap JDK1.7说说其put的机制?ConcurrentHashMap JDK1.7是如何扩容的? rehash(注segment 数组不能扩容扩容是 segment 数组某个位置内部的数组 HashEntryK,V[] 进行扩容)ConcurrentHashMap JDK1.8是如何扩容的? tryPresizeConcurrentHashMap JDK1.8链表转红黑树的时机是什么? 临界值为什么是8? 链表的长度大于8ConcurrentHashMap JDK1.8是如何进行数据迁移的? transfer
2、为什么HashTable慢
Hashtable之所以效率低下主要是因为其实现使用了synchronized关键字对put等操作进行加锁而synchronized关键字加锁是对整个对象进行加锁也就是说在进行put等修改Hash表的操作时锁住了整个Hash表从而使得其表现的效率低下。
3、ConcurrentHashMap - JDK 1.7
在JDK1.5~1.7版本Java使用了分段锁机制实现ConcurrentHashMap.
简而言之ConcurrentHashMap在对象中保存了一个Segment数组即将整个Hash表划分为多个分段而每个Segment元素即每个分段则类似于一个Hashtable这样在执行put操作时首先根据hash算法定位到元素属于哪个Segment然后对该Segment加锁即可。因此ConcurrentHashMap在多线程并发编程中可是实现多线程put操作。接下来分析JDK1.7版本中ConcurrentHashMap的实现原理。
3.1、数据结构
整个 ConcurrentHashMap 由一个个 Segment 组成Segment 代表”部分“或”一段“的意思所以很多地方都会将其描述为分段锁。注意行文中我很多地方用了“槽”来代表一个 segment。
简单理解就是ConcurrentHashMap 是一个 Segment 数组Segment 通过继承 ReentrantLock 来进行加锁所以每次需要加锁的操作锁住的是一个 segment这样只要保证每个 Segment 是线程安全的也就实现了全局的线程安全。 concurrencyLevel: 并行级别、并发数、Segment 数怎么翻译不重要理解它。默认是 16也就是说 ConcurrentHashMap 有 16 个 Segments所以理论上这个时候最多可以同时支持 16 个线程并发写只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值但是一旦初始化以后它是不可以扩容的。
再具体到每个 Segment 内部其实每个 Segment 很像之前介绍的 HashMap不过它要保证线程安全所以处理起来要麻烦些。
3.2、初始化
initialCapacity: 初始容量这个值指的是整个 ConcurrentHashMap 的初始容量实际操作的时候需要平均分给每个 Segment。loadFactor: 负载因子之前我们说了Segment 数组不可以扩容所以这个负载因子是给每个 Segment 内部使用的。
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {if (!(loadFactor 0) || initialCapacity 0 || concurrencyLevel 0)throw new IllegalArgumentException();if (concurrencyLevel MAX_SEGMENTS)concurrencyLevel MAX_SEGMENTS;// Find power-of-two sizes best matching argumentsint sshift 0;int ssize 1;// 计算并行级别 ssize因为要保持并行级别是 2 的 n 次方while (ssize concurrencyLevel) {sshift;ssize 1;}// 我们这里先不要那么烧脑用默认值concurrencyLevel 为 16sshift 为 4// 那么计算出 segmentShift 为 28segmentMask 为 15后面会用到这两个值this.segmentShift 32 - sshift;this.segmentMask ssize - 1;if (initialCapacity MAXIMUM_CAPACITY)initialCapacity MAXIMUM_CAPACITY;// initialCapacity 是设置整个 map 初始的大小// 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小// 如 initialCapacity 为 64那么每个 Segment 或称之为槽可以分到 4 个int c initialCapacity / ssize;if (c * ssize initialCapacity)c;// 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2这个值也是有讲究的因为这样的话对于具体的槽上// 插入一个元素不至于扩容插入第二个的时候才会扩容int cap MIN_SEGMENT_TABLE_CAPACITY; while (cap c)cap 1;// 创建 Segment 数组// 并创建数组的第一个元素 segment[0]SegmentK,V s0 new SegmentK,V(loadFactor, (int)(cap * loadFactor),(HashEntryK,V[])new HashEntry[cap]);SegmentK,V[] ss (SegmentK,V[])new Segment[ssize];// 往数组写入 segment[0]UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments ss;
}初始化完成我们得到了一个 Segment 数组。
我们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的那么初始化完成后:
Segment 数组长度为 16不可以扩容Segment[i] 的默认大小为 2负载因子是 0.75得出初始阈值为 1.5也就是以后插入第一个元素不会触发扩容插入第二个会进行第一次扩容这里初始化了 segment[0]其他位置还是 null至于为什么要初始化 segment[0]后面的代码会介绍当前 segmentShift 的值为 32 - 4 28segmentMask 为 16 - 1 15姑且把它们简单翻译为移位数和掩码这两个值马上就会用到
3.3、put 过程分析
我们先看 put 的主流程对于其中的一些关键细节操作后面会进行详细介绍。
public V put(K key, V value) {SegmentK,V s;if (value null)throw new NullPointerException();// 1. 计算 key 的 hash 值int hash hash(key);// 2. 根据 hash 值找到 Segment 数组中的位置 j// hash 是 32 位无符号右移 segmentShift(28) 位剩下高 4 位// 然后和 segmentMask(15) 做一次与操作也就是说 j 是 hash 值的高 4 位也就是槽的数组下标int j (hash segmentShift) segmentMask;// 刚刚说了初始化的时候初始化了 segment[0]但是其他位置还是 null// ensureSegment(j) 对 segment[j] 进行初始化if ((s (SegmentK,V)UNSAFE.getObject // nonvolatile; recheck(segments, (j SSHIFT) SBASE)) null) // in ensureSegments ensureSegment(j);// 3. 插入新值到 槽 s 中return s.put(key, hash, value, false);
}第一层皮很简单根据 hash 值很快就能找到相应的 Segment之后就是 Segment 内部的 put 操作了。
Segment 内部是由 数组链表 组成的。
final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 在往该 segment 写入前需要先获取该 segment 的独占锁// 先看主流程后面还会具体介绍这部分内容HashEntryK,V node tryLock() ? null :scanAndLockForPut(key, hash, value);V oldValue;try {// 这个是 segment 内部的数组HashEntryK,V[] tab table;// 再利用 hash 值求应该放置的数组下标int index (tab.length - 1) hash;// first 是数组该位置处的链表的表头HashEntryK,V first entryAt(tab, index);// 下面这串 for 循环虽然很长不过也很好理解想想该位置没有任何元素和已经存在一个链表这两种情况for (HashEntryK,V e first;;) {if (e ! null) {K k;if ((k e.key) key ||(e.hash hash key.equals(k))) {oldValue e.value;if (!onlyIfAbsent) {// 覆盖旧值e.value value;modCount;}break;}// 继续顺着链表走e e.next;}else {// node 到底是不是 null这个要看获取锁的过程不过和这里都没有关系。// 如果不为 null那就直接将它设置为链表表头如果是null初始化并设置为链表表头。if (node ! null)node.setNext(first);elsenode new HashEntryK,V(hash, key, value, first);int c count 1;// 如果超过了该 segment 的阈值这个 segment 需要扩容if (c threshold tab.length MAXIMUM_CAPACITY)rehash(node); // 扩容后面也会具体分析else// 没有达到阈值将 node 放到数组 tab 的 index 位置// 其实就是将新的节点设置成原链表的表头setEntryAt(tab, index, node);modCount;count c;oldValue null;break;}}} finally {// 解锁unlock();}return oldValue;
}整体流程还是比较简单的由于有独占锁的保护所以 segment 内部的操作并不复杂。至于这里面的并发问题我们稍后再进行介绍。
到这里 put 操作就结束了接下来我们说一说其中几步关键的操作。
3.4、初始化槽: ensureSegment
ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0]对于其他槽来说在插入第一个值的时候进行初始化。
这里需要考虑并发因为很可能会有多个线程同时进来初始化同一个槽 segment[k]不过只要有一个成功了就可以。
private SegmentK,V ensureSegment(int k) {final SegmentK,V[] ss this.segments;long u (k SSHIFT) SBASE; // raw offsetSegmentK,V seg;if ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) {// 这里看到为什么之前要初始化 segment[0] 了// 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]// 为什么要用“当前”因为 segment[0] 可能早就扩容过了SegmentK,V proto ss[0];int cap proto.table.length;float lf proto.loadFactor;int threshold (int)(cap * lf);// 初始化 segment[k] 内部的数组HashEntryK,V[] tab (HashEntryK,V[])new HashEntry[cap];if ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) { // 再次检查一遍该槽是否被其他线程初始化了。SegmentK,V s new SegmentK,V(lf, threshold, tab);// 使用 while 循环内部用 CAS当前线程成功设值或其他线程成功设值后退出while ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) {if (UNSAFE.compareAndSwapObject(ss, u, null, seg s))break;}}}return seg;
}总的来说ensureSegment(int k) 比较简单对于并发操作使用 CAS 进行控制。
3.5、获取写入锁: scanAndLockForPut
前面我们看到在往某个 segment 中 put 的时候首先会调用 node tryLock() ? null : scanAndLockForPut(key, hash, value)也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁如果失败那么进入到 scanAndLockForPut 这个方法来获取锁。
下面我们来具体分析这个方法中是怎么控制加锁的。
private HashEntryK,V scanAndLockForPut(K key, int hash, V value) {HashEntryK,V first entryForHash(this, hash);HashEntryK,V e first;HashEntryK,V node null;int retries -1; // negative while locating node// 循环获取锁while (!tryLock()) {HashEntryK,V f; // to recheck first belowif (retries 0) {if (e null) {if (node null) // speculatively create node// 进到这里说明数组该位置的链表是空的没有任何元素// 当然进到这里的另一个原因是 tryLock() 失败所以该槽存在并发不一定是该位置node new HashEntryK,V(hash, key, value, null);retries 0;}else if (key.equals(e.key))retries 0;else// 顺着链表往下走e e.next;}// 重试次数如果超过 MAX_SCAN_RETRIES(单核1多核64)那么不抢了进入到阻塞队列等待锁// lock() 是阻塞方法直到获取锁后返回else if (retries MAX_SCAN_RETRIES) {lock();break;}else if ((retries 1) 0 // 这个时候是有大问题了那就是有新的元素进到了链表成为了新的表头// 所以这边的策略是相当于重新走一遍这个 scanAndLockForPut 方法(f entryForHash(this, hash)) ! first) {e first f; // re-traverse if entry changedretries -1;}}return node;
}这个方法有两个出口一个是 tryLock() 成功了循环终止另一个就是重试次数超过了 MAX_SCAN_RETRIES进到 lock() 方法此方法会阻塞等待直到成功拿到独占锁。
这个方法就是看似复杂但是其实就是做了一件事那就是获取该 segment 的独占锁如果需要的话顺便实例化了一下 node。
3.6、扩容: rehash
重复一下segment 数组不能扩容扩容是 segment 数组某个位置内部的数组 HashEntryK,V[] 进行扩容扩容后容量为原来的 2 倍。
首先我们要回顾一下触发扩容的地方put 的时候如果判断该值的插入会导致该 segment 的元素个数超过阈值那么先进行扩容再插值读者这个时候可以回去 put 方法看一眼。
该方法不需要考虑并发因为到这里的时候是持有该 segment 的独占锁的。
// 方法参数上的 node 是这次扩容后需要添加到新的数组中的数据。
private void rehash(HashEntryK,V node) {HashEntryK,V[] oldTable table;int oldCapacity oldTable.length;// 2 倍int newCapacity oldCapacity 1;threshold (int)(newCapacity * loadFactor);// 创建新数组HashEntryK,V[] newTable (HashEntryK,V[]) new HashEntry[newCapacity];// 新的掩码如从 16 扩容到 32那么 sizeMask 为 31对应二进制 ‘000...00011111’int sizeMask newCapacity - 1;// 遍历原数组老套路将原数组位置 i 处的链表拆分到 新数组位置 i 和 ioldCap 两个位置for (int i 0; i oldCapacity ; i) {// e 是链表的第一个元素HashEntryK,V e oldTable[i];if (e ! null) {HashEntryK,V next e.next;// 计算应该放置在新数组中的位置// 假设原数组长度为 16e 在 oldTable[3] 处那么 idx 只可能是 3 或者是 3 16 19int idx e.hash sizeMask;if (next null) // 该位置处只有一个元素那比较好办newTable[idx] e;else { // Reuse consecutive sequence at same slot// e 是链表表头HashEntryK,V lastRun e;// idx 是当前链表的头节点 e 的新位置int lastIdx idx;// 下面这个 for 循环会找到一个 lastRun 节点这个节点之后的所有元素是将要放到一起的for (HashEntryK,V last next;last ! null;last last.next) {int k last.hash sizeMask;if (k ! lastIdx) {lastIdx k;lastRun last;}}// 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置newTable[lastIdx] lastRun;// 下面的操作是处理 lastRun 之前的节点// 这些节点可能分配在另一个链表中也可能分配到上面的那个链表中for (HashEntryK,V p e; p ! lastRun; p p.next) {V v p.value;int h p.hash;int k h sizeMask;HashEntryK,V n newTable[k];newTable[k] new HashEntryK,V(h, p.key, v, n);}}}}// 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部int nodeIndex node.hash sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] node;table newTable;
}这里的扩容比之前的 HashMap 要复杂一些代码难懂一点。上面有两个挨着的 for 循环第一个 for 有什么用呢?
仔细一看发现如果没有第一个 for 循环也是可以工作的但是这个 for 循环下来如果 lastRun 的后面还有比较多的节点那么这次就是值得的。因为我们只需要克隆 lastRun 前面的节点后面的一串节点跟着 lastRun 走就是了不需要做任何操作。
我觉得 Doug Lea 的这个想法也是挺有意思的不过比较坏的情况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素那么这次遍历就有点浪费了。不过 Doug Lea 也说了根据统计如果使用默认的阈值大约只有 1/6 的节点需要克隆。
3.7、get 过程分析
相对于 put 来说get 就很简单了。
计算 hash 值找到 segment 数组中的具体位置或我们前面用的“槽”槽中也是一个数组根据 hash 找到数组中具体的位置到这里是链表了顺着链表进行查找即可
public V get(Object key) {SegmentK,V s; // manually integrate access methods to reduce overheadHashEntryK,V[] tab;// 1. hash 值int h hash(key);long u (((h segmentShift) segmentMask) SSHIFT) SBASE;// 2. 根据 hash 找到对应的 segmentif ((s (SegmentK,V)UNSAFE.getObjectVolatile(segments, u)) ! null (tab s.table) ! null) {// 3. 找到segment 内部数组相应位置的链表遍历for (HashEntryK,V e (HashEntryK,V) UNSAFE.getObjectVolatile(tab, ((long)(((tab.length - 1) h)) TSHIFT) TBASE);e ! null; e e.next) {K k;if ((k e.key) key || (e.hash h key.equals(k)))return e.value;}}return null;
}3.8、并发问题分析
现在我们已经说完了 put 过程和 get 过程我们可以看到 get 过程中是没有加锁的那自然我们就需要去考虑并发问题。
添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的所以它们之间自然不会有问题我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。
put 操作的线程安全性。 初始化槽这个我们之前就说过了使用了 CAS 来初始化 Segment 中的数组。添加节点到链表的操作是插入到表头的所以如果这个时候 get 操作在链表遍历的过程已经到了中间是不会影响的。当然另一个并发问题就是 get 操作在 put 之后需要保证刚刚插入表头的节点被读取这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。扩容。扩容是新创建了数组然后进行迁移数据最后面将 newTable 设置给属性 table。所以如果 get 操作此时也在进行那么也没关系如果 get 先行那么就是在旧的 table 上做查询操作而 put 先行那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。 remove 操作的线程安全性。 remove 操作我们没有分析源码所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。get 操作需要遍历链表但是 remove 操作会破坏链表。如果 remove 破坏的节点 get 操作已经过去了那么这里不存在任何问题。如果 remove 先破坏了一个节点分两种情况考虑。 1、如果此节点是头节点那么需要将头节点的 next 设置为数组该位置的元素table 虽然使用了 volatile 修饰但是 volatile 并不能提供数组内部操作的可见性保证所以源码中使用了 UNSAFE 来操作数组请看方法 setEntryAt。2、如果要删除的节点不是头节点它会将要删除节点的后继节点接到前驱节点中这里的并发保证就是 next 属性是 volatile 的。
4、ConcurrentHashMap - JDK 1.8
在JDK1.7之前ConcurrentHashMap是通过分段锁机制来实现的所以其最大并发度受Segment的个数限制。因此在JDK1.8中ConcurrentHashMap的实现原理摒弃了这种设计而是选择了与HashMap类似的数组链表红黑树的方式实现而加锁则采用CAS和synchronized实现。
4.1、数据结构 结构上和 Java8 的 HashMap 基本上一样不过它要保证线程安全性所以在源码上确实要复杂一些。
4.2、初始化
// 这构造函数里什么都不干
public ConcurrentHashMap() {
}
public ConcurrentHashMap(int initialCapacity) {if (initialCapacity 0)throw new IllegalArgumentException();int cap ((initialCapacity (MAXIMUM_CAPACITY 1)) ?MAXIMUM_CAPACITY :tableSizeFor(initialCapacity (initialCapacity 1) 1));this.sizeCtl cap;
}这个初始化方法有点意思通过提供初始容量计算了 sizeCtlsizeCtl 【 (1.5 * initialCapacity 1)然后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10那么得到 sizeCtl 为 16如果 initialCapacity 为 11得到 sizeCtl 为 32。
sizeCtl 这个属性使用的场景很多不过只要跟着文章的思路来就不会被它搞晕了。
4.3、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();// 得到 hash 值int hash spread(key.hashCode());// 用于记录相应链表的长度int binCount 0;for (NodeK,V[] tab table;;) {NodeK,V f; int n, i, fh;// 如果数组空进行数组初始化if (tab null || (n tab.length) 0)// 初始化数组后面会详细介绍tab initTable();// 找该 hash 值对应的数组下标得到第一个节点 felse if ((f tabAt(tab, i (n - 1) hash)) null) {// 如果数组该位置为空// 用一次 CAS 操作将这个新值放入其中即可这个 put 操作差不多就结束了可以拉到最后面了// 如果 CAS 失败那就是有并发操作进到下一个循环就好了if (casTabAt(tab, i, null,new NodeK,V(hash, key, value, null)))break; // no lock when adding to empty bin}// hash 居然可以等于 MOVED这个需要到后面才能看明白不过从名字上也能猜到肯定是因为在扩容else if ((fh f.hash) MOVED)// 帮助数据迁移这个等到看完数据迁移部分的介绍后再理解这个就很简单了tab helpTransfer(tab, f);else { // 到这里就是说f 是该位置的头节点而且不为空V oldVal null;// 获取数组该位置的头节点的监视器锁synchronized (f) {if (tabAt(tab, i) f) {if (fh 0) { // 头节点的 hash 值大于 0说明是链表// 用于累加记录链表的长度binCount 1;// 遍历链表for (NodeK,V e f;; binCount) {K ek;// 如果发现了相等的 key判断是否要进行值覆盖然后也就可以 break 了if (e.hash hash ((ek e.key) key ||(ek ! null key.equals(ek)))) {oldVal e.val;if (!onlyIfAbsent)e.val value;break;}// 到了链表的最末端将这个新值放到链表的最后面NodeK,V pred e;if ((e e.next) null) {pred.next new NodeK,V(hash, key,value, null);break;}}}else if (f instanceof TreeBin) { // 红黑树NodeK,V p;binCount 2;// 调用红黑树的插值方法插入新节点if ((p ((TreeBinK,V)f).putTreeVal(hash, key,value)) ! null) {oldVal p.val;if (!onlyIfAbsent)p.val value;}}}}if (binCount ! 0) {// 判断是否要将链表转换为红黑树临界值和 HashMap 一样也是 8if (binCount TREEIFY_THRESHOLD)// 这个方法和 HashMap 中稍微有一点点不同那就是它不是一定会进行红黑树转换// 如果当前数组的长度小于 64那么会选择进行数组扩容而不是转换为红黑树// 具体源码我们就不看了扩容部分后面说treeifyBin(tab, i);if (oldVal ! null)return oldVal;break;}}}// addCount(1L, binCount);return null;
}4.4、初始化数组: initTable
这个比较简单主要就是初始化一个合适大小的数组然后会设置 sizeCtl。
初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。
private final NodeK,V[] initTable() {NodeK,V[] tab; int sc;while ((tab table) null || tab.length 0) {// 初始化的功劳被其他线程抢去了if ((sc sizeCtl) 0)Thread.yield(); // lost initialization race; just spin// CAS 一下将 sizeCtl 设置为 -1代表抢到了锁else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if ((tab table) null || tab.length 0) {// DEFAULT_CAPACITY 默认初始容量是 16int n (sc 0) ? sc : DEFAULT_CAPACITY;// 初始化数组长度为 16 或初始化时提供的长度NodeK,V[] nt (NodeK,V[])new Node?,?[n];// 将这个数组赋值给 tabletable 是 volatile 的table tab nt;// 如果 n 为 16 的话那么这里 sc 12// 其实就是 0.75 * nsc n - (n 2);}} finally {// 设置 sizeCtl 为 sc我们就当是 12 吧sizeCtl sc;}break;}}return tab;
}4.5、链表转红黑树: treeifyBin
前面我们在 put 源码分析也说过treeifyBin 不一定就会进行红黑树转换也可能是仅仅做数组扩容。我们还是进行源码分析吧。
private final void treeifyBin(NodeK,V[] tab, int index) {NodeK,V b; int n, sc;if (tab ! null) {// MIN_TREEIFY_CAPACITY 为 64// 所以如果数组长度小于 64 的时候其实也就是 32 或者 16 或者更小的时候会进行数组扩容if ((n tab.length) MIN_TREEIFY_CAPACITY)// 后面我们再详细分析这个方法tryPresize(n 1);// b 是头节点else if ((b tabAt(tab, index)) ! null b.hash 0) {// 加锁synchronized (b) {if (tabAt(tab, index) b) {// 下面就是遍历链表建立一颗红黑树TreeNodeK,V hd null, tl null;for (NodeK,V e b; e ! null; e e.next) {TreeNodeK,V p new TreeNodeK,V(e.hash, e.key, e.val,null, null);if ((p.prev tl) null)hd p;elsetl.next p;tl p;}// 将红黑树设置到数组相应位置中setTabAt(tab, index, new TreeBinK,V(hd));}}}}
}4.6、扩容: tryPresize
如果说 Java8 ConcurrentHashMap 的源码不简单那么说的就是扩容操作和迁移操作。
这个方法要完完全全看懂还需要看之后的 transfer 方法读者应该提前知道这点。
这里的扩容也是做翻倍扩容的扩容后数组容量为原来的 2 倍。
// 首先要说明的是方法参数 size 传进来的时候就已经翻了倍了
private final void tryPresize(int size) {// c: size 的 1.5 倍再加 1再往上取最近的 2 的 n 次方。int c (size (MAXIMUM_CAPACITY 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size (size 1) 1);int sc;while ((sc sizeCtl) 0) {NodeK,V[] tab table; int n;// 这个 if 分支和之前说的初始化数组的代码基本上是一样的在这里我们可以不用管这块代码if (tab null || (n tab.length) 0) {n (sc c) ? sc : c;if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if (table tab) {SuppressWarnings(unchecked)NodeK,V[] nt (NodeK,V[])new Node?,?[n];table nt;sc n - (n 2); // 0.75 * n}} finally {sizeCtl sc;}}}else if (c sc || n MAXIMUM_CAPACITY)break;else if (tab table) {// 我没看懂 rs 的真正含义是什么不过也关系不大int rs resizeStamp(n);if (sc 0) {NodeK,V[] nt;if ((sc RESIZE_STAMP_SHIFT) ! rs || sc rs 1 ||sc rs MAX_RESIZERS || (nt nextTable) null ||transferIndex 0)break;// 2. 用 CAS 将 sizeCtl 加 1然后执行 transfer 方法// 此时 nextTab 不为 nullif (U.compareAndSwapInt(this, SIZECTL, sc, sc 1))transfer(tab, nt);}// 1. 将 sizeCtl 设置为 (rs RESIZE_STAMP_SHIFT) 2)// 我是没看懂这个值真正的意义是什么? 不过可以计算出来的是结果是一个比较大的负数// 调用 transfer 方法此时 nextTab 参数为 nullelse if (U.compareAndSwapInt(this, SIZECTL, sc,(rs RESIZE_STAMP_SHIFT) 2))transfer(tab, null);}}
}这个方法的核心在于 sizeCtl 值的操作首先将其设置为一个负数然后执行 transfer(tab, null)再下一个循环将 sizeCtl 加 1并执行 transfer(tab, nt)之后可能是继续 sizeCtl 加 1并执行 transfer(tab, nt)。
所以可能的操作就是执行 1 次 transfer(tab, null) 多次 transfer(tab, nt)这里怎么结束循环的需要看完 transfer 源码才清楚。
4.7、数据迁移: transfer
下面这个方法有点长将原来的 tab 数组的元素迁移到新的 nextTab 数组中。
虽然我们之前说的 tryPresize 方法中多次调用 transfer 不涉及多线程但是这个 transfer 方法可以在其他地方被调用典型地我们之前在说 put 方法的时候就说过了请往上看 put 方法是不是有个地方调用了 helpTransfer 方法helpTransfer 方法会调用 transfer 方法的。
此方法支持多线程执行外围调用此方法的时候会保证第一个发起数据迁移的线程nextTab 参数为 null之后再调用此方法的时候nextTab 不会为 null。
阅读源码之前先要理解并发操作的机制。原数组长度为 n所以我们有 n 个迁移任务让每个线程每次负责一个小任务是最简单的每做完一个任务再检测是否有其他没做完的任务帮助迁移就可以了而 Doug Lea 使用了一个 stride简单理解就是步长每个线程每次负责迁移其中的一部分如每次迁移 16 个小任务。所以我们就需要一个全局的调度者来安排哪个线程执行哪几个任务这个就是属性 transferIndex 的作用。
第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置然后从后往前的 stride 个任务属于第一个线程然后将 transferIndex 指向新的位置再往前的 stride 个任务属于第二个线程依此类推。当然这里说的第二个线程不是真的一定指代了第二个线程也可以是同一个线程这个读者应该能理解吧。其实就是将一个大的迁移任务分为了一个个任务包。
private final void transfer(NodeK,V[] tab, NodeK,V[] nextTab) {int n tab.length, stride;// stride 在单核下直接等于 n多核模式下为 (n3)/NCPU最小值是 16// stride 可以理解为”步长“有 n 个位置是需要进行迁移的// 将这 n 个任务分为多个任务包每个任务包有 stride 个任务if ((stride (NCPU 1) ? (n 3) / NCPU : n) MIN_TRANSFER_STRIDE)stride MIN_TRANSFER_STRIDE; // subdivide range// 如果 nextTab 为 null先进行一次初始化// 前面我们说了外围会保证第一个发起迁移的线程调用此方法时参数 nextTab 为 null// 之后参与迁移的线程调用此方法时nextTab 不会为 nullif (nextTab null) {try {// 容量翻倍NodeK,V[] nt (NodeK,V[])new Node?,?[n 1];nextTab nt;} catch (Throwable ex) { // try to cope with OOMEsizeCtl Integer.MAX_VALUE;return;}// nextTable 是 ConcurrentHashMap 中的属性nextTable nextTab;// transferIndex 也是 ConcurrentHashMap 的属性用于控制迁移的位置transferIndex n;}int nextn nextTab.length;// ForwardingNode 翻译过来就是正在被迁移的 Node// 这个构造方法会生成一个Nodekey、value 和 next 都为 null关键是 hash 为 MOVED// 后面我们会看到原数组中位置 i 处的节点完成迁移工作后// 就会将位置 i 处设置为这个 ForwardingNode用来告诉其他线程该位置已经处理过了// 所以它其实相当于是一个标志。ForwardingNodeK,V fwd new ForwardingNodeK,V(nextTab);// advance 指的是做完了一个位置的迁移工作可以准备做下一个位置的了boolean advance true;boolean finishing false; // to ensure sweep before committing nextTab/** 下面这个 for 循环最难理解的在前面而要看懂它们应该先看懂后面的然后再倒回来看* */// i 是位置索引bound 是边界注意是从后往前for (int i 0, bound 0;;) {NodeK,V f; int fh;// 下面这个 while 真的是不好理解// advance 为 true 表示可以进行下一个位置的迁移了// 简单理解结局: i 指向了 transferIndexbound 指向了 transferIndex-stridewhile (advance) {int nextIndex, nextBound;if (--i bound || finishing)advance false;// 将 transferIndex 值赋给 nextIndex// 这里 transferIndex 一旦小于等于 0说明原数组的所有位置都有相应的线程去处理了else if ((nextIndex transferIndex) 0) {i -1;advance false;}else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound (nextIndex stride ?nextIndex - stride : 0))) {// 看括号中的代码nextBound 是这次迁移任务的边界注意是从后往前bound nextBound;i nextIndex - 1;advance false;}}if (i 0 || i n || i n nextn) {int sc;if (finishing) {// 所有的迁移操作已经完成nextTable null;// 将新的 nextTab 赋值给 table 属性完成迁移table nextTab;// 重新计算 sizeCtl: n 是原数组长度所以 sizeCtl 得出的值将是新数组长度的 0.75 倍sizeCtl (n 1) - (n 1);return;}// 之前我们说过sizeCtl 在迁移前会设置为 (rs RESIZE_STAMP_SHIFT) 2// 然后每有一个线程参与迁移就会将 sizeCtl 加 1// 这里使用 CAS 操作对 sizeCtl 进行减 1代表做完了属于自己的任务if (U.compareAndSwapInt(this, SIZECTL, sc sizeCtl, sc - 1)) {// 任务结束方法退出if ((sc - 2) ! resizeStamp(n) RESIZE_STAMP_SHIFT)return;// 到这里说明 (sc - 2) resizeStamp(n) RESIZE_STAMP_SHIFT// 也就是说所有的迁移任务都做完了也就会进入到上面的 if(finishing){} 分支了finishing advance true;i n; // recheck before commit}}// 如果位置 i 处是空的没有任何节点那么放入刚刚初始化的 ForwardingNode ”空节点“else if ((f tabAt(tab, i)) null)advance casTabAt(tab, i, null, fwd);// 该位置处是一个 ForwardingNode代表该位置已经迁移过了else if ((fh f.hash) MOVED)advance true; // already processedelse {// 对数组该位置处的结点加锁开始处理数组该位置处的迁移工作synchronized (f) {if (tabAt(tab, i) f) {NodeK,V ln, hn;// 头节点的 hash 大于 0说明是链表的 Node 节点if (fh 0) {// 下面这一块和 Java7 中的 ConcurrentHashMap 迁移是差不多的// 需要将链表一分为二// 找到原链表中的 lastRun然后 lastRun 及其之后的节点是一起进行迁移的// lastRun 之前的节点需要进行克隆然后分到两个链表中int runBit fh n;NodeK,V lastRun f;for (NodeK,V 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 (NodeK,V 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 NodeK,V(ph, pk, pv, ln);elsehn new NodeK,V(ph, pk, pv, hn);}// 其中的一个链表放在新数组的位置 isetTabAt(nextTab, i, ln);// 另一个链表放在新数组的位置 insetTabAt(nextTab, i n, hn);// 将原数组该位置处设置为 fwd代表该位置已经处理完毕// 其他线程一旦看到该位置的 hash 值为 MOVED就不会进行迁移了setTabAt(tab, i, fwd);// advance 设置为 true代表该位置已经迁移完毕advance true;}else if (f instanceof TreeBin) {// 红黑树的迁移TreeBinK,V t (TreeBinK,V)f;TreeNodeK,V lo null, loTail null;TreeNodeK,V hi null, hiTail null;int lc 0, hc 0;for (NodeK,V e t.first; e ! null; e e.next) {int h e.hash;TreeNodeK,V p new TreeNodeK,V(h, e.key, e.val, null, null);if ((h n) 0) {if ((p.prev loTail) null)lo p;elseloTail.next p;loTail p;lc;}else {if ((p.prev hiTail) null)hi p;elsehiTail.next p;hiTail p;hc;}}// 如果一分为二后节点数小于等于6那么将红黑树转换回链表ln (lc UNTREEIFY_THRESHOLD) ? untreeify(lo) :(hc ! 0) ? new TreeBinK,V(lo) : t;hn (hc UNTREEIFY_THRESHOLD) ? untreeify(hi) :(lc ! 0) ? new TreeBinK,V(hi) : t;// 将 ln 放置在新数组的位置 isetTabAt(nextTab, i, ln);// 将 hn 放置在新数组的位置 insetTabAt(nextTab, i n, hn);// 将原数组该位置处设置为 fwd代表该位置已经处理完毕// 其他线程一旦看到该位置的 hash 值为 MOVED就不会进行迁移了setTabAt(tab, i, fwd);// advance 设置为 true代表该位置已经迁移完毕advance true;}}}}}
}说到底transfer 这个方法并没有实现所有的迁移任务每次调用这个方法只实现了 transferIndex 往前 stride 个位置的迁移工作其他的需要由外围来控制。
这个时候再回去仔细看 tryPresize 方法可能就会更加清晰一些了。
4.8、get 过程分析
get 方法从来都是最简单的这里也不例外:
计算 hash 值根据 hash 值找到数组对应位置: (n - 1) h根据该位置处结点性质进行相应查找 如果该位置为 null那么直接返回 null 就可以了如果该位置处的节点刚好就是我们需要的返回该节点的值即可如果该位置节点的 hash 值小于 0说明正在扩容或者是红黑树后面我们再介绍 find 方法如果以上 3 条都不满足那就是链表进行遍历比对即可
public V get(Object key) {NodeK,V[] tab; NodeK,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;}// 如果头节点的 hash 小于 0说明 正在扩容或者该位置是红黑树else if (eh 0)// 参考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k)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;
}简单说一句此方法的大部分内容都很简单只有正好碰到扩容的情况ForwardingNode.find(int h, Object k) 稍微复杂一些不过在了解了数据迁移的过程后这个也就不难了所以限于篇幅这里也不展开说了。
5、对比总结 HashTable : 使用了synchronized关键字对put等操作进行加锁; ConcurrentHashMap JDK1.7: 使用分段锁机制实现; ConcurrentHashMap JDK1.8: 则使用数组链表红黑树数据结构和CAS原子操作实现; 不同版本的ConcurrentHashMap区别
版本数据结构线程安全实现方式jdk1.7Segment 数组和多个 HashEntry 组成分段锁(Segment继承了ReentrantLock)jdk1.8数组链表红黑树CAS Synchronized
6、ConcurrentHashMap使用案例
这个案例是来演示session会话续期策略的倘若是用户再次登录系统会更新其sessionKey的超期时间。renewal为延续session会话使用的数据结构为ReentrantReadWriteLock读写锁以及ConcurrentHashMap来保存会话信息。
// redis session读取会话续期
public class RedisSessionManager {/*** 会话map ReentrantReadWriteLock 读锁共享 写锁互斥*/private final ReadWriteLock rwl new ReentrantReadWriteLock();private ConcurrentMapString, Long sessionKeyMap new ConcurrentHashMap();private void addToReNewMap(String id, long lastAccessAt) {rwl.readLock().lock();try {if (sessionKeyMap.size() 102400) {sessionKeyMap.put(id, lastAccessAt);}} finally {rwl.readLock().unlock();}}public void renewal() {MapString, Long localSessionKeyMap;rwl.writeLock().lock();try {localSessionKeyMap sessionKeyMap;sessionKeyMap new ConcurrentHashMap();} finally {rwl.writeLock().unlock();}localSessionKeyMap Iters.nullToEmpty(localSessionKeyMap);if (localSessionKeyMap.size() 0) {LOGGER.warn(renewal session size {}, localSessionKeyMap.size());localSessionKeyMap.forEach(this.expirationPolicy::onExpirationUpdated);}}
}ConcurrentHashMap在项目中的使用
todo
Action1讲讲为什么ConcurrentHashMap是并发安全的吧既然有锁怎么去统计size呢
xxx从1.7讲到1.8,从分段锁讲到cassync…size的统计1.7和1.8也有区别
Action2ConcurrentHashMap为啥线程安全呢(10分)
put操作采用 CASsynchronized 实现并发插入或更新操作
参考文章 Java 并发工具包 java.util.concurrent 用户指南 Java Concurrency and Multithreading Tutorial 并发容器之ConcurrentHashMap(JDK 1.8版本) Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析 Java并发包concurrent——ConcurrentHashMap 【JUC】JDK1.8源码分析之ConcurrentHashMap 探索jdk8之ConcurrentHashMap 的实现机制 面试被问到 ConcurrentHashMap答不出 看这一篇就够了
