【集合】ConcurrentHashMap 源码分析

男娘i 2022-12-26 11:19 126阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]

主要分析 JDK8 版本源码。

[HashTable 源码分析][HashTable]

[HashMap 源码分析][HashMap]

[SynchronizedMap 源码分析][SynchronizedMap]

# 一 概述 #

ConcurrentHashMap 是一种支持高并发线程安全的散列表，通过 **CAS + synchronized + volatile** 保证线程安全。

# 二 实例 #

package com.jpeony.collection.hash;
    
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * 高并发，高性能，线程安全
     * jdk1.7 分段锁保证线程安全
     * jdk1.8 CAS + synchronized + volatile 保证线程安全
     *
     * @author yihonglei
     */
    public class ConcurrentHashMapTest {
        public static void main(String[] args) {
            ConcurrentHashMap<String, String> conMap = new ConcurrentHashMap<>();
    
            conMap.put("key1", "hello world!");
            System.out.println(conMap.get("key1"));
        }
    }

**运行结果**

![20201205113927913.png][]

# 三 ConcurrentHashMap 底层实现 #

**重点分析 put() 和 get() 方法。**

## 类结构 ##

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
        implements ConcurrentMap<K,V>, Serializable {
        private static final long serialVersionUID = 7249069246763182397L;
    
        /**
         * 最大数组容量
         */
        private static final int MAXIMUM_CAPACITY = 1 << 30;
    
        /**
         * 初始化默认数组容量
         */
        private static final int DEFAULT_CAPACITY = 16;
    
         /**
          * 数组最大容量
          */
         static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    
         /**
          * 最大并发度，即最多运行多少个线程同时进行访问
          */
         private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    
         /**
          * 负载因子
          */
         private static final float LOAD_FACTOR = 0.75f;
    
         /**
          * 链表转换为红黑树阀值（链表长度>=8，转换为红黑树）
          */
         static final int TREEIFY_THRESHOLD = 8;
    
         /**
          * 红黑树转换为链表阀值（当红黑树节点个数<=6，转换为链表）
          */
         static final int UNTREEIFY_THRESHOLD = 6;
    
         /**
          * 多线程间共享的竞态资源，用于维护各种状态，保存各类信息。
          * 1）sizeCtl默认值为0；
          * 2）sizeCtl > 0，当前数组尚未初始化时，表示初始化容量和扩容阀值，如果已经初始化，则表示扩容后的阀值；
          * 3）sizeCtl = -1，表示当前数组正在初始化或者正在进行扩容；
          * 4）sizeCtl < -1: 担扩容时，标识符（高16位）和参与线程数目（低16位）的存储。需要扩容时，使用CAS设置为sizeCtl+1，完成扩容时，使用CAS设置为sizeCtl-1；
          */
        private transient volatile int sizeCtl;
    
         /**
          * cpu 核数
          */
         static final int NCPU = Runtime.getRuntime().availableProcessors();
    
         /**
          * 链表节点
          */
         static class Node<K,V> implements Map.Entry<K,V> {
            // 哈希值
            final int hash;
            // 键值
            final K key;
            // value值，这里跟 HashMap 的区别在于用 volatile 修饰，保证多线程的可见性
            volatile V val;
            // 指向下一个节点的引用（指针），这里跟 HashMap 的区别在于用 volatile 修饰，保证多线程的可见性
            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;
            }
        }
        /**
         * 红黑树节点
         */
        static final class TreeNode<K,V> extends Node<K,V> {
            TreeNode<K,V> parent;  // red-black tree links
            TreeNode<K,V> left;
            TreeNode<K,V> right;
            TreeNode<K,V> prev;    // needed to unlink next upon deletion
            boolean red;
    
            TreeNode(int hash, K key, V val, Node<K,V> next,
                     TreeNode<K,V> parent) {
                super(hash, key, val, next);
                this.parent = parent;
            }
            // ......
        }
    
    
        static final class TreeBin<K,V> extends Node<K,V> {
            TreeNode<K,V> root;
            volatile TreeNode<K,V> first;
            volatile Thread waiter;
            volatile int lockState;
            // values for lockState
            static final int WRITER = 1; // set while holding write lock
            static final int WAITER = 2; // set when waiting for write lock
            static final int READER = 4; // increment value for setting read lock
            // ......
        }
    
        // ......
    }

