ConcurrentHashMap 和 HashMap 思路是差很少的,可是由于它支持并发操做,因此要复杂一些。java
整个 ConcurrentHashMap 由一个个 Segment 组成,Segment 表明”部分“或”一段“的意思,因此不少地方都会将其描述为分段锁。注意,行文中,我不少地方用了“槽”来表明一个 segment。node
简单理解就是,ConcurrentHashMap 是一个 Segment 数组,Segment 经过继承 ReentrantLock 来进行加锁,因此每次须要加锁的操做锁住的是一个 segment,这样只要保证每一个 Segment 是线程安全的,也就实现了全局的线程安全。数组
concurrencyLevel:并行级别、并发数、Segment 数,怎么翻译不重要,理解它。默认是 16,也就是说 ConcurrentHashMap 有 16 个 Segments,因此理论上,这个时候,最多能够同时支持 16 个线程并发写,只要它们的操做分别分布在不一样的 Segment 上。这个值能够在初始化的时候设置为其余值,可是一旦初始化之后,它是不能够扩容的。安全
再具体到每一个 Segment 内部,其实每一个 Segment 很像以前介绍的 HashMap,不过它要保证线程安全,因此处理起来要麻烦些。多线程
initialCapacity:初始容量,这个值指的是整个 ConcurrentHashMap 的初始容量,实际操做的时候须要平均分给每一个 Segment。并发
loadFactor:负载因子,以前咱们说了,Segment 数组不能够扩容,因此这个负载因子是给每一个 Segment 内部使用的。ssh
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 arguments int sshift = 0; int ssize = 1; // 计算并行级别 ssize,由于要保持并行级别是 2 的 n 次方 while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } // 咱们这里先不要那么烧脑,用默认值,concurrencyLevel 为 16,sshift 为 4 // 那么计算出 segmentShift 为 28,segmentMask 为 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] Segment<K,V> s0 = new Segment<K,V>(loadFactor, (int)(cap * loadFactor), (HashEntry<K,V>[])new HashEntry[cap]); Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize]; // 往数组写入 segment[0] UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] this.segments = ss; }
初始化完成,咱们获得了一个 Segment 数组。ide
咱们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的,那么初始化完成后:函数
咱们先看 put 的主流程,对于其中的一些关键细节操做,后面会进行详细介绍。oop
public V put(K key, V value) { Segment<K,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 = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = 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 的独占锁 // 先看主流程,后面还会具体介绍这部份内容 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { // 这个是 segment 内部的数组 HashEntry<K,V>[] tab = table; // 再利用 hash 值,求应该放置的数组下标 int index = (tab.length - 1) & hash; // first 是数组该位置处的链表的表头 HashEntry<K,V> first = entryAt(tab, index); // 下面这串 for 循环虽然很长,不过也很好理解,想一想该位置没有任何元素和已经存在一个链表这两种状况 for (HashEntry<K,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); else node = new HashEntry<K,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 操做就结束了,接下来,咱们说一说其中几步关键的操做。
ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0],对于其余槽来讲,在插入第一个值的时候进行初始化。
这里须要考虑并发,由于极可能会有多个线程同时进来初始化同一个槽 segment[k],不过只要有一个成功了就能够。
private Segment<K,V> ensureSegment(int k) { final Segment<K,V>[] ss = this.segments; long u = (k << SSHIFT) + SBASE; // raw offset Segment<K,V> seg; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 这里看到为何以前要初始化 segment[0] 了, // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k] // 为何要用“当前”,由于 segment[0] 可能早就扩容过了 Segment<K,V> proto = ss[0]; int cap = proto.table.length; float lf = proto.loadFactor; int threshold = (int)(cap * lf); // 初始化 segment[k] 内部的数组 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 再次检查一遍该槽是否被其余线程初始化了。 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); // 使用 while 循环,内部用 CAS,当前线程成功设值或其余线程成功设值后,退出 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }
总的来讲,ensureSegment(int k) 比较简单,对于并发操做使用 CAS 进行控制。
我没搞懂这里为何要搞一个 while 循环,CAS 失败不就表明有其余线程成功了吗,为何要再进行判断?
感谢评论区的李子木,若是当前线程 CAS 失败,这里的 while 循环是为了将 seg 赋值返回。
前面咱们看到,在往某个 segment 中 put 的时候,首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value),也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁,若是失败,那么进入到 scanAndLockForPut 这个方法来获取锁。
下面咱们来具体分析这个方法中是怎么控制加锁的。
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; // negative while locating node // 循环获取锁 while (!tryLock()) { HashEntry<K,V> f; // to recheck first below if (retries < 0) { if (e == null) { if (node == null) // speculatively create node // 进到这里说明数组该位置的链表是空的,没有任何元素 // 固然,进到这里的另外一个缘由是 tryLock() 失败,因此该槽存在并发,不必定是该位置 node = new HashEntry<K,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 changed retries = -1; } } return node; }
这个方法有两个出口,一个是 tryLock() 成功了,循环终止,另外一个就是重试次数超过了 MAX_SCAN_RETRIES,进到 lock() 方法,此方法会阻塞等待,直到成功拿到独占锁。
这个方法就是看似复杂,可是其实就是作了一件事,那就是获取该 segment 的独占锁,若是须要的话顺便实例化了一下 node。
重复一下,segment 数组不能扩容,扩容是 segment 数组某个位置内部的数组 HashEntry[] 进行扩容,扩容后,容量为原来的 2 倍。
首先,咱们要回顾一下触发扩容的地方,put 的时候,若是判断该值的插入会致使该 segment 的元素个数超过阈值,那么先进行扩容,再插值,读者这个时候能够回去 put 方法看一眼。
该方法不须要考虑并发,由于到这里的时候,是持有该 segment 的独占锁的。
// 方法参数上的 node 是此次扩容后,须要添加到新的数组中的数据。 private void rehash(HashEntry<K,V> node) { HashEntry<K,V>[] oldTable = table; int oldCapacity = oldTable.length; // 2 倍 int newCapacity = oldCapacity << 1; threshold = (int)(newCapacity * loadFactor); // 建立新数组 HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 新的掩码,如从 16 扩容到 32,那么 sizeMask 为 31,对应二进制 ‘000...00011111’ int sizeMask = newCapacity - 1; // 遍历原数组,老套路,将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置 for (int i = 0; i < oldCapacity ; i++) { // e 是链表的第一个元素 HashEntry<K,V> e = oldTable[i]; if (e != null) { HashEntry<K,V> next = e.next; // 计算应该放置在新数组中的位置, // 假设原数组长度为 16,e 在 oldTable[3] 处,那么 idx 只多是 3 或者是 3 + 16 = 19 int idx = e.hash & sizeMask; if (next == null) // 该位置处只有一个元素,那比较好办 newTable[idx] = e; else { // Reuse consecutive sequence at same slot // e 是链表表头 HashEntry<K,V> lastRun = e; // idx 是当前链表的头结点 e 的新位置 int lastIdx = idx; // 下面这个 for 循环会找到一个 lastRun 节点,这个节点以后的全部元素是将要放到一块儿的 for (HashEntry<K,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 (HashEntry<K,V> p = e; p != lastRun; p = p.next) { V v = p.value; int h = p.hash; int k = h & sizeMask; HashEntry<K,V> n = newTable[k]; newTable[k] = new HashEntry<K,V>(h, p.key, v, n); } } } } // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部 int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; }
这里的扩容比以前的 HashMap 要复杂一些,代码难懂一点。上面有两个挨着的 for 循环,第一个 for 有什么用呢?
仔细一看发现,若是没有第一个 for 循环,也是能够工做的,可是,这个 for 循环下来,若是 lastRun 的后面还有比较多的节点,那么此次就是值得的。由于咱们只须要克隆 lastRun 前面的节点,后面的一串节点跟着 lastRun 走就是了,不须要作任何操做。
我以为 Doug Lea 的这个想法也是挺有意思的,不过比较坏的状况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素,那么此次遍历就有点浪费了。不过 Doug Lea 也说了,根据统计,若是使用默认的阈值,大约只有 1/6 的节点须要克隆。
相对于 put 来讲,get 真的不要太简单。
public V get(Object key) { Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; // 1. hash 值 int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 2. 根据 hash 找到对应的 segment if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { // 3. 找到segment 内部数组相应位置的链表,遍历 for (HashEntry<K,V> e = (HashEntry<K,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; }
如今咱们已经说完了 put 过程和 get 过程,咱们能够看到 get 过程当中是没有加锁的,那天然咱们就须要去考虑并发问题。
添加节点的操做 put 和删除节点的操做 remove 都是要加 segment 上的独占锁的,因此它们之间天然不会有问题,咱们须要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操做。
remove 操做咱们没有分析源码,因此这里说的读者感兴趣的话仍是须要到源码中去求实一下的。
get 操做须要遍历链表,可是 remove 操做会"破坏"链表。
若是 remove 破坏的节点 get 操做已通过去了,那么这里不存在任何问题。
若是 remove 先破坏了一个节点,分两种状况考虑。 一、若是此节点是头结点,那么须要将头结点的 next 设置为数组该位置的元素,table 虽然使用了 volatile 修饰,可是 volatile 并不能提供数组内部操做的可见性保证,因此源码中使用了 UNSAFE 来操做数组,请看方法 setEntryAt。二、若是要删除的节点不是头结点,它会将要删除节点的后继节点接到前驱节点中,这里的并发保证就是 next 属性是 volatile 的。
Java7 中实现的 ConcurrentHashMap 说实话仍是比较复杂的,Java8 对 ConcurrentHashMap 进行了比较大的改动。建议读者能够参考 Java8 中 HashMap 相对于 Java7 HashMap 的改动,对于 ConcurrentHashMap,Java8 也引入了红黑树。
说实话,Java8 ConcurrentHashMap 源码真心不简单,最难的在于扩容,数据迁移操做不容易看懂。
咱们先用一个示意图来描述下其结构:
结构上和 Java8 的 HashMap 基本上同样,不过它要保证线程安全性,因此在源码上确实要复杂一些。
// 这构造函数里,什么都不干 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; }
这个初始化方法有点意思,经过提供初始容量,计算了 sizeCtl,sizeCtl = 【 (1.5 * initialCapacity + 1),而后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10,那么获得 sizeCtl 为 16,若是 initialCapacity 为 11,获得 sizeCtl 为 32。
sizeCtl 这个属性使用的场景不少,不过只要跟着文章的思路来,就不会被它搞晕了。
若是你爱折腾,也能够看下另外一个有三个参数的构造方法,这里我就不说了,大部分时候,咱们会使用无参构造函数进行实例化,咱们也按照这个思路来进行源码分析吧。
仔细地一行一行代码看下去:
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 (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // 若是数组"空",进行数组初始化 if (tab == null || (n = tab.length) == 0) // 初始化数组,后面会详细介绍 tab = initTable(); // 找该 hash 值对应的数组下标,获得第一个节点 f else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 若是数组该位置为空, // 用一次 CAS 操做将这个新值放入其中便可,这个 put 操做差很少就结束了,能够拉到最后面了 // 若是 CAS 失败,那就是有并发操做,进到下一个循环就行了 if (casTabAt(tab, i, null, new Node<K,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 (Node<K,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; } // 到了链表的最末端,将这个新值放到链表的最后面 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) { // 判断是否要将链表转换为红黑树,临界值和 HashMap 同样,也是 8 if (binCount >= TREEIFY_THRESHOLD) // 这个方法和 HashMap 中稍微有一点点不一样,那就是它不是必定会进行红黑树转换, // 若是当前数组的长度小于 64,那么会选择进行数组扩容,而不是转换为红黑树 // 具体源码咱们就不看了,扩容部分后面说 treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // addCount(1L, binCount); return null; }
put 的主流程看完了,可是至少留下了几个问题,第一个是初始化,第二个是扩容,第三个是帮助数据迁移,这些咱们都会在后面进行一一介绍。
这个比较简单,主要就是初始化一个合适大小的数组,而后会设置 sizeCtl。
初始化方法中的并发问题是经过对 sizeCtl 进行一个 CAS 操做来控制的。
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 // CAS 一下,将 sizeCtl 设置为 -1,表明抢到了锁 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { // DEFAULT_CAPACITY 默认初始容量是 16 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 初始化数组,长度为 16 或初始化时提供的长度 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 将这个数组赋值给 table,table 是 volatile 的 table = tab = nt; // 若是 n 为 16 的话,那么这里 sc = 12 // 其实就是 0.75 * n sc = n - (n >>> 2); } } finally { // 设置 sizeCtl 为 sc,咱们就当是 12 吧 sizeCtl = sc; } break; } } return tab; }
前面咱们在 put 源码分析也说过,treeifyBin 不必定就会进行红黑树转换,也多是仅仅作数组扩容。咱们仍是进行源码分析吧。
private final void treeifyBin(Node<K,V>[] tab, int index) { Node<K,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) { // 下面就是遍历链表,创建一颗红黑树 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)); } } } } }
若是说 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) { Node<K,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") Node<K,V>[] nt = (Node<K,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) { Node<K,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 不为 null if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } // 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2) // 我是没看懂这个值真正的意义是什么?不过能够计算出来的是,结果是一个比较大的负数 // 调用 transfer 方法,此时 nextTab 参数为 null else 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 源码才清楚。
下面这个方法很点长,将原来的 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(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; // stride 在单核下直接等于 n,多核模式下为 (n>>>3)/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 不会为 null if (nextTab == null) { try { // 容量翻倍 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 是 ConcurrentHashMap 中的属性 nextTable = nextTab; // transferIndex 也是 ConcurrentHashMap 的属性,用于控制迁移的位置 transferIndex = n; } int nextn = nextTab.length; // ForwardingNode 翻译过来就是正在被迁移的 Node // 这个构造方法会生成一个Node,key、value 和 next 都为 null,关键是 hash 为 MOVED // 后面咱们会看到,原数组中位置 i 处的节点完成迁移工做后, // 就会将位置 i 处设置为这个 ForwardingNode,用来告诉其余线程该位置已经处理过了 // 因此它其实至关因而一个标志。 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // advance 指的是作完了一个位置的迁移工做,能够准备作下一个位置的了 boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab /* * 下面这个 for 循环,最难理解的在前面,而要看懂它们,应该先看懂后面的,而后再倒回来看 * */ // i 是位置索引,bound 是边界,注意是从后往前 for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; // 下面这个 while 真的是很差理解 // advance 为 true 表示能够进行下一个位置的迁移了 // 简单理解结局:i 指向了 transferIndex,bound 指向了 transferIndex-stride while (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 processed else { // 对数组该位置处的结点加锁,开始处理数组该位置处的迁移工做 synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; // 头结点的 hash 大于 0,说明是链表的 Node 节点 if (fh >= 0) { // 下面这一块和 Java7 中的 ConcurrentHashMap 迁移是差很少的, // 须要将链表一分为二, // 找到原链表中的 lastRun,而后 lastRun 及其以后的节点是一块儿进行迁移的 // lastRun 以前的节点须要进行克隆,而后分到两个链表中 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; } } 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; if ((ph & n) == 0) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); } // 其中的一个链表放在新数组的位置 i setTabAt(nextTab, i, ln); // 另外一个链表放在新数组的位置 i+n setTabAt(nextTab, i + n, hn); // 将原数组该位置处设置为 fwd,表明该位置已经处理完毕, // 其余线程一旦看到该位置的 hash 值为 MOVED,就不会进行迁移了 setTabAt(tab, i, fwd); // advance 设置为 true,表明该位置已经迁移完毕 advance = true; } else if (f instanceof TreeBin) { // 红黑树的迁移 TreeBin<K,V> t = (TreeBin<K,V>)f; 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; } } // 若是一分为二后,节点数少于 8,那么将红黑树转换回链表 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; // 将 ln 放置在新数组的位置 i setTabAt(nextTab, i, ln); // 将 hn 放置在新数组的位置 i+n setTabAt(nextTab, i + n, hn); // 将原数组该位置处设置为 fwd,表明该位置已经处理完毕, // 其余线程一旦看到该位置的 hash 值为 MOVED,就不会进行迁移了 setTabAt(tab, i, fwd); // advance 设置为 true,表明该位置已经迁移完毕 advance = true; } } } } } }
说到底,transfer 这个方法并无实现全部的迁移任务,每次调用这个方法只实现了 transferIndex 往前 stride 个位置的迁移工做,其余的须要由外围来控制。
这个时候,再回去仔细看 tryPresize 方法可能就会更加清晰一些了。
get 方法历来都是最简单的,这里也不例外:
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; } // 若是头结点的 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) 稍微复杂一些,不过在了解了数据迁移的过程后,这个也就不难了,因此限于篇幅这里也不展开说了。
来源: javadoop