-
ConcurrentHashMap底层原理（1.7&1.8）
ConcurrentHashMap底层原理（1.7&1.8）
一、ConcurrentHashMap解决线程安全问题
我们知道，在多线程下HashMap会产生线程安全问题。为了保证线程安全，可以使用Hshtable、SynchronizedMap和ConcurrentHashMap，那我们为什么不使用前两个呢？原因如下：
Hashtable和SynchronizedMap不管是 get 还是 put 操作，都是锁住了整个 table，效率低下，因此并不适合高并发场景。
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-djIcc5xo-1660553331975)(E:/Blog/lansg/source/img/image-20220815154228416.png)]](https://1000bd.com/contentImg/2022/08/17/183219634.png)
作为并发情况下多使用的ConcurrentHashMap，我们看看它是如何保证效率的。
二、ConcurrentHashMap（1.7）
想想，既然锁住整张表的话，并发效率低下，那我把整张表分成 N 个部分，并使元素尽量均匀的分布到每个部分中，分别给他们加锁，互相之间并不影响，这种方式岂不是更好 。这就是在 JDK1.7 中 ConcurrentHashMap 采用的方案，被叫做锁分段技术，每个部分就是一个 Segment（段）。
1.底层数据结构（Segment数组+HashEntry数组）
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-okBJznEr-1660553331981)(E:/Blog/lansg/source/img/image-20220406201705727.png)]](https://1000bd.com/contentImg/2022/08/17/183219977.png)
在jdk1.7中，本质上还是采用链表+数组的形式存储键值对的。但是，为了提高并发，把原来的整个 table 划分为 n 个 Segment ，组成一个Segment数组。由上图可知，一个Segment本身可以理解为一个HashMap对象，一个Segment中包含了一个HashEntry数组，也即16个entry元素。
Segment继承了ReentrantLock类，所以每个Segment都拥有自己的一把锁，这样不同的Segment并发操作时就不会互相影响了。如果容量大小是16他的并发度就是16，可以同时允许16个线程操作16个Segment而且还是线程安全的。而同一个Segment中，可以并发读-读，读-写，但是不能写-写，写入操作需要上锁，所以并发写会被阻塞。
2.Segment内部类
static final class Segment<K,V> extends ReentrantLock implements Serializable { private static final long serialVersionUID = 2249069246763182397L; // 和 HashMap 中的 HashEntry 作用一样，真正存放数据的桶 transient volatile HashEntry<K,V>[] table; transient int count; transient int modCount; // 大小 transient int threshold; // 负载因子 final float loadFactor; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
使用volatile去修饰了他的数据Value还有下一个节点next。
2.方法步骤概述
1）Get方法（无需加锁）
1.为输入的Key做Hash运算，得到hash值。
2.通过hash值，定位到对应的Segment对象
3.再次通过hash值，定位到Segment当中数组的具体位置。
（2）Put方法
final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry<K,V>[] tab = table; int index = (tab.length - 1) & hash; HashEntry<K,V> first = entryAt(tab, index); for (HashEntry<K,V> e = first;;) { if (e != null) { K k; // 遍历该 HashEntry，如果不为空则判断传入的 key 和当前遍历的 key 是否相等，相等则覆盖旧的 value。 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 { // 不为空则需要新建一个 HashEntry 并加入到 Segment 中，同时会先判断是否需要扩容。 if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); else setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { //释放锁 unlock(); } return oldValue; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
1.为输入的Key做Hash运算，得到hash值。
2.通过hash值，定位到对应的Segment对象
3.tryLock获取可重入锁，如果获取失败则通过
scanAndLockForPut()自旋获取锁，达到一定重试次数改为阻塞获取。 4.再次通过hash值，定位到Segment当中数组的具体位置。
5.插入或覆盖HashEntry对象。
6.释放锁。
三、ConcurrentHashMap（1.8）
1.底层数据结构（采用Node数组+链表+红黑树）
![[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Yc8jOxik-1660553331982)(E:/Blog/lansg/source/img/image-20220406204430563.png)]](https://1000bd.com/contentImg/2022/08/17/183220151.png)
有点类似HashMap，在jdk1.8中，摒弃了原来的Segment分段锁，采用 CAS + synchronized 来保证并发安全性。Node数组相当于1.7中的HashEntry。
2.成员属性
// 散列表数组最大容量值 private static final int MAXIMUM_CAPACITY = 1 << 30; // 默认容量16 private static final int DEFAULT_CAPACITY = 16; // 最大的数组大小（非2的幂） toArray和相关方法需要(并不是核心属性) static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; // jdk1.7遗留下来的，用来表示并发级别的属性 // jdk1.8只有在初始化的时候用到，不再表示并发级别了~ 1.8以后并发级别由散列表长度决定 private static final int DEFAULT_CONCURRENCY_LEVEL = 16; // 负载因子：在ConcurrentHashMap中，该属性是固定值0.75，不可修改 private static final float LOAD_FACTOR = 0.75f; // 树化阈值：散列表的一个桶中链表长度达到8时候，可能发生链表树化 static final int TREEIFY_THRESHOLD = 8; // 反树化阈值：散列表的一个桶中的红黑树元素个数小于6时候，将红黑树转换回链表结构 static final int UNTREEIFY_THRESHOLD = 6; // 散列表长度达到64，且某个桶位中的链表长度达到8，才会发生树化 static final int MIN_TREEIFY_CAPACITY = 64; // 控制线程迁移数据的最小步长(桶位的跨度~) private static final int MIN_TRANSFER_STRIDE = 16; // 固定值16，与扩容相关，计算扩容时会根据该属性值生成一个扩容标识戳 private static int RESIZE_STAMP_BITS = 16; // (1 << (32 - RESIZE_STAMP_BITS)) - 1 = 65535：1 << 16 -1 // 表示并发扩容最多容纳的线程数 private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; // 也是扩容相关属性 private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; // 当node节点的hash值为-1：表示当前节点是FWD(forwarding)节点(已经被迁移的节点) static final int MOVED = -1; // 当node节点的hash值为-2：表示当前节点已经树化，且当前节点为TreeBin对象，TreeBin对象代理操作红黑树 static final int TREEBIN = -2; // 当node节点的hash值为-3： hash for transient reservations static final int RESERVED = -3; static final int HASH_BITS = 0x7fffffff; // 当前系统的CPU数量 static final int NCPU = Runtime.getRuntime().availableProcessors(); // JDK1.8 序列化为了兼容 JDK1.7的ConcurrentHashMap用到的属性 (非核心属性) private static final ObjectStreamField[] serialPersistentFields = { new ObjectStreamField("segments", Segment[].class), new ObjectStreamField("segmentMask", Integer.TYPE), new ObjectStreamField("segmentShift", Integer.TYPE) }; // 散列表table，采用了volatile去修饰 transient volatile Node<K,V>[] table; // 新表的引用：扩容过程中，会将扩容中的新table赋值给nextTable，（保持引用），扩容结束之后，这里就会被设置为NULL private transient volatile Node<K,V>[] nextTable; // 与LongAdder中的baseCount作用相同: 当未发生线程竞争或当前LongAdder处于加锁状态时，增量会被累加到baseCount private transient volatile long baseCount; // 表示散列表table的状态: // sizeCtl<0时： // 情况一、sizeCtl=-1: 表示当前table正在进行初始化(即，有线程在创建table数组)，当前线程需要自旋等待... // 情况二、表示当前table散列表正在进行扩容，高16位表示扩容的标识戳，低16位表示扩容线程数：(1 + nThread) 即，当前参与并发扩容的线程数量。 // sizeCtl=0时：表示创建table散列表时，使用默认初始容量DEFAULT_CAPACITY=16 // sizeCtl>0时： // 情况一、如果table未初始化，表示初始化大小 // 情况二、如果table已经初始化，表示下次扩容时，触发条件(阈值) private transient volatile int sizeCtl; // 扩容过程中，记录当前进度。所有的线程都需要从transferIndex中分配区间任务，并去执行自己的任务 private transient volatile int transferIndex; // LongAdder中，cellsBusy表示对象的加锁状态： // 0: 表示当前LongAdder对象处于无锁状态 // 1: 表示当前LongAdder对象处于加锁状态 private transient volatile int cellsBusy; // LongAdder中的cells数组，当baseCount发生线程竞争后，会创建cells数组， // 线程会通过计算hash值，去取到自己的cell，将增量累加到指定的cell中 // 总数 = sum(cells) + baseCount private transient volatile CounterCell[] counterCells;- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
4.重要方法
put（添加元素）方法源码
public V put(K key, V value) { return putVal(key, value, false); } final V putVal(K key, V value, boolean onlyIfAbsent) { //可以看到，在并发情况下，key 和 value 都是不支持为空的。 if (key == null || value == null) throw new NullPointerException(); //这里和1.8 HashMap 的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(); //若表已经初始化，则找到当前 key 所在的桶，并且判断是否为空 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //若当前桶为空，则通过 CAS 原子操作，把新节点插入到此位置， //这保证了只有一个线程可以 CAS 成功，其它线程都会失败。 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } //若所在桶不为空，则判断节点的 hash 值是否为 MOVED（值是-1） else if ((fh = f.hash) == MOVED) //若为-1，说明当前数组正在进行扩容，则需要当前线程帮忙迁移数据 tab = helpTransfer(tab, f); else { V oldVal = null; //这里用加同步锁的方式，来保证线程安全，给桶中第一个节点对象加锁 synchronized (f) { //recheck 一下，保证当前桶的第一个节点无变化，后边很多这样类似的操作，不再赘述 if (tabAt(tab, i) == f) { //如果hash值大于等于0，说明是正常的链表结构 if (fh >= 0) { binCount = 1; //从头结点开始遍历，每遍历一次，binCount计数加1 for (Node<K,V> e = f;; ++binCount) { K ek; //如果找到了和当前 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; //若遍历到了尾结点，则把新节点尾插进去 if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } //否则判断是否是树节点。这里提一下，TreeBin只是头结点对TreeNode的再封装 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; } } } } //注意下，这个判断是在同步锁外部，因为 treeifyBin内部也有同步锁，并不影响 if (binCount != 0) { //如果节点个数大于等于 8，则转化为红黑树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); //把旧节点值返回 if (oldVal != null) return oldVal; break; } } } //给元素个数加 1，并有可能会触发扩容 addCount(1L, binCount); return null; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
-
添加元素大致步骤
（1）如果桶数组未初始化，则初始化；
（2）如果待插入的元素所在的桶为空，则尝试把此元素插入到桶的第一个位置（cas写入）；
（3）如果正在扩容，则当前线程一起加入到扩容的过程中；
（4）如果待插入的元素所在的桶不为空且不在迁移元素，则锁住这个桶（synchronize）；
（5）如果当前桶中元素以链表方式存储，则在链表中寻找该元素或者插入元素；
（6）如果当前桶中元素以红黑树方式存储，则在红黑树中寻找该元素或者插入元素；
（7）如果元素存在，则返回旧值；
（8）如果元素不存在，整个Map的元素个数加1，并检查是否需要扩容；
（9）判断当前链表上元素的个数binCount的值是否达到红黑树阈值，给总元素个数+1；
添加元素操作中使用的锁主要有（自旋锁 CAS + synchronized ）。
为什么使用synchronized而不是ReentrantLock？
因为synchronized已经得到了极大地优化，在特定情况下并不比ReentrantLock差，synchronize只锁定当前链表或红黑二叉树的首节点。
get方法
根据目标key所在桶的第一个元素的不同采用不同的方式获取元素，关键点在于
find()方法的重写。public V get(Object key) { // tab 引用map.table // e 当前元素(用于循环遍历) // p 目标节点 // n table数组长度 // eh 当前元素hash // ek 当前元素key Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // 根据key.hashCode()计算hash: 扰动运算后得到得到更散列的hash值 int h = spread(key.hashCode()); // CASE1: // 如果元素所在的桶存在且里面有元素 // 条件一：(tab = table) != null // true -> 表示已经put过数据，并且map内部的table也已经初始化完毕 // false -> 表示创建完map后，并没有put过数据，map内部的table是延迟初始化的，只有第一次写数据时会触发初始化创建table逻辑 // 条件二：(n = tab.length) > 0 如果为 true-> 表示table已经初始化 // 条件三：(e = tabAt(tab, (n - 1) & h)) != null // true -> 当前key寻址的桶位有值 // false -> 当前key寻址的桶位中是null，是null直接返回null if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { // 进入if代码块内部的前置条件：当前桶位有数据 // 如果第一个元素就是要找的元素，则直接返回 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } // CASE2: eh < 0 // 条件成立：即，hash小于0 分2种情况，是树或者正在扩容,需要借助find方法寻找元素，find的寻找方式依据Node的不同子类有不同的实现方式: // 情况一：eh=-1 是fwd结点 -> 说明当前table正在扩容，且当前查询的这个桶位的数据已经被迁移走了，需要借助fwd结点的内部方法find去查询 // 情况二：eh=-2 是TreeBin节点 -> 需要使用TreeBin 提供的find方法查询。 else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; // CASE3: // 当前桶位已经形成链表: 遍历整个链表寻找元素 while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
总结：get方法是不用加锁的，是非阻塞的。Node节点设置了volatile关键字修饰，致使它每次获取的都是最新设置的值。
addCount()方法
若 put 方法元素插入成功之后，则会调用此方法，传入参数为 addCount(1L, binCount)。这个方法的目的很简单，就是把整个 table 的元素个数加 1 。但是，实现比较难。
我们先思考一下，如果让我们自己去实现这样的统计元素个数，怎么实现？
类比 1.8 的 HashMap ，我们可以搞一个 size 变量来存储个数统计。但是，这是在多线程环境下，需要考虑并发的问题。因此，可以把 size 设置为 volatile 的，保证可见性，然后通过 CAS 乐观锁来自增 1。
这样虽然也可以实现。但是，设想一下现在有非常多的线程，都在同一时间操作这个 size 变量，将会造成特别严重的竞争。所以，基于此，这里做了更好的优化。让这些竞争的线程，分散到不同的对象里边，单独操作它自己的数据(计数变量)，用这样的方式尽量降低竞争。到最后需要统计 size 的时候，再把所有对象里边的计数相加就可以了。
initTable（初始化）方法源码
第一次放元素时，初始化桶数组：
/** * table初始化 */ private final Node<K,V>[] initTable() { // tab: 引用map.table // sizeCtl：默认为0，用来控制table的状态、以及初始化和扩容操作: // sizeCtl<0表示table的状态: //（1）=-1，表示有线程正在进行初始化操作。（其他线程就不能再进行初始化，相当于一把锁） //（2）=-(1 + nThreads)，表示有n个线程正在一起扩容。 // sizeCtl>=0表示table的初始化和扩容相关操作： //（3）=0，默认值，后续在真正初始化table的时候使用，设置为默认容量DEFAULT_CAPACITY --> 16。 //（4）>0，将sizeCtl设置为table初始容量或扩容完成后的下一次扩容的门槛。 Node<K,V>[] tab; int sc; // 附加条件的自旋： 条件是map.table尚未初始化 while ((tab = table) == null || tab.length == 0) { // sizeCtl < 0可能是以下2种情况： //（1）-1，表示有其他线程正在进行table初始化操作。 //（2）-(1 + nThreads)，表示有n个线程正在一起扩容。 if ((sc = sizeCtl) < 0) // 这里sizeCtl大概率就是-1，表示其它线程正在进行创建table的过程 //当前线程没有竞争到初始化table的锁，当前线程被迫等待 Thread.yield(); // ----------------------------------------------------------------------------- // sizeCtl) >= 0 且U.compareAndSwapInt(this, SIZECTL, sc, -1)结果为true // U.compareAndSwapInt(this, SIZECTL, sc, -1)：以CAS的方式修改当前线程的sizeCtl为-1， // sizeCtl如果成功被修改为-1，就返回true，否则返回false。 // sizeCtl=-1相当于获取一把锁，表示该线程开始初始化，其他线程不能再进入 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { // 这里为什么又要判断呢？ // 为了防止其它线程已经初始化table完毕了，然后当前线程再次对其初始化，导致丢失数据。 //if语句中为true表示未被初始化 if ((tab = table) == null || tab.length == 0) { // sc>=0的情况如下： //sc=0，默认值，后续在真正初始化table的时候使用 //设置为默认容量DEFAULT_CAPACITY --> 16。 //sc>0，将sizeCtl设置为table初始容量或扩容完成后的下一次扩容的门槛。 // 如果sc大于0,则创建table时使用sc为指定table初始容量大小， // 否则使用16默认值DEFAULT_CAPACITY int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") // 创建新数组nt Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 将新数组nt赋值给table、tab table = tab = nt; // sc设置为下次散列表扩容的门槛：0.75n // n - (n >>> 2) = n - n/4 = 0.75n //value >>> num num 指定要移位值value 移动的位数。 //n无符号右移2位=n/4 sc = n - (n >>> 2); // 可见这里装载因子和扩容门槛都是写死了的 // 这也正是没有threshold和loadFactor属性的原因 } } finally { // 将sc赋值给sizeCtl，分为一下2种情况： // 1、if ((tab = table) == null || tab.length == 0)判断成功 // 则当前线程是第一次创建map.table的线程，sc就表示下一次扩容的阈值。 // 2、if判断失败 // 则当前线程并不是第一次创建map.table的线程， //这是表示将sizeCtl恢复原值（前面通过加锁将其设为-1了） sizeCtl = sc; } break; } } return tab; }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
总结：通过CAS锁保证只有一个线程在某一时间进行初始化
transfer（迁移元素，扩容）方法
扩容时容量变为两倍，并把部分元素迁移到其它桶中。
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range if (nextTab == null) { // initiating // 如果nextTab为空，说明还没开始迁移 // 就新建一个新桶数组 try { // 新桶数组是原桶的两倍 @SuppressWarnings("unchecked") 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 = nextTab; transferIndex = n; } //新桶数组大小 int nextn = nextTab.length; ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 新建一个ForwardingNode类型的节点，并把新桶数组存储在里面 boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; // 整个while循环就是在算i的值，过程太复杂，不用太关心 // i的值会从n-1依次递减，感兴趣的可以打下断点就知道了 // 其中n是旧桶数组的大小，也就是说i从15开始一直减到1这样去迁移元素 while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound; i = nextIndex - 1; advance = false; } } if (i < 0 || i >= n || i + n >= nextn) { // 如果一次遍历完成了 // 也就是整个map所有桶中的元素都迁移完成了 int sc; if (finishing) { // 如果全部迁移完成了，则替换旧桶数组 // 并设置下一次扩容门槛为新桶数组容量的0.75倍 nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1); return; } if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 当前线程扩容完成，把扩容线程数-1 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) // 扩容完成两边肯定相等 return; // 把finishing设置为true // finishing为true才会走到上面的if条件 finishing = advance = true; // i重新赋值为n // 这样会再重新遍历一次桶数组，看看是不是都迁移完成了 // 也就是第二次遍历都会走到下面的(fh = f.hash) == MOVED这个条件 i = n; // recheck before commit } } else if ((f = tabAt(tab, i)) == null) // 如果桶中无数据，直接放入ForwardingNode标记该桶已迁移 advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) // 如果桶中第一个元素的hash值为MOVED // 说明它是ForwardingNode节点 // 也就是该桶已迁移 advance = true; // already processed else { // 锁定该桶并迁移元素 synchronized (f) { // 再次判断当前桶第一个元素是否有修改 // 也就是可能其它线程先一步迁移了元素 if (tabAt(tab, i) == f) { // 把一个链表分化成两个链表 // 规则是桶中各元素的hash与桶大小n进行与操作 // 等于0的放到低位链表(low)中，不等于0的放到高位链表(high)中 // 其中低位链表迁移到新桶中的位置相对旧桶不变 // 高位链表迁移到新桶中位置正好是其在旧桶的位置加n // 这也正是为什么扩容时容量在变成两倍的原因 Node<K,V> ln, hn; if (fh >= 0) { // 第一个元素的hash值大于等于0 // 说明该桶中元素是以链表形式存储的 // 这里与HashMap迁移算法基本类似 // 唯一不同的是多了一步寻找lastRun // 这里的lastRun是提取出链表后面不用处理再特殊处理的子链表 // 比如所有元素的hash值与桶大小n与操作后的值分别为 0 0 4 4 0 0 0 // 则最后后面三个0对应的元素肯定还是在同一个桶中 // 这时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; } // 遍历链表，把hash&n为0的放在低位链表中 // 不为0的放在高位链表中 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); } // 低位链表的位置不变 setTabAt(nextTab, i, ln); // 高位链表的位置是原位置加n setTabAt(nextTab, i + n, hn); // 标记当前桶已迁移 setTabAt(tab, i, fwd); // advance为true，返回上面进行--i操作 advance = true; } else if (f instanceof TreeBin) { // 如果第一个元素是树节点 // 也是一样，分化成两颗树 // 也是根据hash&n为0放在低位树中 // 不为0放在高位树中 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; // 遍历整颗树，根据hash&n是否为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; } } // 如果分化的树中元素个数小于等于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); // 高位树的位置是原位置加n setTabAt(nextTab, i + n, hn); // 标记该桶已迁移 setTabAt(tab, i, fwd); // advance为true，返回上面进行--i操作 advance = true; } } } } } }- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
- 177
- 178
- 179
- 180
- 181
- 182
- 183
- 184
- 185
- 186
- 187
- 188
- 189
步骤总结：
（1）新桶数组大小是旧桶数组的两倍；
（2）迁移元素先从靠后的桶开始；
（3）迁移完成的桶在里面放置一ForwardingNode类型的元素，标记该桶迁移完成；
（4）迁移时根据hash&n是否等于0把桶中元素分化成两个链表或树；
（5）低位链表（树）存储在原来的位置；
（6）高们链表（树）存储在原来的位置加n的位置；
（7）迁移元素时会锁住当前桶，也是分段锁的思想；
协助扩容helpTransfer方法
线程添加元素时发现正在扩容且当前元素所在的桶元素已经迁移完成了，则协助迁移其它桶的元素。
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { // nextTab 引用的是 fwd.nextTable == map.nextTable 理论上是这样。 // sc 保存map.sizeCtl Node<K,V>[] nextTab; int sc; // CASE0: 如果桶数组不为空，并且当前桶第一个元素为ForwardingNode类型，并且nextTab不为空 // 说明当前桶已经迁移完毕了，才去帮忙迁移其它桶的元素 // 扩容时会把旧桶的第一个元素置为ForwardingNode，并让其nextTab指向新桶数组 // 条件一：tab != null 恒成立 true // 条件二：(f instanceof ForwardingNode) 恒成立 true // 条件三：((ForwardingNode- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
-
相关阅读:
聊聊“JVM 调优&JVM 性能优化”是怎么个事？
ATF启动（六）：bl32（OP-TEE）--＞bl33 ATF ending
2022/11/27[指针] 指针与函数基础
C#学习记录——网络编程基础
Mybatis核心源码赏析（三）
【附源码】计算机毕业设计JAVA重工教师职称管理系统
IMX6Q的SD卡启动使用教程（1）：uboot与kernel编译移植
字节码打桩插入代码块|IOC框架之对象注入~研究
关于sqli-labs-master靶场的SQL注入问题。
读书笔记：Effective C++ 3.0版2005年Scott Meyers ： 55条建议(47-55)
- 原文地址：https://blog.csdn.net/lans_g/article/details/126350049
