Java 并发编程（十一）： Java 8 基于CAS 的ConcurrentHashMap 的实现原理

旧城等待， 2022-03-27 08:42 179阅读 0赞

## 1、ConcurrentHashMap 简介 ##

ConcurrentHashMap 是一个支持完全并发检索和更新的哈希表。该类遵循了与 java.util.Hashtable 相同的功能规范，与 Hashtable 的每个方法都有相对应的版本。它的所有操作都是线程安全的，检索操作也不需要锁定，但是不支持锁定整个表。在依赖于线程安全但不依赖于其同步细节的程序中，此类可与Hashtable完全互操作。检索操作（包括get）通常不会阻塞，因此可能与更新操作（包括put和remove）重叠。

## 2、ConcurrentHashMap 出现的原因 ##

在并发编程中使用HashMap可能导致程序死循环。而使用线程安全的HashTable效率又非常低下，基于以上两个原因，ConcurrentHashMap 也就诞生了。

（1）非线程安全的HashMap在多线程环境下，使用HashMap进行put操作会引起死循环，导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。例如，以下代码就会引起死循环。

public static void main(String[] args) throws InterruptedException {
            final HashMap<String, String> map = new HashMap<String, String>(2);
            Thread t = new Thread(() -> {
                for (int i = 0; i < 10000; i++) {
                    new Thread(() -> map.put(UUID.randomUUID().toString(), ""), "child_thread_" + i).start();
                }
            }, "parent_thread");
            t.start();
            t.join();
        }

HashMap在并发执行put操作时会引起死循环，是因为多线程会导致HashMap的Entry链表形成环形数据结构，一旦形成环形数据结构，Entry的next节点永远不为空，就会产生死循环获取Entry。

（2）效率低下的HashTable

HashTable 容器使用synchronized来保证线程安全，在线程竞争激烈的情况下 HashTable 的效率非常低下。因为当一个线程访问 HashTable 的同步方法，其他线程也访问HashTable的同步方法时，会进入阻塞或轮询状态。如线程1使用put进行元素添加，线程2不但不能使用put方法添加元素，也不能使用get方法来获取元素，所以竞争越激烈效率越低。

（3）ConcurrentHashMap 用分离锁实现多个线程间的并发写操作

在 ConcurrentHashMap 中，线程对映射表做读操作时，一般情况下不需要加锁就可以完成，对容器做结构性修改的操作才需要加锁。

（4）用 Volatile 变量协调读写线程间的内存可见性

由于内存可见性问题，未正确同步的情况下，写线程写入的值可能并不为后续的读线程可见。

根据 Java 内存模型，对 同一个 volatile 变量的写 / 读操作可以确保：写线程写入的值，能够被之后未加锁的读线程“看到”。

## 3、ConcurrentHashMap 数据结构 ##

Java 8为进一步提高并发性，摒弃了分段锁（Segment）的方案，而是启用了一种全新的方式实现,利用CAS算法。它沿用了 HashMap 的思想，底层由 数组+链表+红黑树 的方式实现，但是为了做到并发，又增加了很多辅助的类。

同时为了提高哈希碰撞下的寻址性能，Java 8在链表长度超过一定阈值（8）时将链表（寻址时间复杂度为O(N)）转换为红黑树（寻址时间复杂度为O(log(N))）

将原先table数组＋单向链表的数据结构，变更为table数组＋单向链表＋红黑树的结构。对于hash表来说，最核心的能力在于将key hash之后能均匀的分布在数组中。如果hash之后散列的很均匀，那么table数组中的每个队列长度主要为0或者1。但实际情况并非总是如此理想，虽然ConcurrentHashMap类默认的加载因子为0.75，但是在数据量过大或者运气不佳的情况下，还是会存在一些队列长度过长的情况，如果还是采用单向列表方式，那么查询某个节点的时间复杂度为O(n)；因此，对于个数超过8(默认值)的列表，jdk1.8中采用了红黑树的结构，那么查询的时间复杂度可以降低到O(logN)，大大提高了检索效率。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2NDcwMzU_size_16_color_FFFFFF_t_70]

## 4、ConcurrentHashMap 元素定位方式 ##

Java 8 的 ConcurrentHashMap 同样是通过Key的哈希值与数组长度取模确定该Key在数组中的索引。为了避免不太好的Key的hashCode设计，它通过如下方法计算得到Key的最终哈希值。不同的是，Java 8的ConcurrentHashMap作者认为引入红黑树后，即使哈希冲突比较严重，寻址效率也足够高，所以作者并未在哈希值的计算上做过多设计，只是将Key的hashCode值与其高16位作异或并保证最高位为0。

static final int spread(int h) {
            return (h ^ (h >>> 16)) & HASH_BITS;
        }

## 5、ConcurrentHashMap 同步方式 ##

