ConcurrentHashMap 源码详细分析(JDK1.8)

ConcurrentHashMap 源码详细分析(JDK1.8)

1. 概述

《HashMap 源码详细分析(JDK1.8)》：https://segmentfault.com/a/1190000012926722

Java7 整个 ConcurrentHashMap 是一个 Segment 数组，Segment 通过继承 ReentrantLock 来进行加锁，所以每次需要加锁的操作锁住的是一个 segment，这样只要保证每个 Segment 是线程安全的，也就实现了全局的线程安全。所以很多地方都会将其描述为分段锁。

Java8 对 ConcurrentHashMap 进行了比较大的改动。结构上和 Java8 的 HashMap 基本上一样，不过它要保证线程安全性，所以在源码上确实要复杂一些。

2. 源码分析

ConcurrentHashMap 重点关注它是如何保证线程安全的，和 HashMap 类似的地方就不再赘述。

2.1 构造方法

// 这构造函数里，什么都不干
public ConcurrentHashMap() {
}

// HashMap 初始容量
public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    this.sizeCtl = cap;
}

这个初始化方法有点意思，通过提供初始容量，计算了 sizeCtl，sizeCtl = 【 (1.5 * initialCapacity + 1)，然后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10，那么得到 sizeCtl 为 16，如果 initialCapacity 为 11，得到 sizeCtl 为 32。

在 HashMap 中初始容量直接使用的是 tableSizeFor(initialCapacity)，不知道为什么在 ConcurrentHashMap 改在 1.5 * initialCapacity，至于加 1 估计是考虑 initialCapacity=0 的情况。

sizeCtl 这个属性使用的场景很多，这里为第一个使用场景：ConcurrentHashMap 初始化。sizeCtl=0(也就是无参构造器) 表示使用默认的初始化大小，否则使用自定义的容量。

2.2 putVal 分析

由 put 入手分析 ConcurrentHashMap 中可能出现的线程安全性问题。

2.2.1 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();
    // 1. 计算 hash 值，(h ^ (h >>> 16)) & HASH_BITS
    int hash = spread(key.hashCode());
    int binCount = 0;
    // 2. 通过自旋保证新添加的元素一定会成功添加到 HashMap 中
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        // 2.1 如果数组"空"，进行数组初始化。如何保证数组扩容的线程安全(重点)
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // 2.2 该 hash 对应的槽位(也叫桶bucket) 为空，直接将这个新值放入其中即可
        //     数组 tab 是 volatile 修辞的，并不能说明其元素是 volatile 的。 U.getObjectVolatile
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // CAS 操作将这个新值放入该槽位，如果成功就结束了
            // 如果 CAS 失败，那就是有并发操作，则走第 3 或 4 步
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        // 2.3 只有扩容时 f.hash==MOVED，该线程先帮助扩容才添加值
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        // 2.4 锁住对应的槽位，剩下的操作和 HashMap 就差不多
        else {
            V oldVal = null;
            // 下面的操作都是线程安全的了
            synchronized (f) {
                // f 加锁后对 f 的所有操作都是线程安全的，但 tab 本身并不是线程安全的
                // 也就是说 tab[i] 可能发生变化
                if (tabAt(tab, i) == f) {
                    // 2.4.1 头结点的 hash>=0，说明是链表
                    if (fh >= 0) {
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            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;
                            }
                        }
                    }
                    // 2.4.2 表示红黑树
                    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;
                        }
                    }
                }
            }
            // 链表binCount表示添加元素后的长度，红黑树binCount=2不能进行treeifyBin方法
            if (binCount != 0) {
                // 判断是否要将链表转换为红黑树，临界值和 HashMap 一样，也是 8
                if (binCount >= TREEIFY_THRESHOLD)
                    // 这个方法和 HashMap 中稍微有一点点不同，那就是它不是一定会进行红黑树转换，
                    // 如果当前数组的长度小于 64，那么会选择进行数组扩容，而不是转换为红黑树
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // 3. 元素个数加1，并判断是否扩容，如何保证线程安全
    addCount(1L, binCount);
    return null;
}

put 的主流程看完了，但是在下面几个过程中是如何保证线程安全的：

  • 第一个是数组初始化
  • 第二个是扩容
  • 第三个是帮助数据迁移

2.2.1 数据定位 hash

// 总之就一个目的，让节点在 HashMap 中分布更均匀
static final int spread(int h) {
    return (h ^ (h >>> 16)) & HASH_BITS;
}

