【JUC源码】并发容器：ConcurrentHashMap（一）底层结构分析

Myth丶恋晨 2022-12-10 14:31 184阅读 0赞

> ConcurrentHashMap 系列：
> 
>  *  [【JUC源码】并发容器：ConcurrentHashMap（一）底层结构分析][JUC_ConcurrentHashMap]
>  *  [【JUC源码】并发容器：ConcurrentHashMap（二）添加元素及树化源码分析][JUC_ConcurrentHashMap 1]
>  *  [【JUC源码】并发容器：ConcurrentHashMap（三）扩容源码分析][JUC_ConcurrentHashMap 2]
>  *  [【JUC源码】并发容器：ConcurrentHashMap（四）获取 value 源码分析][JUC_ConcurrentHashMap_ value]
>  *  [【JUC源码】并发容器：关于 ConcurrentHashMap 的几个问题][JUC_ ConcurrentHashMap]

我们从类注释上大概可以得到如下信息：

1.  所有的操作都是线程安全的，我们在使用时，无需再加锁；
2.  多个线程同时进行 put、remove 等操作时并不会阻塞，可以同时进行，和 HashTable 不同，HashTable 在操作时，会锁住整个 Map；
3.  迭代过程中，即使 Map 结构被修改，也不会抛 ConcurrentModificationException 异常；
4.  除了数组 + 链表 + 红黑树的基本结构外，新增了转移节点，是为了保证扩容时的线程安全的节点；
5.  提供了很多 Stream 流式方法，比如说：forEach、search、reduce 等等。

从类注释中，我们可以看出 ConcurrentHashMap 和 HashMap 相比，新增了转移节点的数据结构，至于底层如何实现线程安全，转移节点的具体细节，暂且看不出来，接下来我们细看源码。

--------------------

ConcurrentHashMap 继承关系，核心成员变量及主要构造函数：

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    	implements ConcurrentMap<K,V>, Serializable {
        
    	
    	// 数组最大容量
        private static final int MAXIMUM_CAPACITY = 1 << 30;
    
        // 数组的初始容量，容量必须是2的幂次方
        private static final int DEFAULT_CAPACITY = 16;
    
        // java8版本没有用到，主要是用于初始化java7 的Segment的
        private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    
        // 负载因子
        private static final float LOAD_FACTOR = 0.75f;
    
        // 链表转化成红黑树的值
        static final int TREEIFY_THRESHOLD = 8;
    
        // 红黑树转化成链表的值
        static final int UNTREEIFY_THRESHOLD = 6;
    
        // 达到最小的数组容量链表才会转化成红黑树,否则数组容量小时，树上可能挂载了太多节点
        static final int MIN_TREEIFY_CAPACITY = 64;
    	
    	// 哈希表的数组，volatile修饰保证线程间的可见性
        // 注：jdk8中第一次插入时才会初始化,jdk7是在构造器时就初始化了
        transient volatile Node<K,V>[] table;
    
        // 扩容后的数组
        // 存在的意义：在扩容时，若原数组的一个槽点上所有元素已经被移动到新数组，那么原数组的此槽点就会被置为 ForwardingNode。
        // 			  这时如果有线程来查询value，那么就必须要去新数组查。所以在成员变量中要保存下nextTable
        // 注意：nextTable只有在扩容时才有值，其余时刻为null
        private transient volatile Node<K,V>[] nextTable;
    
        // 控制table初始化和扩容的字段（重要）
        // 0 使用默认容量进行初始化 
        // -1 初始化中，其他线程应该让出CPU
        // -n 表示n-1个线程正在扩容中
        // >0 类似hashmap的threshold ，正常状态下的sizeCtl。具体分为下面两种情况：
        //    如果还未初始化,代表需要初始化的大小；如果table已经初始化,代表table扩容阈值（默认为table大小的0.75)
        private transient volatile int sizeCtl;
    	
    	// Node的Hash值有四种
    	// 第一种：根据key计算出的Hash值，一般>=0
    	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
    
        
        //-------------------------------构造方法---------------------------------
        public ConcurrentHashMap() {
        
        }
    	
    	// 指定容量构造
        public ConcurrentHashMap(int initialCapacity) {
        
            if (initialCapacity < 0)
                throw new IllegalArgumentException();
            // 计算初始化容量，若>max则=max，否则用tableSizeFor()进行计算
            int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                       MAXIMUM_CAPACITY :
                       tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
            // 将cap附给sizeCtl，去进行数组初始化容量
            this.sizeCtl = cap;
        }
        
        // 传入一个Map进行构造
        public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
        
           // 初始化容量=16
           this.sizeCtl = DEFAULT_CAPACITY;
            putAll(m);
        }
    	
    	// 指定容量和扩容因子构造
        public ConcurrentHashMap(int initialCapacity, float loadFactor) {
        
            this(initialCapacity, loadFactor, 1);
        }
    	
    	//......
    }

