ConcurrentHashMap源码解析
Posted ouhaitao
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ConcurrentHashMap源码解析相关的知识，希望对你有一定的参考价值。
首先看看CHM的重要成员变量：
1 public class ConcurrentHashMap<K,V> extends AbstractMap<K,V> 2 implements ConcurrentMap<K,V>, Serializable { 3 // table最大容量，为2的幂次方 4 private static final int MAXIMUM_CAPACITY = 1 << 30; 5 // 默认table初始容量大小 6 private static final int DEFAULT_CAPACITY = 16; 7 // 默认支持并发更新的线程数量 8 private static final int DEFAULT_CONCURRENCY_LEVEL = 16; 9 // table的负载因子 10 private static final float LOAD_FACTOR = 0.75f; 11 // 链表转换为红黑树的节点数阈值，超过这个值，链表转换为红黑树 12 static final int TREEIFY_THRESHOLD = 8; 13 // 在扩容期间，由红黑树转换为链表的阈值，小于这个值，resize期间红黑树就会转为链表 14 static final int UNTREEIFY_THRESHOLD = 6; 15 // 转为红黑树时，红黑树中节点的最小个数 16 static final int MIN_TREEIFY_CAPACITY = 64; 17 // 扩容时，并发转移节点(transfer方法)时，每次转移的最小节点数 18 private static final int MIN_TRANSFER_STRIDE = 16; 19 20 // 以下常量定义了特定节点类hash字段的值 21 static final int MOVED = -1; // ForwardingNode类对象的hash值 22 static final int TREEBIN = -2; // TreeBin类对象的hash值 23 static final int RESERVED = -3; // ReservationNode类对象的hash值 24 static final int HASH_BITS = 0x7fffffff; // 普通Node节点的hash初始值 25 26 // table数组 27 transient volatile Node<K,V>[] table; 28 // 扩容时，下一个容量大小的talbe，用于将原table元素移动到这个table中 29 private transient volatile Node<K,V>[] nextTable; 30 // 基础计数器，计算size 31 private transient volatile long baseCount; 32 // table初始容量大小以及扩容容量大小的参数，也用于标识table的状态 33 // 其有几个值来代表也用来代表table的状态: 34 // -1 ：标识table正在初始化 35 // - N : 标识table正在进行扩容，并且有N - 1个线程一起在进行扩容（实际上在使用过程中N仅代表sizeCtl的低16位，我会在代码分析中解释） 36 // 正数：初始table的大小，如果值大于初始容量大小，则表示扩容后的table大小。 37 private transient volatile int sizeCtl; 38 // 扩容时，下一个节点转移的bucket索引下标 39 private transient volatile int transferIndex; 40 // 一种自旋锁，是专为防止多处理器并发而引入的一种锁，用于创建CounterCells时使用， 41 // 主要用于size方法计数时，有并发线程插入而计算修改的节点数量， 42 // 这个数量会与baseCount计数器汇总后得出size的结果。 43 private transient volatile int cellsBusy; 44 // 主要用于size方法计数时，有并发线程插入而计算修改的节点数量， 45 // 这个数量会与baseCount计数器汇总后得出size的结果。 46 private transient volatile CounterCell[] counterCells; 47 // 其他省略 48 }
Put方法
1 final V putVal(K key, V value, boolean onlyIfAbsent) { 2 if (key == null || value == null) throw new NullPointerException(); 3 // 计算hash值 4 int hash = spread(key.hashCode()); 5 int binCount = 0; 6 for (Node<K,V>[] tab = table;;) { 7 Node<K,V> f; int n, i, fh; 8 if (tab == null || (n = tab.length) == 0) 9 // 初始化table 10 tab = initTable(); 11 // 对应的位置上没有节点 12 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { 13 // CAS操作，将当前value插入对应的位置 14 if (casTabAt(tab, i, null, 15 new Node<K,V>(hash, key, value, null))) 16 break; // no lock when adding to empty bin 17 } 18 // 如果对应位置的节点正在进行扩容 19 else if ((fh = f.hash) == MOVED) 20 // 帮助扩容 21 tab = helpTransfer(tab, f); 22 else { 23 V oldVal = null; 24 // 使用对应位置的节点作为锁对象进行加锁 25 // 1.使其他在对应位置也是当前位置的节点插入操作阻塞 26 // 2.扩容线程也会使用数组上的节点作为锁对象锁住，所以对当前节点上的扩容与插入操作是互斥的 27 synchronized (f) { 28 // 判断tab表是否已经被改变 29 if (tabAt(tab, i) == f) { 30 // 当前节点是普通节点 31 if (fh >= 0) { 32 // 在addCount()方法中作为扩容校验字段 33 binCount = 1; 34 for (Node<K,V> e = f;; ++binCount) { 35 K ek; 36 // key相同 37 if (e.hash == hash && 38 ((ek = e.key) == key || 39 (ek != null && key.equals(ek)))) { 40 oldVal = e.val; 41 // 更新值 42 if (!onlyIfAbsent) 43 e.val = value; 44 break; 45 } 46 // 遍历链表，插入值 47 Node<K,V> pred = e; 48 if ((e = e.next) == null) { 49 pred.next = new Node<K,V>(hash, key, 50 value, null); 51 break; 52 } 53 } 54 } 55 // 如果是红黑树节点 56 else if (f instanceof TreeBin) { 57 Node<K,V> p; 58 binCount = 2; 59 // 向树中插入值 60 if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, 61 value)) != null) { 62 oldVal = p.val; 63 if (!onlyIfAbsent) 64 p.val = value; 65 } 66 } 67 } 68 } 69 if (binCount != 0) { 70 // 判断链表长度是否超过了阈值 71 if (binCount >= TREEIFY_THRESHOLD) 72 // 转成红黑树 73 treeifyBin(tab, i); 74 if (oldVal != null) 75 return oldVal; 76 break; 77 } 78 } 79 } 80 // size+1，并判断是否需要扩容 81 addCount(1L, binCount); 82 return null; 83 }
AddCount方法
1 private final void addCount(long x, int check) { 2 CounterCell[] as; long b, s; 3 if ((as = counterCells) != null ||// counterCells已经初始化 4 !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)// 未获取到baseCount（表示存在竞争且当前线程竞争失败） 5 ) { 6 CounterCell a; long v; int m; 7 boolean uncontended = true; 8 if (as == null || (m = as.length - 1) < 0 || // counterCells未初始化 9 (a = as[ThreadLocalRandom.getProbe() & m]) == null ||// CounterCells已初始化且其中找到一个空位能够插入新的CounterCell 10 !(uncontended = 11 U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) // CounterCells中对应的位置已被占用，则更新对应的值，如果失败（表示存在竞争） 12 { 13 fullAddCount(x, uncontended);// 创建新的counterCell并插入countercells（如果counterCell未初始化，则将其初始化） 14 return; 15 } 16 // 以上条件都不成立表示counterCells已初始化且将新增的size成功更新到CounterCells中 17 // 1.check=0表示在数组上插入了一个节点 2.check=1表示在链表上插入了一个节点且该节点是链表的第二个节点 18 // 3.check=2表示在红黑树上插入了一个节点 4.check>1除了情况三之外就是才链表上插入了一个节点并且该节点是链表的第三个或者更后面的节点 19 // 这里的扩容检查条件为什么跟下面的不一样呢？无法理解，希望知道的朋友能讲解一下 20 if (check <= 1) 21 return; 22 s = sumCount(); 23 } 24 // check>=0表示需要检查扩容 25 if (check >= 0) { 26 Node<K,V>[] tab, nt; int n, sc; 27 while (s >= (long)(sc = sizeCtl) && (tab = table) != null && // sizeCtl此时表示扩容阈值，如果当前map大小达到阈值 28 (n = tab.length) < MAXIMUM_CAPACITY) {// 数组长度能够支持继续扩容 29 // 根据当前数组长度获得一个标识位，该值在32768~32798之间，用二进制表示为1000 0000 0000 0000 ~ 1000 0000 0001 1110 30 int rs = resizeStamp(n); 31 // sc<0表示当前正在进行扩容，第二个及以后的扩容线程执行到这里的时候会执行这里 32 if (sc < 0) { 33 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || // 校验sc，如果不等于表示sc异常 34 sc == rs + 1 || // 表示扩容结束，这是个BUG，应该是sc==rs<<RESIZE_STAMP_SHIFT + 1 35 sc == rs + MAX_RESIZERS || // 表示帮助线程数已经达到最大，这是个BUG，应该是sc==rs<<RESIZE_STAMP_SHIFT + MAX_RESIZERS 36 (nt = nextTable) == null || // 扩容结束 37 transferIndex <= 0)// 表示扩容区间已经被领取完毕，不需要新的线程帮助扩容（具体看transfer()方法） 38 break; 39 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) // sc+1表示有新的线程帮助扩容（这里加1在逻辑上是减1） 40 transfer(tab, nt); 41 } 42 // 第一个扩容的线程将执行这里，RESIZE_STAMP_SHIFT=16 43 // rs左移16位，以32768为例，最终得到的结果是 1000 0000 0000 0000 0000 0000 0000 0000，最高位是符号位，为1表示是负数，为0表示是整数 44 // rs左移16位的目的是为了得到一个负数，然后我们看一下sizeCtl上的注释“标识table正在进行扩容，并且有N - 1个线程一起在进行扩容” 45 // 所以如果只有一个线程在进行扩容，则sizeCtl值为-2，所以这里会加2 46 // 执行完这一步之后sizeCtl会变成一个非常大的负数，但是如果我们把高16位忽略掉或者看成符号位，只看低16位，那这个就符合注释的意思，sizeCtl变成-2 47 else if (U.compareAndSwapInt(this, SIZECTL, sc, 48 (rs << RESIZE_STAMP_SHIFT) + 2)) 49 // 具体扩容方法 50 transfer(tab, null); 51 s = sumCount(); 52 } 53 } 54 }
Transfer方法
1 private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { 2 int n = tab.length, stride; 3 // 将 length / 8 然后除以 CPU核心数。如果得到的结果小于 16，那么就使用 16。 4 // 这里的目的是让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象，如果桶较少的话，默认一个 CPU（一个线程）处理 16 个桶 5 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) 6 stride = MIN_TRANSFER_STRIDE; // subdivide range 细分范围 stridea：TODO 7 // 新的 table 尚未初始化 8 if (nextTab == null) { // initiating 9 try { 10 // 扩容 2 倍 11 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; 12 // 更新 13 nextTab = nt; 14 } catch (Throwable ex) { // try to cope with OOME 15 // 扩容失败， sizeCtl 使用 int 最大值。 16 sizeCtl = Integer.MAX_VALUE;// 在addCount()中会校验到sizeCtl异常，退出扩容 17 return;// 结束 18 } 19 // 更新成员变量 20 nextTable = nextTab; 21 // 更新转移下标，就是 老的 tab 的 length 22 transferIndex = n; 23 } 24 // 新 tab 的 length 25 int nextn = nextTab.length; 26 // 创建一个 fwd 节点，用于占位。当别的线程发现这个槽位中是 fwd 类型的节点，则跳过这个节点。 27 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); 28 // 推进标志位，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是 false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进 29 boolean advance = true; 30 // 完成状态，如果是 true，就结束此方法。 31 boolean finishing = false; // to ensure sweep before committing nextTab 32 // i 表示当前处理的下标，bound 表示当前线程可以处理的当前桶区间最小下标 33 for (int i = 0, bound = 0;;) { 34 Node<K,V> f; int fh; 35 // 如果当前线程可以向后推进；这个循环就是控制 i 递减。同时，每个线程都会进入这里取得自己需要转移的桶的区间 36 while (advance) { 37 int nextIndex, nextBound; 38 // 对 i 减一，判断是否大于等于 bound （正常情况下，如果大于 bound 不成立，说明该线程上次领取的任务已经完成了。那么，需要在下面继续领取任务） 39 // 如果对 i 减一大于等于 bound（还需要继续做任务），或者完成了，修改推进状态为 false，不能推进了。任务成功后修改推进状态为 true。 40 // 通常，第一次进入循环，i-- 这个判断会无法通过，从而走下面的 nextIndex 赋值操作（获取最新的转移下标）。其余情况都是：如果可以推进，将 i 减一，然后修改成不可推进。如果 i 对应的桶处理成功了，改成可以推进。 41 if (--i >= bound || finishing) 42 advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进 43 // 这里的目的是：1. 当一个线程进入时，会选取最新的转移下标。2. 当一个线程处理完自己的区间时，如果还有剩余区间的没有别的线程处理。再次获取区间。 44 else if ((nextIndex = transferIndex) <= 0) { 45 // 如果小于等于0，说明没有区间了 ，i 改成 -1，推进状态变成 false，不再推进，表示，扩容结束了，当前线程可以退出了 46 // 这个 -1 会在下面的 if 块里判断，从而进入完成状态判断 47 i = -1; 48 advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进 49 }// CAS 修改 transferIndex，即 length - 区间值，留下剩余的区间值供后面的线程使用 50 else if (U.compareAndSwapInt 51 (this, TRANSFERINDEX, nextIndex, 52 nextBound = (nextIndex > stride ? 53 nextIndex - stride : 0))) { 54 bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标 55 i = nextIndex - 1; // 初次对i 赋值，这个就是当前线程可以处理的当前区间的最大下标 56 advance = false; // 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进，这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。 57 } 58 } 59 // 如果 i 小于0 （不在 tab 下标内，按照上面的判断，领取最后一段区间的线程扩容结束） 60 if (i < 0 || i >= n || i + n >= nextn) { 61 int sc; 62 if (finishing) { // 如果完成了扩容 63 nextTable = null;// 删除临时变量 64 table = nextTab;// 更新 table 65 sizeCtl = (n << 1) - (n >>> 1); // 更新阈值（这里相当于将sizeCtl变成了新tab长度的0.75，相当于默认就使用了加载因子0.75，而不是用用户自定义的机加载因子） 66 return;// 结束方法。 67 } 68 // 如果没完成 69 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 尝试将 sc -1. 表示这个线程结束帮助扩容了，将 sc 的低 16 位减一。 70 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。 71 return;// 不相等，说明没结束，当前线程结束方法。 72 finishing = advance = true;// 如果相等，扩容结束了，更新 finising 变量 73 i = n; // 再次循环检查tab是否还有遗漏的位置（当扩容线程数量不够，只能够同时对tab的部分区间进行扩容时，就需要重新检查整张表的遗漏位置） 74 } 75 } 76 else if ((f = tabAt(tab, i)) == null) // 获取老 tab i 下标位置的变量，如果是 null，就使用 fwd 占位。 77 advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位，再次推进一个下标 78 else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。 79 advance = true; // already processed // 说明别的线程已经处理过了，再次推进一个下标 80 else {// 到这里，说明这个位置有实际值了，且不是占位符。对这个节点上锁。为什么上锁，防止 putVal 的时候向链表插入数据 81 synchronized (f) { 82 // 判断 i 下标处的桶节点是否和 f 相同 83 if (tabAt(tab, i) == f) { 84 Node<K,V> ln, hn;// low, height 高位桶，低位桶 85 // 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2 86 if (fh >= 0) { 87 // 对老长度进行与运算（第一个操作数的的第n位于第二个操作数的第n位如果都是1，那么结果的第n也为1，否则为0） 88 // 由于 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1 89 // 如果是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表重新 hash，放到对应的位置上，让新的取于算法能够击中他。 90 int runBit = fh & n; 91 Node<K,V> lastRun = f; // 尾节点，且和头节点的 hash 值取于不相等 92 // 遍历这个桶 93 for (Node<K,V> p = f.next; p != null; p = p.next) { 94 // 取于桶中每个节点的 hash 值 95 int b = p.hash & n; 96 // 如果节点的 hash 值和首节点的 hash 值取于结果不同 97 if (b != runBit) { 98 runBit = b; // 更新 runBit，用于下面判断 lastRun 该赋值给 ln 还是 hn。 99 lastRun = p; // 这个 lastRun 保证后面的节点与自己的取于值相同，避免后面没有必要的循环 100 } 101 } 102 if (runBit == 0) {// 如果最后更新的 runBit 是 0 ，设置低位节点 103 ln = lastRun; 104 hn = null; 105 } 106 else { 107 hn = lastRun; // 如果最后更新的 runBit 是 1， 设置高位节点 108 ln = null; 109 }// 再次循环，生成两个链表，lastRun 作为停止条件，这样就是避免无谓的循环（lastRun 后面都是相同的取于结果） 110 for (Node<K,V> p = f; p != lastRun; p = p.next) { 111 int ph = p.hash; K pk = p.key; V pv = p.val; 112 // 如果与运算结果是 0，那么就还在低位 113 if ((ph & n) == 0) // 如果是0 ，那么创建低位节点 114 ln = new Node<K,V>(ph, pk, pv, ln); 115 else // 1 则创建高位 116 hn = new Node<K,V>(ph, pk, pv, hn); 117 } 118 // 其实这里类似 hashMap 119 // 设置低位链表放在新链表的 i 120 setTabAt(nextTab, i, ln); 121 // 设置高位链表，在原有长度上加 n 122 setTabAt(nextTab, i + n, hn); 123 // 将旧的链表设置成占位符 124 setTabAt(tab, i, fwd); 125 // 继续向后推进 126 advance = true; 127 }// 如果是红黑树 128 else if (f instanceof TreeBin) { 129 TreeBin<K,V> t = (TreeBin<K,V>)f; 130 TreeNode<K,V> lo = null, loTail = null; 131 TreeNode<K,V> hi = null, hiTail = null; 132 int lc = 0, hc = 0; 133 // 遍历 134 for (Node<K,V> e = t.first; e != null; e = e.next) { 135 int h = e.hash; 136 TreeNode<K,V> p = new TreeNode<K,V> 137 (h, e.key, e.val, null, null); 138 // 和链表相同的判断，与运算 == 0 的放在低位 139 if ((h & n) == 0) { 140 if ((p.prev = loTail) == null) 141 lo = p; 142 else 143 loTail.next = p; 144 loTail = p; 145 ++lc; 146 } // 不是 0 的放在高位 147 else { 148 if ((p.prev = hiTail) == null) 149 hi = p; 150 else 151 hiTail.next = p; 152 hiTail = p; 153 ++hc; 154 } 155 } 156 // 如果树的节点数小于等于 6，那么转成链表，反之，创建一个新的树 157 ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : 158 (hc != 0) ? new TreeBin<K,V>(lo) : t; 159 hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : 160 (lc != 0) ? new TreeBin<K,V>(hi) : t; 161 // 低位树 162 setTabAt(nextTab, i, ln); 163 // 高位数 164 setTabAt(nextTab, i + n, hn); 165 // 旧的设置成占位符 166 setTabAt(tab, i, fwd); 167 // 继续向后推进 168 advance = true; 169 } 170 } 171 } 172 } 173 } 174 }
其中值得注意的是，在进行节点迁移的时候，为什么p.hash&n==0就把他放在新tab下标为i的位置（即位置没有改变），而==1就把他放在tab下标为i+n的位置（即老位置+数组增加的长度）？
首先与运算是满足分配律的，也就是假设b=c+d，那么a&b=(a&c)+(a&d)，回到这里，我们把a看成是key.hash，扩容之前数组长度为n，扩容之后为2n，扩容之前一个节点在数组中的位置计算是key.hash&(n-1)，那么扩容之后是key.hash&(2n-1)= (key.hash&n)+[key.hash&(n-1)]，设一个节点在老数组中的位置是i，那么其在新数组的位置是key.hash&n+i。因为n是2的次幂，与上任意一个数，要么是0要么是n，所以如果key.hash&n=0，那么在老数组中位置为i的元素在新数组中的位置也为i，如果key.hash&n不为0，那么老数组中位置为i的元素在新数组中的位置为i+n，也就对应了代码中的结果。
HelperTransfer方法
1 final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { 2 3 Node<K,V>[] nextTab; int sc; 4 5 if (tab != null && (f instanceof ForwardingNode) &&// 当前节点是占位节点 6 (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {// 扩容未结束 7 int rs = resizeStamp(tab.length); 8 while (nextTab == nextTable && table == tab && 9 (sc = sizeCtl) < 0) { 10 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || // 校验sc，如果不等于表示sc异常 11 sc == rs + 1 ||// 表示扩容结束，这是个BUG，应该是sc==rs<<RESIZE_STAMP_SHIFT + 1 12 sc == rs + MAX_RESIZERS // 表示帮助线程数已经达到最大，这是个BUG，应该是sc==rs<<RESIZE_STAMP_SHIFT + MAX_RESIZERS 13 || transferIndex <= 0)// 表示扩容区间已经被领取完毕，不需要新的线程帮助扩容（具体看transfer()方法） 14 break; 15 16 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {// 将sizeCtl+1表示有新的线程帮助扩容 17 // 扩容 18 transfer(tab, nextTab); 19 break; 20 } 21 } 22 return nextTab; 23 } 24 return table; 25 }
以上是关于ConcurrentHashMap源码解析的主要内容，如果未能解决你的问题，请参考以下文章
ConcurrentHashMap源码解析_04 transfer方法源码分析(难点)