深入理解ConcurrentHashMap

港控/mmm° 2022-05-04 15:26 339阅读 0赞

## 前言 ##

本文的分析的源码是JDK8的版本，上一节我们介绍了HashMap，提到了多线程避免扩容时出现死循环，时要使用ConcurrentHashMap，下面我来讲解一下ConcurrentHashMap这个东西简单的基本原理是什么，我们为什么用。

## 正文 ##

如何保证HashMap的线程安全呢？

方法有很多，比如使用HashTable或者Collections.synchronizedMap，但是这两位选手都有一个共同的问题：性能。因为不管是读还是写操作，他们都会给整个集合上锁，导致同一时间的其他操作被阻塞。并且HashTable和Collections.synchronizedMap解决了HashMap的线程不安全的问题（言外之意HashTable和Collections.synchronizedMap线程安全的），但是带来了运行效率不佳的问题。

基于以上所述，兼顾了线程安全和运行效率的ConcurrentHashMap就出现了，ConcurrentHashMap利用CAS算法。它沿用了与它同时期的HashMap版本的思想，底层依然由“数组”+链表+红黑树的方式思想。

下面我贴了很多源码，其实重要的目的是方便大家阅读，我只做了一个体力活，给大家翻译成了汉语。

### ConcurrentHashMap默认配置参数 ###

/* ---------------- Constants -------------- */
    	//最大容量
        private static final int MAXIMUM_CAPACITY = 1 << 30;
    	//默认容量
        private static final int DEFAULT_CAPACITY = 16;
    	//array的大小，只用于toarray方法
        static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    	//根据这个数来计算segment的个数,segment的个数是仅小于这个数且是2的几次方的一个数（ssize）
        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;
    	// 在转变成树之前，还会有一次判断，只有键值对数量大于 64 才会发生转换。这是为了避免在哈希表建立初期，多个键值对恰好被放入了同一个链表中而导致不必要的转化。
        static final int MIN_TREEIFY_CAPACITY = 64;
    	//每次进行转移的最小值这个值作为一个下限来避免Rsisize遇到过多的内存争用
        private static final int MIN_TRANSFER_STRIDE = 16;
    	// 生成sizeCtl所使用的bit位数
        private static int RESIZE_STAMP_BITS = 16;
    	// 进行扩容所允许的最大线程数
        private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
        // 记录sizeCtl中的大小所需要进行的偏移位数
        private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
     
        // 一系列的标识
        static final int MOVED     = -1; // hash for forwarding nodes
        static final int TREEBIN   = -2; // hash for roots of trees
        static final int RESERVED  = -3; // hash for transient reservations
        static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
    	// 获取可用的CPU个数
        static final int NCPU = Runtime.getRuntime().availableProcessors();
        // 进行序列化的属性
        private static final ObjectStreamField[] serialPersistentFields = {
            new ObjectStreamField("segments", Segment[].class),
            new ObjectStreamField("segmentMask", Integer.TYPE),
            new ObjectStreamField("segmentShift", Integer.TYPE)
        };

### 重要的一个对象Node ###

Node是最核心的内部类，它包装了key-value键值对，所有插入ConcurrentHashMap的数据都包装在这里面，类似于一个HashMap