### 1.链表节点：Node ###

static class Node<K,V> implements Map.Entry<K,V> {
        
            final int hash; // 当前node的hash值,>=0
            final K key;
            volatile V val;
            volatile Node<K,V> next; // 当拉链时需要next指针
    
            Node(int hash, K key, V val, Node<K,V> next) {
        
                this.hash = hash;
                this.key = key;
                this.val = val;
                this.next = next;
            }
    		
    		// Node的基本方法们，用于获取Node的成员变量
            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; }
            public final V setValue(V value) {
        
                throw new UnsupportedOperationException();
            }
    		
    		// equals方法，只要key和val相同node就相同
            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)));
            }
           	
           	// 根据hash值和key寻找node 
            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;
            }
        }

### 2.树节点：TreeNode & TreeBin ###

树节点类，另外一个核心的数据结构。当链表长度过长的时候，会转换为 TreeNode。

> 注：TreeNode 在 ConcurrentHashMap 继承自 Node 类，而并非 HashMap 中的继承自 LinkedHashMap.Entry<K,V>类，也就是说 TreeNode 带有 next 指针，这样做的目的是方便基于 TreeBin 的访问。

// 红黑树节点
    // 注意：这里是继承了Node,不是 HashMap 的extends LinkedHashMap.Entry<K,V> 
    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;
            }
    		
    		// 通过 hash、key 查找 value
            Node<K,V> find(int h, Object k) {
        
                return findTreeNode(h, k, null);
            }
    		
            // 同 HashMap 的红黑树 find 操作
            final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
        
                if (k != null) {
        
                    TreeNode<K,V> p = this;
                    do  {
        
                        int ph, dir; K pk; TreeNode<K,V> q;
                        TreeNode<K,V> pl = p.left, pr = p.right;
                        if ((ph = p.hash) > h)
                            p = pl;
                        else if (ph < h)
                            p = pr;
                        else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
                            return p;
                        else if (pl == null)
                            p = pr;
                        else if (pr == null)
                            p = pl;
                        else if ((kc != null ||
                                  (kc = comparableClassFor(k)) != null) &&
                                 (dir = compareComparables(kc, k, pk)) != 0)
                            p = (dir < 0) ? pl : pr;
                        else if ((q = pr.findTreeNode(h, k, kc)) != null)
                            return q;
                        else
                            p = pl;
                    } while (p != null);
                }
                return null;
            }
    		
    		//.......
        }

**TreeBin**

但是与 HashMap 不相同的是，在链表转换为红黑树时，它并不是直接转换为红黑树，而是

1.  遍历所有 Node，并构造相应 TreeNode（注：是通过 prev 已立案表形式组织节点，并没有用到 next和parent）
2.  用链表的头 TreeNode 构造 TreeBin（注：在构造 TreeBin 时将 TreeNode 组织成红黑树）
3.  用 TreeBin 替换当前扩容槽点的 Node

所以，最终放到数组中的不是头个 TreeNode 而是一个 TreeBin（相当与一个工具容器）

// 不会存储真实数据（key的value），但是对红黑树的结构进行了管理
        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;	
    	
            // 构造函数，入参是所有 TreeNode 的头结点（当前还是以链表形式组织）
            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;
                        // 将这些 TreeNode 组织成红黑树
                        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);
            }
    
    		//......
    	}

这样的好处是，可以将所有对 TreeNode 的操作规整到一块，即放到 TreeBin 中，比如在 HashMap 中通在红黑树中查找 key 时，需要通过 TreeNode\#root() 通过遍历找到根节点，然后再通过根节点的 TreeNode\#find() 找到 key 对应的 value。即 TreeNode\#getTreeNode() -> TreeNode\#root() -> root\#find().

而 TreeBin 中就直接保存了 root，所以 TreeBin\#find() -> TreeBin.root\#find() 就可以了，没有了寻找 root 的过程。

### 3.扩容转发节点：ForwardingNode ###