h 是对应 key 的 hashcode，计算节点的槽位是 key.hashcode 对数组的长度取余(hash%length)，但如果数组长度为 2^n，则可以直接使用位运算 hash&(length-1)。这个运算实际上只有 hash 的后几位参与运算，为了让 hash 散列的更均匀，也就是 hash 更随机，让 hash 的高 16 和低 16 进行异或运算，这样 hash 的后 16 位就更不容易重复了。注意此时 hashcode 可能为负值，负数在 ConcurrentHashMap 中有特殊的含义，为了保证计算的 hash 一定是正数，可以对于计算得到的 hash 值，强制把符号位去掉，保证结果只在正数区间。 更多参考 hashcode 可能为负值

hash = key.hashCode() & Integer.MAX_VALUE;

2.2.2 数组初始化 initTable

初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        // sizeCtl=-1 表示数组正在初始化
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // sizeCtl=0(也就是无参构造器) 表示使用默认的初始化大小，否则使用自定义的容量
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    // n-n/4，即 0.75*n，和 HashMap 中阈值的计算方式一样，只是这里使用位运算
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

sizeCtl 这个属性之前在 ConcurrentHashMap 初始化化时已经提到过，这里引出了 sizeCtl 另外的使用场景：

  • sizeCtl 第一个使用场景：数组初始化前。sizeCtl=0(也就是无参构造器) 表示使用默认的初始化大小，否则使用自定义的初始化容量。
  • sizeCtl 第二个使用场景：数组初始化中。sizeCtl=-1 表示正在初始化数组 table
  • sizeCtl 第三个使用场景：数组初始化后。sizeCtl>0 表示 table 扩容的阈值

2.2.3 链表转红黑树 treeifyBin

前面我们在 put 源码分析也说过，treeifyBin 不一定就会进行红黑树转换，也可能是仅仅做数组扩容。我们还是进行源码分析吧。

// index是需要链表转红黑树的节点索引
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);
        // 加锁后链表转红黑树，这个和 HashMap 一样
        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));
                }
            }
        }
    }
}

2.3 扩容

2.3.1 扩容 tryPresize

如果说 Java8 ConcurrentHashMap 的源码不简单，那么说的就是扩容操作和迁移操作。数组扩容后真正进行数据迁移的实际上是 transfer 方法，读者应该提前知道这点。这里的扩容也是做翻倍扩容的，扩容后数组容量为原来的 2 倍。

