ConcurrentHashMap源码解析详解（JDK1.8面试必备）

野性酷女 2023-05-31 02:45 16阅读 0赞

> # 一、概述 #

1.很多人看源代码只是单纯的看，也许并不知道为什么要看源代码。jdk源代码是经过很多大牛无数次的版本升级、更新最后沉淀下来的精华，如果只是单纯的为了熟悉这个api。完全不需要看源代码，只需要记住api文档就行。看源代码有两个好处。

a.能够学习大牛的设计思想，它为什么是这样设计的，而不是那样设计的，这个也是最重要的。

b.当然在看我一遍源代码后，对里面的方法也更加熟悉，运行也更加熟练，排查问题的能力也能得到提升。

2.不光HashMap在jdk1.8前后有了变化，ConcurrentHashMap也在jdk1.8前后大有不同

3.在jdk1.8之前。ConcurrentHashMap采用的数据结构是数组加链表。

采用的锁机制，是分段锁。将一个数组，分成很多段，每一段都相当于一个HashTable。HashTable采用的是全局锁。当多线程访问时，只要线程两个线程访问的不是同一段数据，便能异步执行。但是这种设计性能将会受到分段个数的数量而限制。如下图，此时分段个数最多为5，也就意味着最多5个线程异步进行操作。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MTUxMzQ2_size_16_color_FFFFFF_t_70][]

３.所以在jdk1.8放弃了分段锁，而采用CAS算法和synchorized关键字。

jdk1.8采用的数据结构和HashMap一致，也是使用的数组加红黑树加链表。对插入元素，如过Node节点为null,则使用CAS算法进行插入，如果节点不为空，则表示该桶里面要么是链表，要么是红黑树，则使用java关键字synchorized将node节点锁住。如果没有发生冲突，则意味着我们的map不会有重量级锁（synchorized）

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MTUxMzQ2_size_16_color_FFFFFF_t_70 1][]

４.jdk1,8前后进行比较，则可以看出，以前的并发受分段个数的限制，现在的并发，只要不是同一个节点（桶），便能同时执行插入操作。

> # 二、源码 #

## ##

### 1.常量，之前HashMap中介绍 ###

在讲解方法之前，我们有必要先了解一下常用的常量和变量，以便在后面的方法中我们不至于对这变量的用途不了解。

1). java虚拟机限制数组的最大使用长度

static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

2).只有当数组容量大于该值时，才允许树行化链表（即将链表转换为红黑树），否则直接进行扩容，不需要转换成红黑树。

static final int MIN_TREEIFY_CAPACITY = 64;

3).用于生成每次扩容都唯一的生成戳的数，最小是6。

private static int RESIZE_STAMP_BITS = 16;

4).最大扩容线程数量

private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;

5).获取cpu的数量

static final int NCPU = Runtime.getRuntime().availableProcessors();

6).特殊节点的hash值，正常hash不会出现负数，forwarding nodes是一个临时节点，在扩容中出现  
     如果旧数组需要将全部节点转移到新数组，会在旧数组放置一个forwarding nodes节点  
     若读操作碰到该节点时，则将操作转发到扩容后的新数组执行，如果是写操作碰到，则尝试帮助扩容

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

### 2.变量 ###

1).扩容后新的table数组，只有在扩容时使用，若nextTable不为空，则表示扩容未结束

private transient volatile Node<K,V>[] nextTable;

2). sizeCtl = -1，表示有线程正在进行真正的初始化操作  
     sizeCtl = -(1 + nThreads)，表示有nThreads个线程正在进行扩容操作  
     sizeCtl > 0，表示接下来的真正的初始化操作中使用的容量，或者初始化/扩容完成后的threshold  
     sizeCtl = 0，默认值，此时在真正的初始化操作中使用默认容量

private transient volatile int sizeCtl;

## ##

### 3.方法 ###

1.)

a.unsafe类，是一个可以直接操控内存的类。不对外提供构造方法。但提供了getUnsafe的方法来获取对象。所以这里unsafe对象只能从启动类加载器加载，

