JUC之ConcurrentHashMap

约定不等于承诺〃 2024-02-19 21:07 21阅读 0赞

基于jdk1.8  
ConcurrentHashMap是一个优秀的数据结构。尽可能的规避锁使用，尽可能使用无锁并发，使用算法与红黑树等优秀的数据结构优化并发，提升其并发的同时保证线程安全。首先我们先总结一下1.8对其的一些优化点。

1.  取消Segment段抽象，1.8之前使用分段加锁提升并发，1.8之后将锁的粒度更加细化，在必须需要锁的时候仅对一个节点加锁，粒度更细。之所以能够如此大胆的使用锁细化主要归功于其算法与结构设计的优化。其hash算法的优化以及扩容的优化
2.  hash冲突在到达阈值8并且表长度大于64时优化为红黑树结构，复杂度由原来的链表O(n)优化为O(logN)
3.  扩容优化，扩容允许并发扩容。最大并发线程数为65535（2的16次方）

### 常用参数解释 ###

#### table ####

实际map存储数据的Node数组

#### nextTable ####

扩容时用的新表，旧表数据搬运完成后会将新表cas至table

#### sizeCtl ####

在创建map时可以指定容积，但是容积会被map结构优化，优化算法为：指定容积+指定容积/2+1之后向上取最接近的2的次方数，例如：16，优化后为32

/**
     * map初始化与resize扩容控制使用的属性.  当是负数时，代表正在初始化或者resize扩容: 
     * -1 代表初始化,其他代表 -(1 + 正在扩容的线程数).
     */
    private transient volatile int sizeCtl;

#### transferIndex ####

当并发resize扩容时，表被切分为若干分段，记录下一次的分段开始index下标值。后面会再描述该值计算方式与使用方式

#### counterCells ####

并发冲突时，每个cpu会对应一个单元格计算，线程封闭来保证并发，最终计算count需要将单元格累加，初始化创建两个实例，每次冲突会按照2的倍数扩增实例，不会超过cpu数量。  
并发期间map的当前size等于baseCount+各个单元格的值

#### LOAD\_FACTOR ####

负载因子=0.75，为什么选择0.75？如果太小则空间浪费大，如果太大则查询效率低，空间与时间的一个权衡值，不容置疑的是大佬们根据实战数据测算得来的-\_-!!!

### resize(rehash)扩容 ###

#### 判断是否需要resize扩容 ####

1.  扩容方法改用transfer方法，put元素时，如果出现hash冲突，则在addCount时进行一次check检查是否需要进行扩容，如果两个count单元格counterCells都已经不为空并且自增成功。则检查当前元素值是否超过阈值（sizeCtl），sizeCtl正常情况下等于map的容积Capacity - Capacity/2/2，例如：容积=32，则sizeCtl=32-32/2/2=24。超过则进行resize