// tryPresize 有两个方法调用 putAll 或 putVal，不管是那个方法 size 都是扩容后的长度
private final void tryPresize(int size) {
    // c：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;
        // 1. 数组没有初始化先初始化，和数组 initTable 类似
        //    putAll 时可能数组还未初始化
        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;
                }
            }
        }
        // 2. 数组容量已经最大或足够了，不需要扩容
        else if (c <= sc || n >= MAXIMUM_CAPACITY)
            break;
        // 3. 真正开始扩容
        else if (tab == table) {
            // 每次扩容时都会生成一个类似时间戳的校验标记，上一次扩容和下一次扩容都不一样
            // 这个值由当前数组的长度决定，格式一定为 0000 0000 100x xxxx
            // 高 16 全部是0，低 16 的第一位是 1，后五位由 n 决定(n转二进制后最高位1之前的0个数，最大32)
            int rs = resizeStamp(n);
            // sc<0 表示其它线程正在扩容，帮助扩容(sc=-1表示数组初始化，已经处理过了)
            // 扩容时 sizeCtl 高 16 位表示扩容戳(校验标记)，低 16 位表示(正在参与扩容的线程数+1)
            if (sc < 0) {
                Node<K,V>[] nt;
                // 3.1 如果 sc 的高 16 位不等于标识符(说明sizeCtl已经改变，扩容已经结束)
                // 3.2 如果 sc == 标识符 + 1 （扩容结束了，不再有线程进行扩容）
                //     第一个线程设置 sc = rs << 16 + 2
                //     第二个线程设置 sc = sc + 1
                //     当一个线程扩容结束 sc = sc -1
                //     最后一个线程扩容结束后 sc == rs + 1(为什么是sc==rs+1，不应该是sc==(rs<<16)+1吗????)
                // 3.3 如果 sc == 标识符 + 65535（辅助扩容线程数已经达到最大，问题同上????）
                // 3.4 如果 nextTable == null（结束扩容了，或者扩容还未开始）
                // 3.5 如果 transferIndex <= 0 (转移状态变化了)
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // 开始扩容(网上说sizeCtl=-1表示数组初始化，因此这里直接+2)
            else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}

首先要注意方法名为 tryPresize，既然方法名中包含了 try 就说明扩容有可能是不成功的。事实上也正是如此，扩容的条件是 sizeCtl>=0，也就是说此时没有线程在进行数组初始化或扩容的操作才会进行扩容。

真正的扩容是由方法 transfer 实现的，这个方法的第二个参数表示扩容后的数组。如果是由当前线程发起的扩容第二个参数为 null，如果其它线程已经在扩容，则当前线程也加入到扩容中去，扩容后的数组已经存在 nextTable。

sizeCtl

前面已经讲解了 sizeCtl 在数组初始化前中后值的变化，这里需要重点关注一下 sizeCtl 在扩容中的使用。sizeCtl 分为两部分，高 16 位表示扩容戳(校验标记)，低 16 位表示正在参与扩容的线程数(线程数+1)。

高 16 位表示扩容戳(校验标记)

static final int resizeStamp(int n) {
    // Integer.numberOfLeadingZeros(n) 表示 n 转二进制后最高位 1 之前的 0 个数，这个数一定小于 32
    // 1 << (RESIZE_STAMP_BITS - 1) 表示 0x7fff，也就是将低 16 位的第一位改成 1
    // 最终结果为 0000 0000 100x xxxx
    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

这个值由当前数组的长度决定，格式一定为 0000 0000 100x xxxx，这个数 rs<<16 后最高位一定是 1，也就是说 sizeCtl 是一个负数。

低 16 位正在扩容的线程数 + 1

初始化时直接 +2，之后每多一个线程参与扩容 +1，这个线程扩容线束则 -1，最终扩容完成则是 1。

2.3.2 数据迁移 helpTransfer

putVal 时发现节点为节点的 f.hash=MOVED，说明有其它线程在对数组进行扩容则会调用 helpTransfer，也就是当前线程先帮助数组扩容后再添加元素。

// 和 tryPresize 差不多，最复杂的部分还是这个if条件判断
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        int rs = resizeStamp(tab.length);
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            // 即然节点已经修改为 ForwardingNode 则说明扩容后的数组已经创建
            // 所以这里的条件判断少了一个 nextTable=null
            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}

2.3.3 数据迁移 transfer

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

第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置，然后从后往前的 stride 个任务属于第一个线程，然后将 transferIndex 指向新的位置，再往前的 stride 个任务属于第二个线程，依此类推。当然，这里说的第二个线程不是真的一定指代了第二个线程，也可以是同一个线程，这个读者应该能理解吧。其实就是将一个大的迁移任务分为了一个个任务包。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 1. stride 可以理解为”步长“，有 n 个位置是需要进行迁移的
    //    将这 n 个任务分为多个任务包，每个任务包有 stride 个任务
    //    stride 在单核下直接等于 n，多核模式下为 (n>>>3)/NCPU，最小值是 16
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range

    // 2. 如果 nextTab 为 null，先进行一次初始化，为什么是线程安全的????
    //    前面我们说了，外围会保证第一个发起迁移的线程调用此方法时，参数 nextTab=null
    //    之后参与迁移的线程调用此方法时，nextTab!=null
    if (nextTab == null) {            // initiating
        try {
            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;

    // 3. ForwardingNode 是占位用的，标记该节点已经处理过了
    //    这个构造方法会生成一个 Node，key、value 和 next 都为 null，关键是 hash 为 MOVED
    //    后面我们会看到，原数组中位置 i 处的节点完成迁移工作后，
    //    就会将位置 i 处设置为这个 ForwardingNode，用来告诉其他线程该位置已经处理过了
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // advance=true 表示一个节点的数据已经处理完了，准备获取下一个节点
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab

    // i 每次任务的上边界，bound 是下边界，注意是从后往前
    // --i < bound 时就领取一次的任务，直到任务处理完毕
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 4. advance 为 true 表示可以进行下一个位置的迁移了
        //    第一次while循环：当前线程领取任务，走第三个else(一直自旋尝试领取任务)
        //          最终的结果为：i 指向了 transferIndex，bound 指向了 transferIndex-stride
        //    之后每处理完一个节点：走第一个if，处理的下一个槽位的节点，直到当前线程领取的任务处理完毕
        //    再次走第三个else，领取步长stride的任务直到transferIndex<=0
        while (advance) {
            int nextIndex, nextBound;
            // 4.1 --i表示处理一下槽位的节点
            if (--i >= bound || finishing)
                advance = false;
            // 4.2 transferIndex每领取一次任务减去一个步长stride
            //     transferIndex初始值为table.length
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            // 4.3 自旋尝试领取步长stride的任务
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                // nextBound 是这次迁移任务的下边界，注意，是从后往前
                bound = nextBound;
                // i 是这次迁移任务的上边界
                i = nextIndex - 1;
                advance = false;
            }
        }
        // 5. 如果 i 小于0 （不在 tab 下标内，按照上面的判断，领取最后一段区间的线程扩容结束）
        //    如果 i >= tab.length(不知道为什么这么判断)
        //    如果 i + tab.length >= nextTable.length（不知道为什么这么判断）
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            // 5.1 完成扩容
            if (finishing) {                    // 完成扩容
                nextTable = null;
                table = nextTab;                // 更新 table
                sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
                return;
            }
            // 5.2 sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2
            //     然后，每有一个线程参与迁移就会将 sizeCtl 加 1
            //     这里使用 CAS 操作对 sizeCtl 进行减 1，代表做完了属于自己的任务
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                // 不相等说明有其它线程在辅助扩容，当前线程直接返回，注意 sc 是-1之前的值
                // 还有其它线程在参与扩容，也就是说扩容还未结束，当前线程直接返回
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;
                // 相等说明没有线程在帮助他们扩容了。也就是说，扩容结束了。
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        // 6. 如果位置 i 处是空的，没有任何节点，那么放入刚刚初始化的 ForwardingNode ”空节点“
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // 7. 该位置处是一个 ForwardingNode，代表该位置已经迁移过了
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        // 8. 数组迁移，原理和 HashMap 一样
        else {
            // 对数组该位置处的结点加锁，开始处理数组该位置处的迁移工作
            synchronized (f) {
                ...
            }
        }
    }
}

