ConcurrentHashMap的源码分析-transfer

扩容是ConcurrentHashMap的精华之一，扩容操作的核心在于数据的转移，在单线程环境下数据的转移很简单，无非就是把旧数组中的数据迁移到新的数组。但是这在多线程环境下，在扩容的时候其他线程也可能正在添加元素，这时又触发了扩容怎么办？可能大家想到的第一个解决方案是加互斥锁，把转移过程锁住，虽然是可行的解决方案，但是会带来较大的性能开销。因为互斥锁会导致所有访问临界区的线程陷入到阻塞状态，持有锁的线程耗时越长，其他竞争线程就会一直被阻塞，导致吞吐量较低。而且还可能导致死锁。

而ConcurrentHashMap并没有直接加锁，而是采用CAS实现无锁的并发同步策略，最精华的部分是它可以利用多线程来进行协同扩容

简单来说，它把Node数组当作多个线程之间共享的任务队列，然后通过维护一个指针来划分每个线程锁负责的区间，每个线程通过区间逆向遍历来实现扩容，一个已经迁移完的bucket会被替换为一个ForwardingNode节点，标记当前bucket已经被其他线程迁移完了。接下来分析一下它的源码实现

1、fwd:这个类是个标识类，用于指向新表用的，其他线程遇到这个类会主动跳过这个类，因为这个类要么就是扩容迁移正在进行，要么就是已经完成扩容迁移，也就是这个类要保证线程安全，再进行操作。

2、advance:这个变量是用于提示代码是否进行推进处理，也就是当前桶处理完，处理下一个桶的标识

3、finishing:这个变量用于提示扩容是否结束用的

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; //将 (n>>>3相当于 n/8) 然后除以 CPU核心数。如果得到的结果小于 16，那么就使用 16 // 这里的目的是让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象，如果桶较少的话，默认一个 CPU（一个线程）处理 16 个桶，也就是长度为16的时候，扩容的时候只会有一  个线程来扩容 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range //nextTab未初始化，nextTab是用来扩容的node数组 if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked")//新建一个n<<1原始table大小的nextTab,也就是32 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt;//赋值给nextTab } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; //扩容失败，sizeCtl使用int的最大值 return; } nextTable = nextTab; //更新成员变量 transferIndex = n;//更新转移下标，表示转移时的下标 } int nextn = nextTab.length;//新的tab的长度// 创建一个 fwd 节点，表示一个正在被迁移的Node，并且它的hash值为-1(MOVED)，也就是前面我们在讲putval方法的时候，会有一个判断MOVED的逻辑。它的作用是用来占位，表示原数组中位置i处的节点完成迁移以后，就会在i位置设置一个fwd来告诉其他线程这个位置已经处理过了，具体后续还会在讲 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 首次推进为 true，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进 boolean advance = true; //判断是否已经扩容完成，完成就return，退出循环 boolean finishing = false; // to ensure sweep before committing nextTab 通过for自循环处理每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值，并初始化i和bound值，i指当前处理的槽位序号，bound指需要处理的槽位边界，先处理槽位15的节点； for (int i = 0, bound = 0;;) {// 这个循环使用CAS不断尝试为当前线程分配任务 // 直到分配成功或任务队列已经被全部分配完毕 // 如果当前线程已经被分配过bucket区域 // 那么会通过--i指向下一个待处理bucket然后退出该循环 Node<K,V> f; int fh; while (advance) { int nextIndex, nextBound; //--i表示下一个待处理的bucket，如果它>=bound,表示当前线程已经分配过bucket区域 if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) {//表示所有bucket已经被分配完毕 i = -1; advance = false; } //通过cas来修改TRANSFERINDEX,为当前线程分配任务，处理的节点区间为(nextBound,nextIndex)->(0,15) else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound;//0 i = nextIndex - 1;//15 advance = false;} } //i<0说明已经遍历完旧的数组，也就是当前线程已经处理完所有负责的bucket if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) {//如果完成了扩容 nextTable = null;//删除成员变量 table = nextTab;//更新table数组 sizeCtl = (n << 1) - (n >>> 1);//更新阈值(32*0.75=24) return; } // sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2 (详细介绍点击这里) // 然后，每增加一个线程参与迁移就会将 sizeCtl 加 1，// 这里使用 CAS 操作对 sizeCtl 的低16位进行减 1，代表做完了属于自己的任务 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {第一个扩容的线程，执行transfer方法之前，会设置 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 // 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 如果相等，扩容结束了，更新 finising 变量finishing = advance = true; // 再次循环检查一下整张表i = n; // recheck before commit } } // 如果位置 i 处是空的，没有任何节点，那么放入刚刚初始化的 ForwardingNode ”空节点“ else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); //表示该位置已经完成了迁移，也就是如果线程A已经处理过这个节点，那么线程B处理这个节点时，hash值一定为MOVED else if ((fh = f.hash) == MOVED) advance = true; // already processed         }
}

ConcurrentHashMap的源码分析-transfer相关推荐

HashMap与ConcurrentHashMap万字源码分析
HashMap与ConcurrentHashMap`源码解析 JDK版本:1.7 & 1.8 开发中常见的数据结构有三种: 1.数组结构:存储区间连续.内存占用严重.空间复杂度大优点:因 ...
ConcurrentHashMap的源码分析-JDK1.7和Jdk1.8版本的变化
ConcurrentHashMap和HashMap的实现原理是差不多的,但是因为ConcurrentHashMap需要支持并发操作,所以在实现上要比hashmap稍微复杂一些. 在JDK1.7的实现上 ...
[集合]ConcurrentHashMap的源码分析
前言: 强推:一文读懂HashMap 这感觉讲的HashMap很明白. 1. 多线程环境下面,HashMap和Hashtable会怎么样? 1.1 HashMap 因为put会调用: // 新增Ent ...
ConcurrentHashMap的源码分析-高低位原理分析
ConcurrentHashMap在做链表迁移时,会用高低位来实现,这里有两个问题要分析一下 1. 如何实现高低位链表的区分假如我们有这样一个队列第14个槽位插入新节点之后,链表元素个数已经达到了 ...
ConcurrentHashMap的源码分析-扩容过程图解
ConcurrentHashMap支持并发扩容,实现方式是,把Node数组进行拆分,让每个线程处理自己的区域,假设table数组总长度是64,默认情况下,那么每个线程可以分到16个bucket. 然后 ...
ConcurrentHashMap的源码分析-CounterCells初始化图解
初始化长度为2的数组,然后随机得到指定的一个数组下标,将需要新增的值加入到对应下标位置处 transfer扩容阶段判断是否需要扩容,也就是当更新后的键值对总数baseCount >= 阈值si ...
ConcurrentHashMap的源码分析-CounterCells解释
ConcurrentHashMap是采用CounterCell数组来记录元素个数的,像一般的集合记录集合大小,直接定义一个size的成员变量即可,当出现改变的时候只要更新这个变量就行.为什么Concu ...
ConcurrentHashMap的源码分析-put方法第二阶段
在putVal方法执行完成以后,会通过addCount来增加ConcurrentHashMap中的元素个数,并且还会可能触发扩容操作.这里会有两个非常经典的设计 1. 高并发下的扩容 2. 如何保证a ...
ConcurrentHashMap的源码分析-tryPresize
tryPresize里面部分代码和addCount的部分代码类似,看起来会稍微简单一些 private final void tryPresize(int size) {//对size进行修复,主要目 ...

ConcurrentHashMap的源码分析-transfer

ConcurrentHashMap的源码分析-transfer相关推荐

最新文章

热门文章