// sizeCtl计算方式,n=容积：sizeCtl = n - (n >>> 2);
    /**
     * 增加count值, 如果表很小并且还没有扩容，开始转移数据进行扩容。
     * 如果正在扩容，帮助执行转移数据如果当前工作是有效的。
     * 转移元素后再次检查占用情况，看看是否需要再继续进行扩容，因为扩容已经滞后与增加count值
     * @param x the count to add
     * @param check 如果小于0, 不检查是否需要resize, 如果小于等于1则仅检查是否无竞争
     */
    private final void addCount(long x, int check) {
        
        CounterCell[] as; long b, s;
        if ((as = counterCells) != null ||
            // cas操作失败说明存在竞争，则继续
            // 也就是下面的如果check大于等于0则进行检查是否需要resize
            // 如果存在竞争cas会操作失败
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                // 获取当前线程对应的counter单元格，如果为空则直接进行fullAddCount操作后返回
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                // 如果操作增加失败则进行fullAddCount操作后返回，
                // 操作失败代表多个线程在竞争操作同一个counter单元格
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
        
                // 存在竞争操作失败后尝试扩容单元格，不能超过cpu数量且只能为2的倍数
                fullAddCount(x, uncontended);
                return;
            }
            // 如果小于等于1则仅检查是否无竞争，小于等于1代表hash没有冲突，或者冲突很少没有到达阈值8。
            // 不再继续检查是否需要扩容
            if (check <= 1)
                return;
            // 如果大于1则需要进一步检查是否需要扩容，因为冲突已经超出阈值8开始了链表转红黑树结构的优化
            s = sumCount();
        }
        if (check >= 0) {
        
            Node<K,V>[] tab, nt; int n, sc;
            // 如果超出了阈值，并且table不为空，并且table的长度小于最大值。则进行resize扩容
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
        
                // 获取1左移15位+当前table长度的2进制表示中最高的"1"位前面
                // 零的数量（例如：000101，"1"位前面零的数量为3）
                int rs = resizeStamp(n);
                //在没有并发扩容的情况下，sc一定是大于等于零的，sizeCtl的算法不再多说
                if (sc < 0) {
        
                    // 1. 第一步一般为相等状态。不相等时说明低16位已经被填满，也就是达到了最大值MAX_RESIZERS限制，继续往下看可以发现实际的线程并发数是要小于MAX_RESIZERS的（大约=MAX_RESIZERS-长度二进制高位零的数量-1），高16位用来在扩容时判断是否可以终止transfer搬运数据
                    // 2. 第二步判断如果当前sc等于rs+1则终止搬运数据，sc是-21亿左右的值，rs是正数，没有理解这一步的含义，理论上永远不会满足该条件就已经完成了扩容
                    // 3. 第三步判断如果超出rs + MAX_RESIZERS阈值则终止。没有理解这一步的含义，理论上永远不会满足该条件就已经完成了扩容
                    // 4. 第四步说明当前并发线程搬运数据已经完成，不需要再继续处理
                    // 5. 第五步说明当前最后一个线程正在搬运最后剩余的数据，线程按照stride步进已经走到了最后的index为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))
                        // nextTable是扩容后的新表
                        transfer(tab, nt);
                }
                // 没有并发扩容的情况下，进行扩容，并将sizeCtl置为负数
                // RESIZE_STAMP_SHIFT=32-RESIZE_STAMP_BITS=16
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             // 低16位的最右侧两位10代表完成扩容，与transfer中的完成判断对应
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    // 开始转移原table表中的数据准备扩容
                    transfer(tab, null);
                s = sumCount();
            }
        }
    }

#### 搬运数据至新表transfer ####

迁移或者复制table表中的数据至新表nextTable