## 构造器 ##

/**
     * 无参构造器
     */
    public ConcurrentHashMap() {
    }
    
    /**
     * 固定容量构造器
     */
    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;
    }
    
    // ......

默认是无参构造器，可以指定初始化容量等。

## ConcurrentHashMap\#put() ##

**put()** 往散列表添加元素入口。

/**
     * 添加元素入口
     */
    public V put(K key, V value) {
        return putVal(key, value, false);
    }

**putVal()** 添加元素实现。

1）初始化数组；

2）hash 冲突先按链表法解决，当链表长度 >= 8时，转换为红黑树；

3）当 key 相同时 新值覆盖掉旧值；

4）当超过数组容量阀值时进行扩容；

看上去逻辑跟 HashMap 一样，但是 ConcurrentHashMap 为了实现高性能、高并发、线程安全的散列表费劲了脑子，

跟 HashMap 的实现代码可以说有天壤之别，不是像 JDK7 哪样简单的 Segment 分段锁加锁搞定的，JDK8 加锁的力度更加的细。

/**
     * 添加元素实现
     */
    final V putVal(K key, V value, boolean onlyIfAbsent) {
        // key 和 value 都不能为空
        if (key == null || value == null) throw new NullPointerException();
        // 计算 hash 值
        int hash = spread(key.hashCode());
        int binCount = 0;
        // 自旋，不断尝试添加元素，因为 table 初始化和 casTabAt() 用到了 compareAndSwapInt、compareAndSwapObject(CAS轻量级锁)
        for (Node<K,V>[] tab = table;;) {
            // 声明从数组取出的元素 f，数组长度 n，取出数组原位置 i，对应 f 元素的hash值
            Node<K,V> f; int n, i, fh;
            // 数组为空，初始化数组（CAS操作）
            if (tab == null || (n = tab.length) == 0)
                // 初始化数组
                tab = initTable();
            // 计算新插入元素在数组中的位置，从数组获取元素，如果没有，即没有hash冲突，通过 casTabAt() 线程安全方式直接添加元素
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                // 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，说明正在进行扩容，添加元素会进入等待阶段，直到扩容完成
            // 通过 helpTransfer() 扩容时的辅助方法，操作添加到扩容后的散列表，不能按照常规方式添加元素
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            // hash 冲突添加元素
            else {
                // 旧节点元素
                V oldVal = null;
                // 加上同步锁（锁链表头结点或红黑树头结点）
                synchronized (f) {
                    // 判断值是否被改变，double-check，因为是一个工业级线程安全并发的HashMap，时时刻刻都不能忘记了并发操作带来的线程安全问题
                    if (tabAt(tab, i) == f) {
                        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;
                                }
                                // 如果 key 不存在，则构建节点添加到链表尾部，并退出循环
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        // 如果是红黑树节点，这里用的是 TreeBin，TreeNode 只是 TreeBin 的一个成员变量
                        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;
                            }
                        }
                    }
                }
                // 如果链表长度大于等于 8，则转换为红黑树
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 判断是否需要扩容，调用 transfer() 方法进行库容
        addCount(1L, binCount);
        return null;
    }

**initTable()** 初始化散列表；