b.而启动类加载器：这个类加载器负责放在<JAVA\_HOME>\\lib目录中的，或者被-Xbootclasspath参数所指定的路径中的，并且是虚拟机识别的类库。用户无法直接使用。

c.用户能直接使用的只有扩展类加载器和应用类加载器，这种设计无疑为类提供更安全的访问限制。

d.objectFieldOffset方法，获取对象内存地址该字段的偏移量。其实就是获取字段的内存地址。

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

2).putVal方法map中一个非常重要的方法，为了方便阅读，我对每一句代码进行了注释，先看一下它是怎么运行的，最好再来说它为什么这么设计。

//put方法；参数：onlyIfAbsent，如果key相同，是否覆盖原来的值,false表示覆盖
        final V putVal(K key, V value, boolean onlyIfAbsent) {
            
            //如果传入key或value为空，抛异常，所以ConcurrentHashMap不允许键或值为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表示桶，ｎ是table的长度,i是node的下标，fn是node的hash值
                Node<K,V> f; int n, i, fh;
    
                //如果table为空，则进行初始化，然后继续循环，所以上面是循环tab = table
                if (tab == null || (n = tab.length) == 0)
                    tab = initTable();
    
                //如果通过hash找到的下标为null
                else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
    
                    //通过CAS算法将键值插入到tab数组中，然后跳出（后面讲这个方法）
                    if (casTabAt(tab, i, null,
                                 new Node<K,V>(hash, key, value, null)))
                        break;                   // no lock when adding to empty bin
                }
    
                //如果hash值为-1，则表明是一个特殊的ForwardingNode节点
                //如果是读操作，则到新数组中去执行
                //这里是写操作，则帮助扩容
                else if ((fh = f.hash) == MOVED)
                    //数组扩容，帮助扩容
                    tab = helpTransfer(tab, f);
                
                //如果下标内容不为空，则进行的操作
                
                else {
                    V oldVal = null;
                    //对这个桶的Node节点f加锁。
                    synchronized (f) {
    
                        
                        if (tabAt(tab, i) == f) {
    
                            //node的hash值大于0,表示不是特殊节点
                            if (fh >= 0) {
                                
                                binCount = 1;
                                for (Node<K,V> e = f;; ++binCount) {
    
                                    //声明变量k
                                    K ek;
    
                                    //如果hash值相同，key的比较也相同，把旧值赋值给oldVal
                                    if (e.hash == hash &&
                                        ((ek = e.key) == key ||
                                         (ek != null && key.equals(ek)))) {
                                        oldVal = e.val;
    
                                        //put方法传入的onlyIfAbsent为false，则将新值给覆盖                                                    　　　　　　　　　　　　　　　　　　　　　　//掉旧值    
                                        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;
                                }
                            }
                        }
                    }
                    if (binCount != 0) {
                       //bintCount在链表中，初始值为１，循环一次加１，则表示链表的长度是否大于８
                        if (binCount >= TREEIFY_THRESHOLD)
                            //将该桶转换成红黑树
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            addCount(1L, binCount);
            return null;
        }

3). initTable()方法，是对容器进行初始化的操作。

//使用sizeCtl中记录的大小初始化表。
        private final Node<K,V>[] initTable() {
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) {
                if ((sc = sizeCtl) < 0)
                    Thread.yield(); // lost initialization race; just spin
                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;
        }

4).在put方法和iniTable方法中都有使用到CAS算法，来看一下CAS算法究竟是什么。下面这是方法是在putVal中使用的一个方法。

目的是新建一个node节点插入到数组下标。

static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                            Node<K,V> c, Node<K,V> v) {
            return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
        }

5).unsafe类的对象在静态代码块中已经获取到了，所以只需要调用它的方法即可，每一个参数都在代码中标明了注释。