/**
     * 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;
        // 步进垮位：当前表长度/2/2/2/cpu数量，最小值不能低于16
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        // cas操作是原子的，不会出现并发初始化问题，所以该内部代码是线程安全的。
        // 因为nextTab要么为空要么不为空
        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 = nextTab;
            transferIndex = n;
        }
        int nextn = nextTab.length;
        // 创建指向新表的node节点，该节点hash值固定位MOVED，表示已迁移至新表
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        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 (advance) {
        
                int nextIndex, nextBound;
                // 搬运数据的指针没有走到当前bound边界继续搬运不需要向前跳跃
                // 或者已经完成所有搬运不需要再向前跳跃
                if (--i >= bound || finishing)
                    advance = false;
                // transferIndex已经搬运到小于等于0的下标处，即已经完成了搬运不再需要向前跳跃
                else if ((nextIndex = transferIndex) <= 0) {
        
                    i = -1;
                    advance = false;
                }
                // 将transferIndex向前跳跃一个步进，迁移分区是按照cpu数量的1/6进行切分
                // 即每个cpu最多迁移6个分区
                else if (U.compareAndSwapInt
                         (this, TRANSFERINDEX, nextIndex,
                          nextBound = (nextIndex > stride ?
                                       nextIndex - stride : 0))) {
        
                    // 更新当前边界值为新的下标
                    bound = nextBound;
                    // 从右向左搬运数据直到bound边界处再进行下一次跳跃
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            // 如果i<0或者i大于老表长度或者i+老表长度大于等于新表长度。
            // 则说明完成了搬运数据，判断是否需要结束，还是继续扩容
            if (i < 0 || i >= n || i + n >= nextn) {
        
                int sc;
                // 搬运是否已经完成，是则将老表指针指向新表
                // 并且按照算法：老表的长度*2-老表长度/2 计算的值更新为新sizeCtl阈值
                // 即新容积-老容积/2
                if (finishing) {
        
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                // 当前线程工作已完成，退出前递减sizeCtl值，负数时，低16位代表当前帮助扩容的线程数量
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
        
                    // 如果sizeCtl前16位已经被占满，则直接退出。
                    // 低16位的最右侧两位10代表完成扩容,与上一步addCount中的开始先+2对应
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    // 标识已经完成搬运，advance置为true，i置为n，代表再次重n-1位置，
                    // 进行一次全表检查是否完成了所有搬运
                    finishing = advance = true;
                    // 提交变更前再次检查是否已经完成搬运所有数据
                    i = n; // recheck before commit
                }
            }
            // 如果搬运的节点为空，即还没有写入数据直接迁移即可
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            // 与其他线程搬移的边界碰撞，当前线程向前按照步进跳跃至下一个分区继续搬运（如果还有没有处理的分区存在，否则递减sizeCtl退出）
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
        
                // 当前节点存在数据，按照节点进行加锁后搬运数据。搬运过程不会
                synchronized (f) {
        
                    ...
                }
            }
        }
    }

### put ###

1.  按照扰动函数\_spread对\_key的hashcode计算index索引下标
2.  如果table中对应下标为空则直接放入即可
3.  如果不为空，hash值为MOVED(-1)，如果是说明正在扩容，尝试加入扩容大军，如果扩容的线程已经到达最大值则返回的是nextTable也就是新表直接将数据放入新表中
4.  如果不为空，hash值不是MOVED，则对该节点进行加synchronized锁处理
5.  前几个步骤binCount都是默认值0
6.  如果节点的hash值大于等于0，首先binCount值改为1，并且遍历链表添加至链表最后一个节点，并且将链表的长度记到binCount中。
7.  如果节点hash值小于0，首先binCount值改为2，判断是否为红黑树结构TreeBin类型，是则添加至红黑树
8.  写红黑树会将root节点锁住lockRoot（必须当前状态为0，无写入无读取状态时才会成功获取），将共享状态\_LOCKSTATE使用CAS算法置为WRITER，如果\_lockRoot\_成功则继续写入数据，如果设置失败说明存在竞争尝试加竞争锁\_contendedLock
9.  非写入WRITER、读取状态（**即处于无锁或者等待的状态**），再次尝试获取WRITER写锁，如果waiting为true，说明当前节点自旋期间尝试获取写锁成功，不需要进入等待状态，将等待线程恢复为null，当前节点进入写状态返回。如果尝试获取WRITER写锁失败即判断期间状态发生了变更则继续判断  
    1.如果当前状态不是等待（**即处于写入或读取状态，如果是无锁状态上一个判断分支就可以成功获取到锁资源**）即当前没有等待线程，则当前节点加入等待并将等待位置1（准备进入等待状态即将阻塞），进入下一轮自旋回到上一步继续判断
10. 如果当前状态不是WRITER，并且已经准备好进入WAITER状态，已经处于了等待状态即waiting为True，使用LockSupport.*park阻塞等待唤醒。***\_假如此时再有新线程进入，并且也获取写锁失败尝试竞争锁，会在\_contendedLock代码块中进入自旋**
11. 存在阻塞线程情况是：存在WRITER写线程，并且当前是第一个进入等待的线程。后面的线程自旋互相竞争WRITER或者WAITER资源

private final void contendedLock() {
        
        boolean waiting = false;
        for (int s;;) {
        
            // 非写入WRITER、读取状态（即处于无锁或者等待的状态），尝试再次获取WRITER锁资源，失败则继续，成功则返回
            if (((s = lockState) & ~WAITER) == 0) {
        
                // 则尝试获取WRITER锁资源，进入写状态
                if (U.compareAndSwapInt(this, LOCKSTATE, s, WRITER)) {
        
                    // 获取写锁成功，如果是进入了等待状态，则将等待的线程置空即可，不需要后面的唤醒，唤醒需要判断等待线程不为空
                    if (waiting)
                        waiter = null;
                    return;
                }
            }
            // 如果当前状态不是等待（即处于写入或读取状态，如果是无锁状态上一个判断分支就可以成功获取到锁资源），则状态更新为WAITER，进入等待状态，绑定等待线程准备进入阻塞（可能不会进入阻塞，中途可能再次获取锁成功）
            else if ((s & WAITER) == 0) {
        
                if (U.compareAndSwapInt(this, LOCKSTATE, s, s | WAITER)) {
        
                    waiting = true;
                    // 记录当前等待中的线程
                    waiter = Thread.currentThread();
                }
            }
            // 成功获取等待锁资源，阻塞等待被唤醒
            else if (waiting)
                LockSupport.park(this);
        }
    }

1.  添加步骤已完成，进行后续处理
2.  判断binCount是否大于阈值8，如果是说明是插入列表的行为，并且超出了阈值，则优化为红黑树TreeBin，索引对应节点为是TreeBin类型其hash值为固定的\_TREEBIN（-2），TreeBin节点下面对应红黑树结构的root节点TreeNode（红黑树）\_，hash值为实际的用户定义的hashcode

### get ###

1.  按照扰动函数spread对key的hashcode计算index索引下标
2.  如果table中对应下标不为空，并且key的equals判断相等则匹配成功返回value值
3.  如果节点的hash值小于0，则调用节点find方式查找数据，ForwardingNode扩容过渡节点与红黑树均重写了该方法。过渡节点直接返回新表在新表查找对应元素。
4.  如果是红黑树则遍历红黑树节点，判断root节点的附加属性TreeBin的状态
5.  如果当前节点处于WAITER状态或者WRITER状态，则线性的按照链表的方式向下next查找读取对应的值。
6.  如果当前节点既不是WAITER状态，也不是WRITER状态，则递增正在读取数据的数量，递增使用的是加法运算也就是当发生进位的时候低位会全部变为0，意味着写的并发自旋可以竞争写资源；更新当前状态为读READER，遍历树结构二分查找对应的值返回，释放READER状态，
7.  判断如果当前是最后一个读线程READER状态并且存在等待写的线程WAITER状态，等待线程不为空，则唤醒对应的WAITER线程开始写数据

final Node<K,V> find(int h, Object k) {
        
        if (k != null) {
        
            for (Node<K,V> e = first; e != null; ) {
        
                int s; K ek;
                // 如果当前状态处于WAITER等待或者WRITER写状态，则线性的查找对应的值
                if (((s = lockState) & (WAITER|WRITER)) != 0) {
        
                    if (e.hash == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                    // 线性查找数据
                    e = e.next;
                }
                // 更新当前状态为读READER，同时递增读取数据线程数，遍历树结构二分查找数据
                else if (U.compareAndSwapInt(this, LOCKSTATE, s,
                                             s + READER)) {
        
                    TreeNode<K,V> r, p;
                    try {
        
                        p = ((r = root) == null ? null :
                             r.findTreeNode(h, k, null));
                    } finally {
        
                        Thread w;
                        // 递减二分读取数据线程数
                        // 判断如果当前是最后一个读线程READER状态并且存在等待写的线程WAITER状态，等待线程不为空，则唤醒对应的WAITER线程开始写数据
                        if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
                            (READER|WAITER) && (w = waiter) != null)
                            // 唤醒等待线程
                            LockSupport.unpark(w);
                    }
                    return p;
                }
            }
        }
        return null;
    }

### 问题 ###

Q：什么样的场景会有写的线程阻塞？  
A：hash冲突操作红黑树时会对红黑树的根节点加排它锁，会出现阻塞现象。第一个等待线程阻塞，其他等待线程自旋竞争等待资源或锁资源  
Q：写线程会阻塞读线程吗？  
A：不会，但是写线程会导致读取线程线性读，影响性能  
Q：读线程会阻塞写线程吗？  
A：会，读线程节点会一直读直到发现下一个节点是等待写入的节点便会唤醒写线程。根据这两个分析可以看出，ConcurrentHashMap是偏向读取线程的。也就是更加适合多读少写的场景