ForwardingNode 不是一种新节点，而是一个有着特殊意义的 Node（原因是继承了Node）；它的 Hash 值是Moved（-1），key value next 指针全部为 null。

ForwardingNode 放在在扩容时，已经被处理过过的槽点，表示该槽点的数据（null/单Node/链表/红黑树）已经被移动扩容后的新数组了。

它的作用如下：

1.  作用一：扩容。在原数组中，若某个槽位上是 ForwardingNode，并且有线程要在这个槽点上 put 操作时，会使等待扩容完成后再 put，等待时也可能会协助扩容（注：协助扩容就是帮助某个为扩容移动的槽点进行扩容移动；帮助条件是扩容未结束，且未达到扩容最大线程数）
2.  作用二：转发。在原数组中，若某个槽位上是 ForwardingNode，并且有线程要在这个槽点上 get 操作时，会通过ForwardingNode 的 nextTable 指针找到扩容后的新数组，即通过里面定义的 find 方法从 nextTable 里进行查询节点，而不是以自身为头节点进行查找。

> 注：当某个槽点在扩容移动未完成时，并不会放入 ForwardingNode，但是要 put 到这个槽点的线程仍然要等待；因为在扩容移动时，该槽点会被加锁；所以，等待的最终结果其实就是 ForwardingNode；所以，最终要么就是上面的继续等待整个扩容完成，要么就是去协助扩容。

// 继承链表的Node
     	// 扩容转发节点，放置此节点， 对原有hash槽的操作转化到新的 nextTable 上
        static final class ForwardingNode<K,V> extends Node<K,V> {
        
        
            final Node<K,V>[] nextTable; // 扩容后的新数组，与ConcurrentHashMap的成员变量nextTable指向的是同一个数组
            
            // 构造函数是传入一个数组，主要做两件事
            ForwardingNode(Node<K,V>[] tab) {
        
                // 1.构造一个hash=MOVED(-1)的Node，表示正在转移
                // key=null value=null next=null
                super(MOVED, null, null, null);
                
                // 2.通过ForwardingNode的nextTable记录了扩容时生成的新数组
                // 正是通过此属性，可以建立了新老数组的联系，可以让线程协作扩容
                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;
                    }
                }
            }
        }

### 4.Unsafe 静态代码块 ###

unsafe 代码块控制了一些属性的修改工作，比如最常用的 SIZECTL 。 在jdk8的concurrentHashMap中，大量应用来unsafe的CAS方法进行变量、属性的修改工作。 利用CAS进行无锁操作，可以大大提高性能。

private static final sun.misc.Unsafe U;
    	// 用来获取指定成员变量地址的
        private static final long SIZECTL;
        private static final long TRANSFERINDEX;
        private static final long BASECOUNT;
        private static final long CELLSBUSY;
        private static final long CELLVALUE;
        private static final long ABASE;
        private static final int ASHIFT;
     
        static {
        
            try {
        
                U = sun.misc.Unsafe.getUnsafe();
                Class<?> k = ConcurrentHashMap.class;
                SIZECTL = U.objectFieldOffset
                    (k.getDeclaredField("sizeCtl"));
                TRANSFERINDEX = U.objectFieldOffset
                    (k.getDeclaredField("transferIndex"));
                BASECOUNT = U.objectFieldOffset
                    (k.getDeclaredField("baseCount"));
                CELLSBUSY = U.objectFieldOffset
                    (k.getDeclaredField("cellsBusy"));
                Class<?> ck = CounterCell.class;
                CELLVALUE = U.objectFieldOffset
                    (ck.getDeclaredField("value"));
                Class<?> ak = Node[].class;
                ABASE = U.arrayBaseOffset(ak);
                int scale = U.arrayIndexScale(ak);
                if ((scale & (scale - 1)) != 0)
                    throw new Error("data type scale not a power of two");
                ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
            } catch (Exception e) {
        
                throw new Error(e);
            }
        }

[JUC_ConcurrentHashMap]: https://blog.csdn.net/weixin_43935927/article/details/108810815
[JUC_ConcurrentHashMap 1]: https://blog.csdn.net/weixin_43935927/article/details/113943932
[JUC_ConcurrentHashMap 2]: https://blog.csdn.net/weixin_43935927/article/details/113945436
[JUC_ConcurrentHashMap_ value]: https://blog.csdn.net/weixin_43935927/article/details/113950339
[JUC_ ConcurrentHashMap]: https://blog.csdn.net/weixin_43935927/article/details/108819013