上面这部分代码看懂了，下面这部分数据迁移的代码可以先不看，和 HashMap 是一样的。

synchronized (f) {
    if (tabAt(tab, i) == f) {
        Node<K,V> ln, hn;
        // 8.1 hash>=0 表示链表
        if (fh >= 0) {
            int runBit = fh & n;    // n一定是2^n，了就是 1000 0000...
            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) {  // 0则是低位迁移
                ln = lastRun;
                hn = null;
            }
            else {              // 1则是高位迁移
                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);
            }
            setTabAt(nextTab, i, ln);
            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);
                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;
                }
            }
            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;
        }
    }
}

2.4 计数器累加 addCount

只要对 ConcurrentHashMap 中的元素进行增删，元素的个数就会发生变化，这时就需要调用 addCount 方法。在看这个方法之前先看一下 ConcurrentHashMap 是如何记录元素的个数的。

2.4.1 ConcurrentHashMap 数据个数记录庐江

public int size() {
    long n = sumCount();
    return ((n < 0L) ? 0 :
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n);
}

// sumCount 用于统计当前的元素个数
final long sumCount() {
    CounterCell[] as = counterCells; CounterCell a;
    long sum = baseCount;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

可以看到元素的个数分为两部分，一是 baseCount；二是 counterCells 数组中的元素累加。为什么这么复杂？弄个 int 自增，或者多线程环境使用 AtomicInteger 不就可以了？

2.4.2 AtomicLong缺点

我们都知道 AtomicLong 是通过 CAS 自旋的方法去设置 value，直到成功为止。那么当并发数比较多时，就会导致 CAS 的失败机率变高，重试次数更多，越多的线程重试，CAS 失败的机率越高，形成恶性循环，从而降低了效率。

《LongAddr源码解析》：https://www.jianshu.com/p/d9d4be67aa56

怎么保证

原文地址：https://www.cnblogs.com/binarylei/p/10921214.html

时间： 2024-10-11 17:30:39

ConcurrentHashMap 源码详细分析(JDK1.8)的相关文章

HashMap 源码详细解析 (JDK1.8)

概要 HashMap 最早出现在 JDK 1.2 中,底层基于散列算法实现.HashMap 允许 null 键和 null 值,在计算哈键的哈希值时,null 键哈希值为 0.HashMap 并不保证键值对的顺序,这意味着在进行某些操作后,键值对的顺序可能会发生变化.另外,需要注意的是,HashMap 是非线程安全类,在多线程环境下可能会存在问题. HashMap 底层是基于散列算法实现,散列算法分为散列再探测和拉链式.HashMap 则使用了拉链式的散列算法,并在 JDK 1.8 中引入了红黑

OpenStack_Swift源码分析——创建Ring及添加设备源码详细分析

1 创建Ring 代码详细分析 在OpenStack_Swift--Ring组织架构中我们详细分析了Ring的具体工作过程,下面就Ring中增加设备,删除设备,已经重新平衡的实现过程作详细的介绍. 首先看RingBuilder类 def __init__(self, part_power, replicas, min_part_hours): #why 最大 2**32 if part_power > 32: raise ValueError("part_power must be at

Java7/8 中 HashMap 和 ConcurrentHashMap源码对比分析

网上关于 HashMap 和 ConcurrentHashMap 的文章确实不少,不过缺斤少两的文章比较多,所以才想自己也写一篇,把细节说清楚说透,尤其像 Java8 中的 ConcurrentHashMap,大部分文章都说不清楚.终归是希望能降低大家学习的成本,不希望大家到处找各种不是很靠谱的文章,看完一篇又一篇,可是还是模模糊糊. 阅读建议:四节基本上可以进行独立阅读,建议初学者可按照 Java7 HashMap -> Java7 ConcurrentHashMap -> Java8 Ha

DownloadProvider 源码详细分析

DownloadProvider 简介 DownloadProvider 是Android提供的DownloadManager的增强版,亮点是支持断点下载,提供了“开始下载”,“暂停下载”,“重新下载”,“删除下载”接口.源码下载地址 DownloadProvider 详细分析 DownloadProvider开始下载的是由DownloadManager 的 enqueue方法启动的,启动一个新的下载任务的时序图  开始新的下载时候会调用DownloadManager的enqueue方法,然后再

JAVA之ConcurrentHashMap源码深度分析

欢迎转载,请附上出处: http://blog.csdn.net/as02446418/article/details/47004573 我们首先来看一下ConcurrentHashMap类的声明: public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable 其中,这个类继承了java.util.AbstractMa

android_launcher的源码详细分析

转载请注明出处:http://blog.csdn.net/fzh0803/archive/2011/03/26/6279995.aspx 去年做了launcher相关的工作,看了很长时间.很多人都在修改launcher,但还没有详细的文档,把自己积累的东西分享出来,大家一起积累.这份源码是基于2.1的launcher2,以后版本虽有变化,但大概的原理一直还是保留了. 一.主要文件和类 1.Launcher.java:launcher中主要的activity. 2.DragLayer.java:l

RWTHLM 源码详细分析(三)

现在是第三篇,后面的顺序是从输出层,到隐层,然后到输入层的顺序来写,最后在写一下整个框架.这篇介绍输出层的实现,整个程序非常关键的是矩阵相乘的函数,所以在看整个输出层实现之前,非常有必要详细的介绍一下里面反复用到的矩阵相乘函数的各个参数的含义.先看一下FastMatrixMatrixMultiply这个函数,如下: inline void FastMatrixMatrixMultiply( const float alpha, const float a[], const bool transp

Backbone的localStorage.js源码详细分析

Backbone.localStorage.js是将数据存储到游览器客户端本地的(当没有服务器的情况下) 地址:https://github.com/jeromegn/Backbone.localStorage 1     整个函数是一个自执行函数,简化版形式如下 (function(a,fn) { //... })(a,fn); 2     9-12行/**---------此处是判断上下文有没有引入AMD模块(如requirejs)-----------*/ if(typeofdefine=

JDK1.7&amp;1.8源码对比分析【集合】ConcurrentHashMap

前言 在JDK1.7&1.8源码对比分析[集合]HashMap中我们对比分析了JDK1.7和1.8版本的HashMap源码,趁热打铁,这篇文章就来看看JDK1.7和1.8版本的ConcurrentHashMap有哪些区别. 目录 一.对比分析 1. 1.7版本 2. 1.8版本 一.对比分析 1. 1.7版本 先来看看 1.7 的实现,下面是他的结构图: 如上图所示,是由 Segment 数组.HashEntry 组成,和 HashMap 一样,仍然是数组 + 链表. 它的核心成员变量: /**