对于put操作，如果Key对应的数组元素为null，则通过CAS操作将其设置为当前值。如果Key对应的数组元素（链表表头或树的根元素）不为null，则对该元素使用synchronized关键字申请锁，然后进行操作。如果该put操作使得当前链表长度超过一定阈值，则将该链表转换为树，从而提高寻址效率。

对于读操作，由于数组被volatile关键字修饰，因此不用担心数组的可见性问题。同时每个元素是一个Node实例（Java 7中每个元素是一个HashEntry），它的Key值和hash值都由final修饰，不可修改，无需关心它们被修改后的可见性问题。而其Value及对下一个元素的引用由volatile修饰，可见性也有保障。

static class Node<K,V> implements Map.Entry<K,V> {
            final int hash;
            final K key;
            volatile V val;
            volatile Node<K,V> next;
    
            Node(int hash, K key, V val, Node<K,V> next) {
                this.hash = hash;
                this.key = key;
                this.val = val;
                this.next = next;
            }
         --------------------省略---------------------
     }

## 6、ConcurrentHashMap 操作方法 ##

### 6.1、Get 操作 ###

总的来说 get 操作是最简单的，除非正好遇到了扩容 ForwardingNode.find(…)

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());
            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)
                    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;
        }

1、根据 key 计算 hash 值

2、根据 hash 值找到数组对应位置: (n – 1) & h

3、判断该位置是链表还是红黑树

3.1、如果该位置为 null，那么直接返回 null
    3.2、如果该位置的节点刚好就是我们需要的，返回该节点的值
    3.3、如果该位置节点的 hash 值小于 0，说明正在扩容，或者是红黑树
    3.4、如果以上都不满足，那就是链表，进行遍历查找

### 6.2、PUT 操作 ###

public V put(K key, V value) {
            return putVal(key, value, false);
        }
    
        /** Implementation for put and putIfAbsent */
        final V putVal(K key, V value, boolean onlyIfAbsent) {
            if (key == null || value == null) throw new NullPointerException();
            //计算key的hash值
            int hash = spread(key.hashCode());
            //记录链表长度
            int binCount = 0;
            for (Node<K,V>[] tab = table;;) {
                Node<K,V> f; int n, i, fh;
                //数据 tab 为空，初始化数组
                if (tab == null || (n = tab.length) == 0)
                    tab = initTable();
                //根据 hash 值，找到对应数组下标，并获取对应位置 的第一个节点 f
                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
                }
                //f 的 hash 值等于 MOVED，说明在扩容
                else if ((fh = f.hash) == MOVED)
                    //进行数据迁移
                    tab = helpTransfer(tab, f);
                else {
                    //f 是该位置的首节点，而且不为空
                    V oldVal = null;
                    // 获取数组该位置的首节点的监视器锁
                    synchronized (f) {
                        if (tabAt(tab, i) == f) {
                            // 首节点 hash 值大于 0，说明是链表
                            if (fh >= 0) {
                                //记录链表长度
                                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;
                                    }
                                }
                            }
                            //如果当前位置是红黑树
                            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;
                                }
                            }
                        }
                    }
                    //如果链表长度不为0
                    if (binCount != 0) {
                        //如果链表长度大于等于临界值【8】，将链表转换为红黑树
                        if (binCount >= TREEIFY_THRESHOLD)
                            //将链表转换成红黑树前判断，如果当前数组的长度小于 64，那么会选择进行数组扩容，而不是转换为红黑树
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            addCount(1L, binCount);
            return null;
        }

put 操作流程梳理：

1、根据 key 计算 hash 值
    
    2、遍历数组 tab 
    2.1、如果数组为空，进行数组初始化 initTable()
    2.2、如果数组不为空，根据 hash 值找到在数组 tab 中的位置，并获取该位置的第一个节点元素对象
        2.2.1、如果数组该位置为空，使用 CAS 操作将值放入该位置
        2.2.2、如果数组该位置的首个元素的hash 值等于 MOVED ，说明在扩容，帮助数据迁移 
        helpTransfer
        
    2.3、如果不满足以上（2.1、2.2）条件，执行以下逻辑：
        2.3.1、获取数组该位置的首节点的监视器锁
        2.3.2、首节点 hash 值大于 0，说明是链表，链表遍历插入值
        2.3.3、否则为红黑树，使用红黑树的 putTreeVal 方法，插入新值
        2.3.4、如果以上执行的是链表操作，判断链表长度是否达到限定值，是否需要转换为红黑树

#### 6.2.1、initTable 初始化数组 ####

主要就是初始化一个合适大小的数组，然后设置 sizeCtl，初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。

