ConcurrentHashMap的实现原理

目录一、为什么用ConcurrentHashMap？二、ConcurrentHashMap结构及初始化1.ConcurrentHashMap结构2. 初始化三、ConcurrentHashMap的API1. put()源码2. get()源码四、参考资料一、为什么用ConcurrentHashMap？ConcurrentHashMap线程安全且高效的操作，如下图对比所示。而HashMap非线程安全

爱我所爱0505

1592人浏览 · 2022-04-16 17:27:26

爱我所爱0505 · 2022-04-16 17:27:26 发布

一、为什么用ConcurrentHashMap？

二、ConcurrentHashMap结构及初始化

1. ConcurrentHashMap结构

2. 初始化

三、ConcurrentHashMap的API

1. put()源码

2. get()源码

四、参考资料

一、为什么用ConcurrentHashMap？

ConcurrentHashMap线程安全且高效的操作，如下图对比所示。而HashMap非线程安全，在put操作时，易出现死循环，即：链表会形成环形（next元素指向前一个元素）。

类型	特点
HashMap	1. 非线程安全 2. 多线程put操作下，易出现死循环
HashTable	1. 线程安全 2. 底层synchronized，并发效率低
ConcurrentHashMap	1. 线程安全 3. 底层volatile且自旋CAS操作，并发效率高 2. 结构： JDK7：Segment数组 + HashEntry数组 + 链表（锁分段技术 _ 锁Segment的数据段） JDK8：Node数组 + 链表/红黑树 (Node数组元素加锁)

二、ConcurrentHashMap结构及初始化

JDK7和JDK8的结构有很大不同，JDK7采用：Segment数组 + HashEntry数组 + 链表的锁分段技术；而JDK8采用：Node数组 + 链表/红黑树。其线程安全都是使用volatile修饰且自旋CAS操作。这里介绍JDK8的结构。

1. ConcurrentHashMap结构

如上图， java8 ConcurrentHashMap结构采用Node数组 + 链表/红黑树。当链表达到一定长度（TREEIFY_THRESHOLD=8）时，链表会转换成红黑树；若减少元素，红黑树可以再次转换成链表。

为了保证线程安全，ConcurrentHashMap主要变量，如：Node<K,V>[] table、Node<K,V>[] nextTable、long baseCount、int sizeCtl都是volatile修饰。同时Node中V val、Node<K,V> next也都是volatile修饰。这些主要变量与HashMap中的主要变量的最大区别，ConcurrentHashMap有volatile修饰，才能保证线程安全性。

2. 初始化

ConcurrentHashMap使用赖初始化Node<K,V>[] table，即：添加第一个元素时初始化table变量，调用initTable()方法完成初始化。

如下代码所示。通过自旋和CAS操作完成初始化。sizeCtl变量决定当前初始化状态。

/**
 * 根据sizeCtl变量完成table变量的初始化
 */
private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    // table为null或长度为0
    while ((tab = table) == null || tab.length == 0) {
        // 读volatile字段sizeCtl
        // sizeCtl = -1时，表示其他线程CAS正在初始化
        // sizeCtl = -x时，表示其他线程正在扩容
        if ((sc = sizeCtl) < 0)
            // 中止当前线程，让出CPU资源
            Thread.yield(); // lost initialization race; just spin
        // 初始化table
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

三、ConcurrentHashMap的API

1. put()源码

如下代码所示。主要逻辑如下：

step1：根据key计算出hashcode；
step2：判断是否需要进行初始化；
step3：即为当前 key 定位出的Node，如果为空表示当前位置可以写入数据，利用CAS尝试写入，失败则自旋保证成功；
step4：如果当前位置的hashcode == MOVED == -1，则需要进行扩容；
step5：如果都不满足，则利用synchronized锁写入数据；
step6：如果数量大于TREEIFY_THRESHOLD则要转换为红黑树。

/**
 * 添加元素
 */
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) {
    // key、value不能为null
    if (key == null || value == null) throw new NullPointerException();
    // 获取key的hash
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        // f：目标位置元素
        // fh：后面存放目标位置的元素hash值
        Node<K,V> f; int n, i, fh;
        // table为null或长度0，则初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // 目标位置元素为null，则直接放入，成功跳出
        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
        }
        // 判定是否需要扩容
        // 当前位置的hashcode == MOVED == -1，则需要进行扩容
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            // 使用 synchronized 加锁加入节点
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    // 链表
                    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;
                            }
                        }
                    }
                    // 红黑树
                    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;
                        }
                    }
                }
            }
            // 数量大于 TREEIFY_THRESHOLD 则要转换为红黑树
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}

2. get()源码

如下代码所示。主要逻辑如下：

step1：根据hash值计算位置；
step2：查找到指定位置，如果头节点就是要找的，直接返回它的value；
step3：如果头节点hash < 0，说明正在扩容或者是红黑树，查找之；
step4：如果是链表，遍历查找之。

/**
 * 获取key对应value
 */
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 计算key的hash，确定table的元素位置
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
        // 如果指定位置元素存在，头结点hash值相同
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // 头结点hash值小于0，说明正在扩容或者是红黑树，find查找
        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;
}