多线程高并发编程(10) -- ConcurrentHashMap源码分析
一.背景
前文讲了HashMap的源码分析，从中能够看到下面的问题：html
- HashMap的put/remove方法不是线程安全的，若是在多线程并发环境下，使用synchronized进行加锁，会致使效率低下；
- 在遍历迭代获取时进行修改(put/remove)操做，会致使发生并发修改异常(ConcurrentModificationException)；
- 在JDK1.7以前，对HashMap进行put添加操做，会致使链表反转，形成链表回路，从而发生get死循环，（固然这个问题在JDK1.8被改进了按照原链表顺序进行重排移动）；
- 若是多个线程同时检测到元素个数超过 数组大小 * loadFactor，这样就会发生多个线程同时对数组进行扩容，都在从新计算元素位置以及复制数据，可是最终只有一个线程扩容后的数组会赋给 table，也就是说其余线程的都会丢失，而且各自线程 put 的数据也丢失；
基于上述问题，均可以使用ConcurrentHashMap进行解决，ConcurrentHashMap使用分段锁技术解决了并发访问效率，在遍历迭代获取时进行修改操做也不会发生并发修改异常等等问题。node
二.源码解析
-
构造方法：
//最大容量大小 private static final int MAXIMUM_CAPACITY = 1 << 30; //默认容量大小 private static final int DEFAULT_CAPACITY = 16; /** *控制标识符，用来控制table的初始化和扩容的操做，不一样的值有不一样的含义 * 多线程之间，以volatile方式读取sizeCtl属性，来判断ConcurrentHashMap当前所处的状态。 * 经过cas设置sizeCtl属性，告知其余线程ConcurrentHashMap的状态变动 *未初始化： * sizeCtl=0：表示没有指定初始容量。 * sizeCtl>0：表示初始容量。 *初始化中： * sizeCtl=-1,标记做用，告知其余线程，正在初始化 *正常状态： * sizeCtl=0.75n ,扩容阈值 *扩容中: * sizeCtl < 0 : 表示有其余线程正在执行扩容 * sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2 :表示此时只有一个线程在执行扩容 */ private transient volatile int sizeCtl; //并发级别 private static final int DEFAULT_CONCURRENCY_LEVEL = 16; //建立一个新的空map，默认大小是16 public ConcurrentHashMap() { } public ConcurrentHashMap(int initialCapacity) { if (initialCapacity < 0) throw new IllegalArgumentException(); //调整table的大小，tableSizeFor的实现查看前文HashMap源码分析的构造方法模块 int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); this.sizeCtl = cap; } public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this.sizeCtl = DEFAULT_CAPACITY; putAll(m); } public ConcurrentHashMap(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, 1); } /** * concurrencyLevel：并发度,预估同时操做数据的线程数量 * 表示可以同时更新ConccurentHashMap且不产生锁竞争的最大线程数。 * 默认值为16，(即容许16个线程并发可能不会产生竞争)。 */ public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); //至少使用一样多的桶容纳一样多的更新线程来操做元素 if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; }
-
put：
public V put(K key, V value) { return putVal(key, value, false); } static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash普通节点哈希的可用位 //把位数控制在int最大整数以内，h ^ (h >>> 16)的含义查看前文的put源码解析 static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS; } final V putVal(K key, V value, boolean onlyIfAbsent) { //key和value为空抛出异常 if (key == null || value == null) throw new NullPointerException(); //获得hash值 int hash = spread(key.hashCode()); int binCount = 0; //自旋对table进行遍历 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //初始化table if (tab == null || (n = tab.length) == 0) tab = initTable(); //若是hash计算出的槽位元素为null，CAS将元素填充进当前槽位并结束遍历 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } //hash为-1，说明正在扩容，那么就帮助其扩容。以加快速度 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; synchronized (f) {// 同步 f 节点，防止增长链表的时候致使链表成环 if (tabAt(tab, i) == f) {// 若是对应的下标位置的节点没有改变 if (fh >= 0) {//f节点的hash值大于0 binCount = 1;//链表初始长度 // 死循环，直到将值添加到链表尾部，并计算链表的长度 for (Node<K,V> e = f;; ++binCount) { K ek; //hash和key相同，值进行覆盖 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; //hash和key不一样，添加到链表后面 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) { //链表长度大于等于8时，将链表转换成红黑树树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // 判断是否须要扩容 addCount(1L, binCount); return null; }
-
initTable：初始化
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { //若是一个线程发现sizeCtl<0，意味着另外的线程执行CAS操做成功，当前线程只须要让出cpu时间片，即保证只有一个线程初始化 //因为sizeCtl是volatile的，保证了顺序性和可见性 if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//cas操做判断并置为-1 try { if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//若没有参数则默认容量为16 @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//建立数组 table = tab = nt;//数组赋值给当前ConcurrentHashMap //计算下一次元素到达扩容的阀值，若是n为16的话，那么这里 sc = 12，其实就是 0.75 * n sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; }
-
tabAt：寻找指定数组在内存中i位置的数据
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { /**getObjectVolatile：获取obj对象中offset偏移地址对应的object型field的值,支持volatile load语义。 * 数组的寻址计算方式:a[i]_address = base_address + i * data_type_size * base_address:起始地址;i:索引;data_type_size:数据类型长度大小 */ return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); }
-
helpTransfer：帮助扩容
private static int RESIZE_STAMP_BITS = 16; /** * numberOfLeadingZeros()的具体算法逻辑请参考：https://www.jianshu.com/p/2c1be41f6e59 * numberOfLeadingZeros（n）返回的是n的二进制标识的从高位开始到第一个非0的数字的之间0的个数，好比numberOfLeadingZeros（8）返回的就是28 ，由于0000 0000 0000 0000 0000 0000 0000 1000在1前面有28个0 * RESIZE_STAMP_BITS 的值是16，1 << (RESIZE_STAMP_BITS - 1)就是将1左移位15位，0000 0000 0000 0000 1000 0000 0000 0000 * 而后将两个数字再按位或，将至关于 将移位后的 两个数相加。 * 好比： * 8的二进制表示是： 0000 0000 0000 0000 0000 0000 0000 1000 = 8 * 7的二进制表示是： 0000 0000 0000 0000 0000 0000 0000 0111 = 7 * 按位或的结果是： 0000 0000 0000 0000 0000 0000 0000 1111 = 15 * 至关于 8 + 7 =15 * 为何会出现这种效果呢？由于8是2的整数次幂，也就是说8的二进制表示只会在某个高位上是1，其他地位都是0，因此在按位或的时候，低位表示的全是7的位值，因此出现了这种效果。 */ static final int resizeStamp(int n) { return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1)); } final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { Node<K,V>[] nextTab; int sc; //若是table不是空，且node节点是转移类型，且node节点的nextTable（新 table）不是空，尝试帮助扩容 if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { //根据length获得一个标识符号 int rs = resizeStamp(tab.length); //若是nextTab没有被并发修改，且tab也没有被并发修改，且sizeCtl<0（说明还在扩容） while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) { /** * 若是 sizeCtl 无符号右移16不等于rs（ sc前16位若是不等于标识符，则标识符变化了） * 或者 sizeCtl == rs + 1（扩容结束了，再也不有线程进行扩容）（默认第一个线程设置 sc ==rs 左移 16 位 + 2，当第一个线程结束扩容了，就会将 sc 减1。这个时候，sc 就等于 rs + 1） * 或者 sizeCtl == rs + 65535 （若是达到最大帮助线程的数量，即 65535） * 或者转移下标正在调整 （扩容结束） * 结束循环，返回 table * 【即若是还在扩容，判断标识符是否变化，判断扩容是否结束，判断是否达到最大线程数，判断扩容转移下标是否在调整（扩容结束），若是知足任意条件，结束循环。】 */ if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break; // 若是以上都不是, 将 sizeCtl + 1, （表示增长了一个线程帮助其扩容） if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { transfer(tab, nextTab);//进行扩容和数据迁移 break; } } return nextTab;//返回扩容后的数组 } return table;//没有扩容，返回原数组 }
-
transfer：扩容和数据迁移，采用多线程扩容，整个扩容过程，经过cas设置sizeCtl、transferIndex等变量协调多个线程进行并发扩容；
-
transferIndex属性：
//扩容索引，表示已经分配给扩容线程的table数组索引位置。主要用来协调多个线程，并发安全地获取迁移任务（hash桶）。 private transient volatile int transferIndex;
-
在扩容以前，transferIndex 在数组的最右边 。此时有一个线程发现已经到达扩容阈值，准备开始扩容。
算法 -
扩容线程，在迁移数据以前，首先要将transferIndex左移（以cas的方式修改 transferIndex=transferIndex-stride(要迁移hash桶的个数)），获取迁移任务。每一个扩容线程都会经过for循环+CAS的方式设置transferIndex，所以能够确保多线程扩容的并发安全。()换个角度，咱们能够将待迁移的table数组，当作一个任务队列，transferIndex当作任务队列的头指针。而扩容线程，就是这个队列的消费者。扩容线程经过CAS设置transferIndex索引的过程，就是消费者从任务队列中获取任务的过程。
-
-
扩容过程：
-
容量已经达到扩容阈值，须要进行扩容操做，此时transferindex=tab.length=32
-
扩容线程A 以cas的方式修改transferindex=31-16=16 ,而后按照降序迁移table[31]--table[16]这个区间的hash桶
-
迁移hash桶时，会将桶内的链表或者红黑树，按照必定算法，拆分红2份，将其插入nextTable[i]和nextTable[i+n]（n是table数组的长度）。 迁移完毕的hash桶,会被设置成ForwardingNode节点，以此告知访问此桶的其余线程，此节点已经迁移完毕
-
此时线程2访问到了ForwardingNode节点，若是线程2执行的put或remove等写操做，那么就会先帮其扩容。若是线程2执行的是get等读方法，则会调用ForwardingNode的find方法，去nextTable里面查找相关元素
-
线程2加入扩容操做
-
若是准备加入扩容的线程，发现如下状况，放弃扩容，直接返回。
-
发现transferIndex=0,即全部node均已分配数组
-
发现扩容线程已经达到最大扩容线程数缓存
-
-
-
-
-
-
-
源码解析
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; //先判断CPU核数，若是是多核，将数组长度/8，再/核数，获得stride，不然stride=数组长度，若是stride<16,则stride=16 //这里的目的是让每一个CPU处理的桶同样多，避免出现转移任务不均匀的现象，若是桶较少的话，默认一个CPU（一个线程）处理16个桶，即确保每次至少获取16个桶(迁移任务) 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<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//扩容2倍 nextTab = nt;//更新 } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE;//扩容失败，sizeCtl使用int最大值。 return; } nextTable = nextTab;//更新成员变量 //transferIndex默认=table.length transferIndex = n; } int nextn = nextTab.length;//新tab的长度 //建立一个fwd节点，用于占位。当别的线程发现这个槽位中是fwd类型的节点，表示其余线程正在扩容，而且此节点已经扩容完毕，跳过这个节点。关联了nextTab，能够经过ForwardingNode.find()访问已经迁移到nextTab的数据。 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); //首次推动为 true，若是等于true，说明须要再次推动一个下标（i--），反之，若是是false，那么就不能推动下标，须要将当前的下标处理完毕才能继续推动 boolean advance = true; //完成状态，若是是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：若是当前线程能够向后推动；这个循环就是控制i递减。同时，每一个线程都会进入这里取得本身须要转移的桶的区间 //分析场景：table.length=32，此时执行到这个地方nextTab.length=64 A,B线程同时进行扩容。 //A，B线程同时执行到while循环中cas这段代码 //A线程获第一时间抢到资源，设置bound=nextBound=16，i = nextIndex - 1=31 A线程搬运table[31]~table[16]中间16个元素 //B线程再次回到while起点，而后在次获取到 bound = nextBound-0，i=nextIndex - 1=15，B线程搬运table[15]~table[0]中间16个元素 //当transferIndex=0的时候，说明table里面全部搬运任务都已经完成，没法在分配任务。 while (advance) { int nextIndex, nextBound; // 对i减1，判断是否大于等于bound（正常状况下，若是大于bound不成立，说明该线程上次领取的任务已经完成了。那么，须要在下面继续领取任务） // 若是对i减1大于等于 bound，或者完成了，修改推动状态为 false，不能推动了。任务成功后修改推动状态为 true。 // 一般，第一次进入循环，i-- 这个判断会没法经过，从而走下面的nextIndex = transferIndex（获取最新的转移下标）。其他状况都是：若是能够推动，将i减1，而后修改为不可推动。若是i对应的桶处理成功了，改为能够推动。 if (--i >= bound || finishing) advance = false;//这里设置false，是为了防止在没有成功处理一个桶的状况下却进行了推动 // 这里的目的是：1. 当一个线程进入时，会选取最新的转移下标。 // 2. 当一个线程处理完本身的区间时，若是还有剩余区间的没有别的线程处理，再次CAS获取区间。 else if ((nextIndex = transferIndex) <= 0) { // 若是小于等于0，说明没有区间能够获取了，i改为-1，推动状态变成false，再也不推动 // 这个-1会在下面的if块里判断，从而进入完成状态判断 i = -1; advance = false;//这里设置false，是为了防止在没有成功处理一个桶的状况下却进行了推动 } // CAS修改transferIndex，即 length - 区间值，留下剩余的区间值供后面的线程使用 else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound;//这个值就是当前线程能够处理的最小当前区间最小下标 i = nextIndex - 1;//初次对i赋值，这个就是当前线程能够处理的当前区间的最大下标 advance = false;// 这里设置false，是为了防止在没有成功处理一个桶的状况下却进行了推动，这样致使漏掉某个桶。下面的 if(tabAt(tab, i) == f) 判断会出现这样的状况。 } } //i<0（不在 tab 下标内，按照上面的判断，领取最后一段区间的线程结束） if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) {// 若是完成了扩容和数据迁移 nextTable = null;//删除成员遍历 table = nextTab;//更新table sizeCtl = (n << 1) - (n >>> 1);//更新阀值 return;//结束transfer } //若是没完成，尝试将sc -1. 表示这个线程结束帮助扩容了，将 sc 的低 16 位减一。 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { //若是 sc - 2 不等于标识符左移 16 位。若是他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。 /** *第一个扩容的线程，执行transfer方法以前(helpTransfer方法中)，会设置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2) *后续帮其扩容的线程，执行transfer方法以前，会设置 sizeCtl = sizeCtl+1 *每个退出transfer的方法的线程，退出以前，会设置 sizeCtl = sizeCtl-1 *那么最后一个线程退出时： *必然有sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT */ if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return;// 不相等，说明不到最后一个线程，直接退出transfer方法 finishing = advance = true;// 若是相等，扩容结束了，更新 finising 变量 i = n; // recheck before commit，最后退出的线程要从新check下是否所有迁移完毕 } } else if ((f = tabAt(tab, i)) == null) // 获取老tab的i下标位置的变量，若是是 null，就使用 fwd 占位。 advance = casTabAt(tab, i, null, fwd);// 若是成功写入 fwd 占位，再次推动一个下标 else if ((fh = f.hash) == MOVED)// 若是不是 null 且 hash 值是 MOVED。 advance = true; // already processed，说明别的线程已经处理过了，再次推动一个下标 else {// 到这里，说明这个位置有实际值了，且不是占位符。对这个节点上锁。为何上锁，防止 putVal 的时候向链表插入数据 synchronized (f) { // 判断 i 下标处的桶节点是否和 f 相同 if (tabAt(tab, i) == f) { Node<K,V> ln, hn;// low, height 高位桶，低位桶 // 若是 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2 if (fh >= 0) { // 对老长度进行与运算（第一个操做数的的第n位于第二个操做数的第n位若是都是1，那么结果的第n为也为1，不然为0） // 因为 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1 // 若是是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表从新 hash，放到对应的位置上，让新的取于算法可以击中他。 int runBit = fh & n; Node<K,V> lastRun = f; // 尾节点，且和头节点的 hash 值取于不相等 // 遍历这个桶 for (Node<K,V> p = f.next; p != null; p = p.next) { // 取于桶中每一个节点的 hash 值 int b = p.hash & n; // 若是节点的 hash 值和首节点的 hash 值取于结果不一样 if (b != runBit) { runBit = b; // 更新 runBit，用于下面判断 lastRun 该赋值给 ln 仍是 hn。 lastRun = p; // 这个 lastRun 保证后面的节点与本身的取于值相同，避免后面没有必要的循环 } } if (runBit == 0) {// 若是最后更新的 runBit 是 0 ，设置低位节点 ln = lastRun; hn = null; } else { hn = lastRun; // 若是最后更新的 runBit 是 1， 设置高位节点 ln = null; }// 再次循环，生成两个链表，lastRun 做为中止条件，这样就是避免无谓的循环（lastRun 后面都是相同的取于结果） for (Node<K,V> p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; // 若是与运算结果是 0，那么就还在低位 if ((ph & n) == 0) // 若是是0 ，那么建立低位节点 ln = new Node<K,V>(ph, pk, pv, ln); else // 1 则建立高位 hn = new Node<K,V>(ph, pk, pv, hn); } // 其实这里相似 hashMap // 设置低位链表放在新数组的 i setTabAt(nextTab, i, ln); // 设置高位链表，在原有长度上加 n setTabAt(nextTab, i + n, hn); // 将旧的链表设置成占位符，表示处理过了 setTabAt(tab, i, fwd); // 继续向后推动 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); // 和链表相同的判断，与运算 == 0 的放在低位 if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } // 不是 0 的放在高位 else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } // 若是树的节点数小于等于 6，那么转成链表，反之，建立一个新的树 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; } } } } } }
-
-
addCount：计数
// 从 putVal 传入的参数是x=1，check=binCount默认是0，只有hash冲突了才会大于1,且他的大小是链表的长度（若是不是红黑树结构的话,红黑树=2）。 private final void addCount(long x, int check) { CounterCell[] as; long b, s; //若是计数盒子不是空或者修改 baseCount 失败 if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true; // 若是计数盒子是空（还没有出现并发） // 若是随机取余一个数组位置为空 或者 // 修改这个槽位的变量失败（出现并发了） // 执行 fullAddCount 方法，在fullAddCount自旋直到CAS操做成功才结束退出 if (as == null || (m = as.length - 1) < 0 || (a = as[ThreadLocalRandom.getProbe() & m]) == null || !(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { fullAddCount(x, uncontended); return; } if (check <= 1) return; s = sumCount(); } // 检查是否须要扩容，在 putVal 方法调用时，默认就是要检查的(check默认是0，链表是链表长度，红黑树是2)，若是是值覆盖了，就忽略 if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; // 若是map.size() 大于 sizeCtl（达到扩容阈值须要扩容） 且 // table 不是空；且 table 的长度小于 1 << 30。（能够扩容） while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { // 根据 length 获得一个标识 int rs = resizeStamp(n); if (sc < 0) {//代表此时有别的线程正在进行扩容 // 若是 sc 的低 16 位不等于 标识符（校验异常 sizeCtl 变化了） // 若是 sc == 标识符 + 1 （扩容结束了，再也不有线程进行扩容）（默认第一个线程设置 sc ==rs 左移 16 位 + 2，当第一个线程结束扩容了，就会将 sc 减一。这个时候，sc 就等于 rs + 1） // 若是 sc == 标识符 + 65535（帮助线程数已经达到最大） // 若是 nextTable == null（结束扩容了） // 若是 transferIndex <= 0 (转移状态变化了) // 结束循环 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; // 不知足前面5个条件时，尝试参与这次扩容，把正在执行transfer任务的线程数加1，+2表明有1个，+1表明有0个，表示多了一个线程在帮助扩容，执行transfer if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } //若是不在扩容，将 sc 更新：标识符左移 16 位 而后 + 2. 也就是变成一个负数。高 16 位是标识符，低 16 位初始是 2. //试着让本身成为第一个执行transfer任务的线程 else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount();// 从新计数，判断是否须要开启下一轮扩容 } } }
-
-
get：
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()); //table有值，且查找到的槽位有值(tabAt方法经过valatile读) if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { //hash、key、value都相同返回当前查找到节点的值 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } //遍历特殊节点:红黑树、已经迁移的节点（ForwardingNode)等 else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; //遍历node链表（e.next也是valitle变量） while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; } Node<K,V> find(int h, Object k) { Node<K,V> e = this; if (k != null) { do { K ek; if (e.hash == h && ((ek = e.key) == k || (ek != null && k.equals(ek)))) return e; } while ((e = e.next) != null); } return null; }
-
remove：
public V remove(Object key) { return replaceNode(key, null, null); } //经过volatile设置第i个节点的值 static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); } final V replaceNode(Object key, V value, Object cv) { int hash = spread(key.hashCode()); //自旋 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //数组或查找的槽位为空，结束自旋返回null if (tab == null || (n = tab.length) == 0 || (f = tabAt(tab, i = (n - 1) & hash)) == null) break; //正在扩容，帮助扩容 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null;//返回的旧值 boolean validated = false;//是否进行删除链表或红黑树节点 synchronized (f) {//槽位加锁 //getObjectVolatile获取tab[i],若是此时tab[i]!=f,说明其余线程修改了tab[i]。回到for循环开始处，从新执行 if (tabAt(tab, i) == f) {//槽位节点没有变化 if (fh >= 0) {//槽位节点是链表 validated = true; //遍历链表 for (Node<K,V> e = f, pred = null;;) { K ek; //hash、key、value相同 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { V ev = e.val;//临时节点缓存当前节点值 //值相同 if (cv == null || cv == ev || (ev != null && cv.equals(ev))) { oldVal = ev;//给旧值赋值 if (value != null)//值覆盖，replace()调用 e.val = value; else if (pred != null)//有前节点，表示当前节点不是头节点 pred.next = e.next;//删除当前节点 else setTabAt(tab, i, e.next);//删除头节点，即更新当前槽位(数组槽位)节点为头节点的下一节点 } break; } //当前节点不是目标节点，继续遍历下一个节点 pred = e; //到达链表尾部，依旧没有找到，跳出循环 if ((e = e.next) == null) break; } } else if (f instanceof TreeBin) {//红黑树 validated = true; TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> r, p; //树有节点且查找的节点不为null if ((r = t.root) != null && (p = r.findTreeNode(hash, key, null)) != null) { V pv = p.val; //值相同 if (cv == null || cv == pv || (pv != null && cv.equals(pv))) { oldVal = pv;//给旧值赋值 if (value != null)//值覆盖，replace()调用 p.val = value; else if (t.removeTreeNode(p))//删除节点成功 setTabAt(tab, i, untreeify(t.first));//更新当前槽位(数组槽位)节点为树的第一个节点 } } } } } if (validated) { //若是删除了节点，更新size if (oldVal != null) { if (value == null) addCount(-1L, -1);//数量-1 return oldVal; } break; } } } return null; }
三.总结
-
put：使用cas插入，若是是链表或树节点才会加锁同步操做，提升了性能
- 不容许有key或value为null，不然抛出异常；
- 在第一次put时初始化table（initTable()），初始化有并发控制，经过sizeCtl变量判断，sizeCtl<0表示已经有线程在初始化，当前线程就不在进行，不然sizeCtl置为-1(CAS)并建立数组；
- 当hash计算出的槽位节点为null时，使用CAS插入元素；
- 当hash为MOVED（-1）时，帮助扩容，但可能帮助不了，由于每一个线程默认16个桶，若是只有16个桶，第二个线程没法帮助扩容；
- 若是hash冲突了，同步槽位节点，若是槽位是链表结构，进行链表操做，覆盖旧值或插入到链表尾部；若是是树结构，添加到树中；
- 元素添加到链表或树中，若是链表长度大于8，将链表转换为红黑树；
- 调用addCount()，对size+1，并判断是否须要扩容addCount()，若是是值覆盖操做就不须要调用该方法；
-
initTable：初始化
- 数组table为null才进行初始化，不然直接返回旧数组；
- 若是当前sizeCtl小于0，表示有线程正在初始化，则当前线程礼让CPU，保证只有一个线程正在初始化数组；
- 若是没有线程在初始化，则当前线程CAS将sizeCtl置为-1并建立数组，而后从新计算阀值；
-
helpTransfer：帮助扩容
- 当尝试插入操做时，发现节点是forward类型，则会帮助扩容；
- 每次加入一个线程都会将sizeCtl的低16位+1，同时校验高16位的标识符；
- 扩容最大的帮助线程是65535，这是低16位的最大值限制；
- 每一个线程默认分配16个桶，若是桶的数量是16，那么第二个线程没法帮助扩容，即桶被分配完其余线程没法进场扩容；
-
transfer：扩容和数据迁移
- 根据CPU核数平均分配给每一个CPU相同数量的桶，若是不够16个，默认就是16个；
- 按照2倍容量进行扩容；
- 每一个线程在处理完本身领取的区间后，还能够继续领取，若是还有的话，经过transferIndex变量递减16实现桶数量控制；
- 每次处理空桶的时候，会把当前桶标识为forward节点，告诉put的其余线程说“我正在扩容，快来帮忙”，但若是只有16个桶，只能有一个线程进行扩容；
- 若是有了占位符MOVED，表示已经被处理过，跳过这个桶，继续推动处理其余桶；
- 若是有真正的实际值，那么就同步加锁头节点，防止putVal的并发；
- 同步块里将链表拆分红两份，根据 hash & length 获得是不是0，若是是0，放在新数组低位，反之放在length+i的高位。这是防止下次取值hash找不到正确的位置；
- 若是该桶类型是红黑树，也会拆分红2个，而后判断拆分过的桶的大小是否小于等于6，若是是转换成链表；
- 线程处理完若是没有可选区间，且任务没有完成，则会将整个表检查一遍，防止遗漏；
-
addCount：扩容判断
- 当插入结束时，会对size+1，并判断是否须要扩容的判断；
- 优先使用计数盒子（若是不是空，说明并发了），若是计数盒子是空，使用baseCount变量+1；
- 若是修改baseCount失败，使用计数盒子，若是仍是修改失败，在fullAddCount()中自旋直到CAS操做成功；
- 检查是否须要扩容；
- 若是size大于等于sizeCtl且长度小于1<<30，能够扩容；
- 若是已经在扩容，帮助其扩容；
- 若是没有在扩容，自行开启扩容，更新sizeCtl变量为负数，赋值为标识符高16位+2；
-
remove：删除元素
- 自旋遍历数量，若是数组或根据hash计算的槽位节点值为null，直接结束自旋返回null；
- 若是槽位节点正在扩容，帮助扩容；
- 若是槽位节点有值，同步加锁；
- 若是该槽位节点仍是没有任何变化，判断是链表结构类型节点仍是树结构类型节点，经过遍历查找元素，找到删除该节点或从新设置头节点；
- 若是删除了节点，更新size-1，若是有旧值则返回旧值，不然返回null；
四.参考
- https://www.jianshu.com/p/2829fe36a8dd
- https://www.jianshu.com/p/487d00afe6ca
- https://juejin.im/post/5b001639f265da0b8f62d0f8#comment
相关文章
- 1. 多线程高并发编程(10) -- ConcurrentHashMap源码分析
- 2. 多线程高并发编程(8) -- Fork/Join源码分析
- 3. 多线程高并发编程(6) -- Semaphere、Exchanger源码分析
- 4. 多线程高并发编程(7) -- Future源码分析
- 5. 多线程高并发编程(5) -- CountDownLatch、CyclicBarrier源码分析
- 6. 并发编程---ConcurrentHashMap源码解析
- 7. 多线程-ConcurrentHashMap源码分析
- 8. 并发编程专题九-并发容器ConcurrentHashMap源码分析
- 9. 【多线程并发编程】十二 CopyOnWriteArrayList源码分析
- 10. 多线程高并发编程(4) -- ReentrantReadWriteLock读写锁源码分析
- 更多相关文章...
- • Rust 并发编程 - RUST 教程
- • C# 多线程 - C#教程
- • Git五分钟教程
- • Java 8 Stream 教程
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 多线程高并发编程(10) -- ConcurrentHashMap源码分析
- 2. 多线程高并发编程(8) -- Fork/Join源码分析
- 3. 多线程高并发编程(6) -- Semaphere、Exchanger源码分析
- 4. 多线程高并发编程(7) -- Future源码分析
- 5. 多线程高并发编程(5) -- CountDownLatch、CyclicBarrier源码分析
- 6. 并发编程---ConcurrentHashMap源码解析
- 7. 多线程-ConcurrentHashMap源码分析
- 8. 并发编程专题九-并发容器ConcurrentHashMap源码分析
- 9. 【多线程并发编程】十二 CopyOnWriteArrayList源码分析
- 10. 多线程高并发编程(4) -- ReentrantReadWriteLock读写锁源码分析