/**
         * Initializes table, using the size recorded in sizeCtl.
         */
        private final Node<K,V>[] initTable() {
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) {
                //sizeCtl 的值小于 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 或指定的长度
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            //将数组赋值给 table，它是 volatile 的
                            table = tab = nt;
                            // 如果 n 为 16 的话，那么这里 sc = 12，其实就是 0.75 * n
                            // 根据负载因子计算负载值（0.75*n）,n = 16，这里 sc = 12
                            sc = n - (n >>> 2);
                        }
                    } finally {
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

#### 6.2.2、treeifyBin 链表转红黑树 ####

treeifyBin 不一定就会进行红黑树转换，也可能是仅仅做数组扩容，转换之前会首先判断链表的长度，如果链表长度小于限定值 64 则进行数组扩容，否则进行链表转换红黑树操作

/**
         * Replaces all linked nodes in bin at given index unless table is
         * too small, in which case resizes instead.
         */
        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，进行数组扩容
                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));
                        }
                    }
                }
            }
        }

#### 6.2.3、tryPresize 数组扩容 ####

数组扩容后是原始大小的两倍

/**
         * Tries to presize table to accommodate the given number of elements.
         *
         * @param size number of elements (doesn't need to be perfectly accurate)
         */
        private final void tryPresize(int size) {
            // 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 (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);
                            }
                        } finally {
                            sizeCtl = sc;
                        }
                    }
                }
                else if (c <= sc || n >= MAXIMUM_CAPACITY)
                    break;
                else if (tab == table) {
                    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;
                        //使用 CAS 将 sizeCtl 加 1，然后执行 transfer 方法
                        if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                            transfer(tab, nt);
                    }
                    else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                                 (rs << RESIZE_STAMP_SHIFT) + 2))
                        transfer(tab, null);
                }
            }
        }

#### 6.2.4、transfer 数据迁移 ####

这个方法很点长，它是将原来的 tab 数组的元素迁移到新的 nextTab 数组中。

此方法支持多线程执行，调用此方法的时候，会保证第一个发起数据迁移的线程，nextTab 参数为 null，之后再调用此方法的时候，nextTab 不会为 null。

原数组长度为 n，所以我们有 n 个迁移任务，让每个线程每次负责一个小任务是最简单的，每做完一个任务再检测是否有其他没做完的任务，帮助迁移就可以了，而 Doug Lea 使用了一个 stride，简单理解就是步长，每个线程每次负责迁移其中的一部分，比如每次迁移 16 个小任务。所以，我们就需要一个全局的调度者来安排哪个线程执行哪几个任务，这个就是属性 transferIndex 的作用。

第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置，然后从后往前的 stride 个任务属于第一个线程，然后将 transferIndex 指向新的位置，再往前的 stride 个任务属于第二个线程，依此类推。

/**
         * Moves and/or copies the nodes in each bin to new table. See
         * above for explanation.
         */
        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，先进行初始化
            if (nextTab == null) {            // initiating
                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 是 ConcurrentHashMap 中的属性
                nextTable = nextTab;
                
                // transferIndex 是 ConcurrentHashMap 的属性，用于控制迁移的位置
                transferIndex = n;
            }
            int nextn = nextTab.length;
            
            
            // 这个构造方法会生成一个Node，key、value 和 next 都为 null，关键是 hash 为 MOVED
            // 后面我们会看到，原数组中位置 i 处的节点完成迁移工作后，就会将位置 i 处设置为这个 ForwardingNode，用来告诉其他线程该位置已经处理过了
            // 它其实相当于是一个标识。
            ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
            boolean advance = true;
            boolean finishing = false; // to ensure sweep before committing nextTab
            
            // i 是位置索引，bound 是边界
            for (int i = 0, bound = 0;;) {
                Node<K,V> f; int fh;
                
                
                // advance 为 true 表示可以进行下一个位置的迁移了
                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;
                        finishing = advance = true;
                        i = n; // recheck before commit
                    }
                }
                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) {
                                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;
                            }
                        }
                    }
                }
            }
        }

### 6.3、Size 方法 ###

由源码可知该方法的最大返回值为 Integer.MAX\_VALUE，但是这个 Map 的长度是有可能超过 int 最大值的，所以 JDK 8 增了 mappingCount 方法。

mappingCount 方法的返回值是 long 类型。所以不必限制最大值必须是 Integer.MAX\_VALUE

public int size() {
            long n = sumCount();
            return ((n < 0L) ? 0 :
                    (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                    (int)n);
        }
        
    /**
         * Returns the number of mappings. This method should be used
         * instead of {@link #size} because a ConcurrentHashMap may
         * contain more mappings than can be represented as an int. The
         * value returned is an estimate; the actual count may differ if
         * there are concurrent insertions or removals.
         *
         * @return the number of mappings
         * @since 1.8
         */
        public long mappingCount() {
            long n = sumCount();
            return (n < 0L) ? 0L : n; // ignore transient negative values
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2NDcwMzU_size_16_color_FFFFFF_t_70]: /images/20220327/2baf80a635ed4a27b64c7511d25ad422.png