/*　在obj的offset位置比较object field和期望的值，如果相同则更新。这个方法
       * 的操作应该是原子的，因此提供了一种不可中断的方式更新object field。
       * 
       * @param obj the object containing the field to modify.
       *    包含要修改field的对象 
       * @param offset the offset of the object field within <code>obj</code>.
       *         <code>obj</code>中object型field的偏移量
       * @param expect the expected value of the field.
       *               希望field中存在的值
       * @param update the new value of the field if it equals <code>expect</code>.
       *               如果期望值expect与field的当前值相同，设置filed的值为这个新值
       * @return true if the field was changed.
       *              如果field的值被更改
       */
      public native boolean compareAndSwapObject(Object obj, long offset,
                                                 Object expect, Object update);

compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);这是map调用的,用来更新对象为tab，我们来看一下这几个参数分别是什么。

a.tab:要跟新的对象

b.((long)i << ASHIFT) + ABASE:这个值获取的是该对象给定数组下标的实际内存地址，接着细看一下，这几个值是怎么来的。

首先：scale:获取用户给定数组寻址的换算因子

Class<?> ak = Node[].class;
    int scale = U.arrayIndexScale(ak);

然后：ASHIFT:通过换算因子获取第一个元素的偏移地址。

ABASE = U.arrayBaseOffset(ak);
    Integer.numberOfLeadingZeros(scale)
    ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);

最后：看一下numberOfLeadingZeros(scale)这个方法的作用

该方法的作用是返回无符号整型i的最高非零位前面的0的个数，包括符号位在内；  
                  如果i为负数，这个方法将会返回0，符号位为1.  
                  比如说，10的二进制表示为 0000 0000 0000 0000 0000 0000 0000 1010  
                  java的整型长度为32位。那么这个方法返回的就是28

Integer.numberOfLeadingZeros(scale)

c.参数c为null,我们期望是null,如果当实际地址的内容也是null时，则进行跟新。

d.new Node<K,V>(hash, key, value, null)这是ｖ。表示将这个node对象更新到指定的内存。

6.当然iniTable方法里面使用的那个unsafe类方法也是同样的道理，使用的是原子操作。

7.这样就实现了更新数组某个下标的内容，但是为什么要这么做了，搞这么复杂，因为unsafe类里面的方法使用的是CAS算法，都是原子操作，如果不这么做，那么我们势必得给这个更新操作加锁，在性能方法不佳。

8. 当然在ConcurrentHashMap中还有很多方法都很经典，一两篇博客肯定是写不完的，有兴趣的可以自己下来分析下那些源代码。

> # 四、为什么这么设计 #

1.有一个java面试题我记得是这样问的，请你谈谈ConcurrentHashMap实现原理，为什么在jdk1.8之后放弃了分段锁，如果是你来设计，你会怎么设计。

答案：

ConcurrentHashMap这个类属于并发容器，主要用于在多线程情况下对数据进行map类型的数据进行存储。在多线程情况势必需要考虑线程安全问题。一个全局的synchronized可以搞定，但是在考虑安全的同时也得考虑性能，最好的设计是在线程安全的情况下性能最优。jdk1.8之前使用分段锁，线程的并非受到了锁个数的现在，而且synchronized属于重量级锁，在性能方面并不好。放弃了分段锁。使用CAS算法，代表着分段个数永远是数值的长度，而且并不会带来重量级锁性能限制的条件，当然如果发生了hash冲突，还是得采用原始的synchorized。

如果是我来设计，在结构这方面不需要改变，要想性能更上一步，就得避免synchorized的使用，也就意味着需要尽可能的不发生hash冲突。需要做的是在hash算法这一部分还能够尽量优化，以此来提高性能。

最后，如果你对文章内容还稍微满意，希望得到更大的提升，可以获取更多的学习资料，面试题以及视频，关注微信公众号，将定时更新各种技术文章，提升知识。

![20191030202900740.jpg][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MTUxMzQ2_size_16_color_FFFFFF_t_70]: /images/20230531/a11f82d2e9d247f5b8d0179edc121851.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1MTUxMzQ2_size_16_color_FFFFFF_t_70 1]: /images/20230531/49fcd3b679114c80b615310c98ca9493.png
[20191030202900740.jpg]: /images/20230531/3e7945d51eb04982a7b9923c5c0118f0.png