/**
     * 初始化散列表
     */
    private final Node<K,V>[] initTable() {
        // 申明散列表，以及并发线程数
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            // 并发线程 让出 cpu 时间片，稍后执行
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            // CAS 轻量级锁，处理并发初始化问题
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                    // 再次校验散列表是否已经初始化，有点像单利模式的double-check思想，多线程共享资源处理线程安全问题就是这么麻烦的
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        // 初始化数组，默认容量 16
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        // 初始化时 sc 算出来是 12，debug 看下
                        sc = n - (n >>> 2);
                    }
                } finally {
                    // sizeCtl 初始化时是 12
                    sizeCtl = sc;
                }
                break;
            }
        }
        // 返回初始化之后的散列表
        return tab;
    }

**addCount()** 判断是否需要扩容；

/**
     * 判断是否需要扩容
     *
     * @param x 需要增加的元素个数，默认是1
     * @param check 是否需要检查扩容
     */
    private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        if ((as = counterCells) != null ||
            !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 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                // 线程竞争标记状态
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            // 计算数组中元素个数
            s = sumCount();
        }
        // 根据 check 判断是否需要检查扩容
        if (check >= 0) {
            Node<K,V>[] tab, nt; int n, sc;
            // 如果 已经大于阀值，并且小于数组最大容量，有扩容空间，进行扩容操作
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
                // 获取扩容标识
                int rs = resizeStamp(n);
                if (sc < 0) {
                    // 如果sc的低16位不等于rs,表示标识符已经改变
                    // 如果nextTable为空,表示扩容已经结束
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    // CAS 替换 sc 值为 sc+1，成功则开始扩容
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                }
                // sc > 0 表示数组此时并不在扩容阶段，更新sizeCtl并开始扩容
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);
                // 计算扩容后最新数组数量
                s = sumCount();
            }
        }
    }

**transfer()** 扩容操作。

/**
     * 扩容方法
     */
    private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        // 原数组长度 n，stride 为此次需要迁移的桶的数目
        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
            try {
                @SuppressWarnings("unchecked")
                // 扩容为 原数组 2倍容量
                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);
        boolean advance = true;
        // 扩容完成标记
        boolean finishing = false; // to ensure sweep before committing nextTab
        // 自旋扩容
        for (int i = 0, bound = 0;;) {
            Node<K,V> f; int fh;
            // true 表示未完成负责范围的库容操作，完成扩容，退出循环
            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) {
                int sc;
                // 标记为扩容操作完成
                if (finishing) {
                    nextTable = null;
                    table = nextTab;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                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
                }
            }
            // 当前节点为 null，cas 更新节点的值为 fwd 转移节点，转移节点的 hash 值为 MOVED
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            // 判断是否已经完成转移
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                // 加同步锁
                synchronized (f) {
                    // 链表节点迁移
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        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);
                            }
                            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;
                        }
                    }
                }
            }
        }
    }

**helpTransfer()** 扩容辅助添加元素。

/**
     * 扩容辅助方法
     */
    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) {
                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;
    }

## ConcurrentHashMap\#get() ##

**get()** 根据 key 获取 value 最新值，因为 散列表 table 是用 volatile 修饰的，volatile 能够保证内存可见性 和 禁止代码改变指令执行顺序，

所以，每次 put 的值，get 时候都会从主内存刷新到线程的副本变量，永远是最新的。

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) {
            // 如果 key 和 hash 都一样，则查找到目标元素，直接返回对应的 value
            if ((eh = e.hash) == h) {
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }
            // 如果是红黑树，则调用红黑树变量方法
            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、ConcurrentHashMap JDK7 基于数组+链表实现，通过 分段锁 保证线程安全；

2、ConcurrentHashMap JDK8 基于数组+链表+红黑树实现，通过 CAS + synchronized + volatile 保证线程安全，

查找速度更快，并发性更高；

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[HashTable]: https://jpeony.blog.csdn.net/article/details/109698072
[HashMap]: https://jpeony.blog.csdn.net/article/details/110446413
[SynchronizedMap]: https://jpeony.blog.csdn.net/article/details/117566333
[20201205113927913.png]: /images/20221120/ac641a05b3a14207a11b8ac5f38163a3.png