//Node节点定义，可以看出Node是一个键值对
        static class Node<K,V> implements Map.Entry<K,V> {
            final int hash;
            final K key;
    		//ConcurrentHashMap中的HashEntry相对于HashMap中的Entry有一定的差异性：HashEntry中的value以及next都被volatile修饰，这样在多线程读写过程中能够保持它们的可见性。
            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;
            }
     
            public final K getKey()       { return key; }
            public final V getValue()     { return val; }
            public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
            public final String toString(){ return key + "=" + val; }
    		//不允许直接setValue
            public final V setValue(V value) {
                throw new UnsupportedOperationException();
            }
     
            public final boolean equals(Object o) {
                Object k, v, u; Map.Entry<?,?> e;
                return ((o instanceof Map.Entry) &&
                        (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
                        (v = e.getValue()) != null &&
                        (k == key || k.equals(key)) &&
                        (v == (u = val) || v.equals(u)));
            }
     
            /**
             * 它增加了find方法辅助map.get()方法。
             */
            Node<K,V> find(int h, Object k) {
                Node<K,V> e = this;
                if (k != null) {
                    do {
                        K ek;
                        if (e.hash == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                    } while ((e = e.next) != null);
                }
                return null;
            }
        }
       这个Node内部类与HashMap中定义的Node类很相似，但是有一些差别  
       它对value和next属性设置了volatile同步锁  
       它不允许调用setValue方法直接改变Node的value域  
       它增加了find方法辅助map.get()方法

这里贴了一个能说明ConcurrentHashMap使用CAS算法的部分源码了，读者自行阅读源码会发现在ConcurrentHashMap中，随处可以看到U, 大量使用了U.compareAndSwapXXX的方法，这个方法是利用一个CAS算法实现无锁化的修改值的操作，简单说CAS就是对比替换，原理跟使用SVN差不多，千万记住是无锁操作哦。这里不过多解释，我会单独再发博客单独说说CAS……

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

### TreeNode  ###

这是一个数据结构，当链表长度过长的时候，会转换为TreeNode，是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。

## TreeBin ##

他是一个对象，包装的很多TreeNode节点，它代替了TreeNode的根节点，也就是说在实际的ConcurrentHashMap“数组”中，存放的是TreeBin对象，而不是TreeNode对象。

这里是它的构造方法源码。可以看到在构造TreeBin节点时，仅仅指定了它的hash值为TREEBIN常量，这也就是个标识为。同时也看到我们熟悉的红黑树构造方法

/**
             * 创建以B为初始节点集的bin.
             */
            TreeBin(TreeNode<K,V> b) {
                super(TREEBIN, null, null, null);
                this.first = b;
                TreeNode<K,V> r = null;
                for (TreeNode<K,V> x = b, next; x != null; x = next) {
                    next = (TreeNode<K,V>)x.next;
                    x.left = x.right = null;
                    if (r == null) {
                        x.parent = null;
                        x.red = false;
                        r = x;
                    }
                    else {
                        K k = x.key;
                        int h = x.hash;
                        Class<?> kc = null;
                        for (TreeNode<K,V> p = r;;) {
                            int dir, ph;
                            K pk = p.key;
                            if ((ph = p.hash) > h)
                                dir = -1;
                            else if (ph < h)
                                dir = 1;
                            else if ((kc == null &&
                                      (kc = comparableClassFor(k)) == null) ||
                                     (dir = compareComparables(kc, k, pk)) == 0)
                                dir = tieBreakOrder(k, pk);
                                TreeNode<K,V> xp = p;
                            if ((p = (dir <= 0) ? p.left : p.right) == null) {
                                x.parent = xp;
                                if (dir <= 0)
                                    xp.left = x;
                                else
                                    xp.right = x;
                                r = balanceInsertion(r, x);
                                break;
                            }
                        }
                    }
                }
                this.root = r;
                assert checkInvariants(root);
            }

### ForwardingNode  ###

一个用于连接两个table的节点类。它包含一个nextTable指针，用于指向下一张表。而且这个节点的key value next指针全部为null，它的hash值为-1. 这里面定义的find的方法是从nextTable里进行查询下一个节点，而不是以自身为头节点进行查找下一个节点，可以比喻成两个车厢之间的链子。

/**
         * 在传输过程中插入箱头的节点.
         */
        static final class ForwardingNode<K,V> extends Node<K,V> {
            final Node<K,V>[] nextTable;
            ForwardingNode(Node<K,V>[] tab) {
                super(MOVED, null, null, null);
                this.nextTable = tab;
            }
    
            Node<K,V> find(int h, Object k) {
                // loop to avoid arbitrarily deep recursion on forwarding nodes
                outer: for (Node<K,V>[] tab = nextTable;;) {
                    Node<K,V> e; int n;
                    if (k == null || tab == null || (n = tab.length) == 0 ||
                        (e = tabAt(tab, (n - 1) & h)) == null)
                        return null;
                    for (;;) {
                        int eh; K ek;
                        if ((eh = e.hash) == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                        if (eh < 0) {
                            if (e instanceof ForwardingNode) {
                                tab = ((ForwardingNode<K,V>)e).nextTable;
                                continue outer;
                            }
                            else
                                return e.find(h, k);
                        }
                        if ((e = e.next) == null)
                            return null;
                    }
                }
            }
        }

# initTable #

这是一个初始化方法，调用它的构造方法仅仅是设置了一些参数而已。而整个table的初始化是在向ConcurrentHashMap中插入元素的时候发生的。如调用put、computeIfAbsent、compute、merge等方法的时候都有调用initTable方法的代码，调用时机是检查table==null。

主要应用了关键属性sizeCtl ，如果这个值小于0，表示其他线程正在进行初始化，就放弃这个操作。ConcurrentHashMap的初始化只能由一个线程完成。如果获得了初始化权限，就用CAS方法将sizeCtl置为-1，防止其他线程进入。初始化数组后，将sizeCtl的值改为0.75\*n

/**
         * 使用大小记录的大小初始化表.
         */
        private final Node<K,V>[] initTable() {
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) {
    //sizeCtl表示有其他线程正在进行初始化操作，把线程挂起。对于table的初始化工作，只能有一个线程在进行。
                if ((sc = sizeCtl) < 0)
                    Thread.yield(); // lost initialization race; just spin
                else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//利用CAS方法把sizectl的值置为-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);//相当于0.75*n 设置一个扩容的阈值
                        }
                    } finally {
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

### 扩容transfer  ###

无论是HashMap还是ConcurrentHashMap使用的时候都需要扩容的，下面我只简单的说，让大家读最少的文字，理解最重要的东西。

上面我提到了ConcurrentHashMap是一个大的table，所以扩容也是针对这个大table操作的，简单的说就是把原来的table复制到一个他两倍大小的nextTable中。这个操作需要两部完成：

第一步：构建一个nextTable,它的容量是原来的两倍，这个操作是单线程完成的。这个单线程的保证是通过RESIZE\_STAMP\_SHIFT这个常量经过一次运算来保证的。

第二步：就是将原来table中的元素复制到nextTable中，这里允许多线程进行操作。

多线程扩容是如何完成。大体思想就是遍历、复制的过程，如果遍历到的节点是forward节点，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后遍历。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。我说的forwad就是ForwardingNode，就是上面说的指向下一张表的指针。

### PUT ###

ConcurrentHashMap最常用的就是put和get两个方法。现在来介绍put方法，这个put方法依然沿用HashMap的put方法的思想，根据hash值计算这个新插入的点在table中的位置i，如果i位置是空的，直接放进去，否则进行判断，如果i位置是树节点，按照树的方式插入新的节点，否则把i插入到链表的末尾。有一个最重要的不同点就是ConcurrentHashMap不允许key或value为null值。另外由于涉及到多线程，put方法就要复杂一点。在多线程中可能有以下两个情况

1.  如果一个或多个线程正在对ConcurrentHashMap进行扩容操作，当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到，是因为transfer方法中在空结点上插入forward节点，如果检测到需要插入的位置被forward节点占有，就帮助进行扩容；
2.  如果检测到要插入的节点是非空且不是forward节点，就对这个节点加锁，这样就保证了线程安全。尽管这个有一些影响效率，但是还是会比hashTable的synchronized要好得多。

### get方法 ###

get方法比较简单，给定一个key来确定value的时候，必须满足两个条件  key相同  hash值相同，对于节点可能在链表或树上的情况，需要分别去查找。

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());  
            //根据hash值确定节点位置  
            if ((tab = table) != null && (n = tab.length) > 0 &&  
                (e = tabAt(tab, (n - 1) & h)) != null) {  
                //如果搜索到的节点key与传入的key相同且不为null,直接返回这个节点    
                if ((eh = e.hash) == h) {  
                    if ((ek = e.key) == key || (ek != null && key.equals(ek)))  
                        return e.val;  
                }  
                //如果eh<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;  
        }

### mappingCount与Size方法 ###

对于ConcurrentHashMap来说，这个table里到底装了多少东西其实是个不确定的数量，因为不可能在调用size()方法的时候像GC的“stop the world”一样让其他线程都停下来让你去统计，因此只能说这个数量是个估计值。对于这个估计值，ConcurrentHashMap也是大费周章才计算出来的。

mappingCount与size方法的类似 ，是JDK1.8的新方法，从给出的注释来看，应该使用mappingCount代替size方法，mappingCount其实也是一个大概的数值，因此可能在统计的时候有其他线程正在执行插入或删除操作。

/**
         * 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
        }
    
    
    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;
        }

注：对本文有异议或不明白的地方微信探讨，wx：15524579896