ConcurrentHashMap源码解析（基于java8）

曾经终败给现在 2022-05-19 09:22 272阅读 0赞

### 文章目录 ###

*  ConcurrentHashMap源码解析（基于java8）
 *   *  hashMap的问题
     *  数据结构
     *  继承关系
     *  几个主要方法分析
     *   *  putVal()
         *  spread
         *  initTable
         *  get
     *  总结

# ConcurrentHashMap源码解析（基于java8） #

## hashMap的问题 ##

hashmap在多线程下重哈希（resize）会导致get的时候死循环。

参考：  
[疫苗：JAVA HASHMAP的死循环][JAVA HASHMAP]

主要由于线程二已经改变了数据的结构。  
是 key(7)的下一个节点变为了key(3)。  
之后线程一继续执行的时候，key(3)变成了key(7)所在桶的首，key(7)的下一个指向key(3)，链表形成了死循环。

![HashMap02][]

针对这个问题：  
[https://bugs.java.com/bugdatabase/view\_bug.do?bug\_id=6423457][https_bugs.java.com_bugdatabase_view_bug.do_bug_id_6423457]

> Doug Lea writes:  
> “This is a classic symptom of an incorrectly synchronized use of  
> HashMap. Clearly, the submitters need to use a thread-safe  
> HashMap. If they upgraded to Java 5, they could just use  
> ConcurrentHashMap. If they can’t do this yet, they can use  
> either the pre-JSR166 version, or better, the unofficial backport  
> as mentioned by Martin. If they can’t do any of these, they can  
> use Hashtable or synchhronizedMap wrappers, and live with poorer  
> performance. In any case, it’s not a JDK or JVM bug.”

> Doug Lea(java并发编程领域的大师，很多并发库都是出自他之手)说的很清楚，hashmap不是线程安全的，java5之后的版本可以使用ConcurrentHashMap确保线程安全，java5之前可以使用hashTable或synchhronizedMap包裹hashMap。  
> 并表示这不是jdk或jvm的bug

## 数据结构 ##

java8对hashmapd的底层数据接口进行了调整，ConcurrentHashMap的底层结构使用和hashMap是一致的： 数组+链表+红黑树。

java8之前,ConcurrentHashMap使用的是段锁，即锁定一部分桶，保证并发性。

在put操作的时候，使用sychronized锁住一个桶,对桶中的链表/红黑树进行操作。

static class Node<K,V> implements Map.Entry<K,V> { 
            final int hash;
            final K key;
            volatile V val;//使用volatile保证节点值的并发正确性
            volatile Node<K,V> next;//使用volatile保证对节点的下一个节点操作的并发正确性
    
            Node(int hash, K key, V val, Node<K,V> next) { 
                this.hash = hash;
                this.key = key;
                this.val = val;
                this.next = next;
            }
        }

## 继承关系 ##

![123][]

![456][]

可以看到ConcurrentHashMap和hashMap都是继承AbstractMap及实现Map,Serializable接口。

## 几个主要方法分析 ##

由于java8的hashMap 我们之前已经分析过了，这里分析ConcurrentHashMap主要分析它是如何保证并发的

### putVal() ###

final V putVal(K key, V value, boolean onlyIfAbsent) { 
            if (key == null || value == null) throw new NullPointerException();//ConcurrentHashMap不存储key/value为null
            int hash = spread(key.hashCode());//计算key的hash值
            int binCount = 0;//桶中元素的大小，如果大于等于8，则旋转为红黑树
            for (Node<K,V>[] tab = table;;) { 
                Node<K,V> f; int n, i, fh;//f :计算出key在 hash中数组桶的链表/红黑树的根，n:桶的长度，i:key在数组的位置,fh：f的hash值
                if (tab == null || (n = tab.length) == 0)
                    tab = initTable();//如果数组为0或者没有元素，初始化
                else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //f指向数组中的值（链表/红黑树）
                    if (casTabAt(tab, i, null,
                                 new Node<K,V>(hash, key, value, null)))//如果f为 null表示链表没有值，则此次放入的key/value存入链表的根 (1)
                        break;                   // no lock when adding to empty bin
                }
                else if ((fh = f.hash) == MOVED)//如果在进行扩容先进行扩容
                    tab = helpTransfer(tab, f);
                else { 
                    V oldVal = null;
                    synchronized (f) { //此次添加key/value的链表有元素，则对链表/红黑树进行加锁 这里是java8和java7的不同之处 java8采用的加锁桶，而不是一段桶
                        if (tabAt(tab, i) == f) { //出现hash碰撞（情况1:线程1和线程2同时插入在上面(1) 由于是CAS操作只有一个线程会成功，第二个线程会进入到这一步 情况2:普通的hash碰撞），
                            if (fh >= 0) { //大于0表示桶是链表 TREEBIN = -2 桶是红黑树
                                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)))) { //如何hash和key都和已存在的元素相等则根据onlyIfAbsebt的值，确定是用之前的值还是新值覆盖
                                        oldVal = e.val;
                                        if (!onlyIfAbsent)//如果onlyIfAbsent为fasle，新值覆盖老值
                                            e.val = value;
                                        break;//退出,操作完成
                                    }
                                    Node<K,V> pred = e;//链表最末尾的值作为新值的前一个元素
                                    if ((e = e.next) == null) { //如果已经到了末尾值，则创建新的node存放此次插入的key/value
                                        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) { //如果不等于0判断是否需要旋转为红黑树
                        if (binCount >= TREEIFY_THRESHOLD)//如果大于8则旋转为红黑树
                            treeifyBin(tab, i);//旋转为红黑树
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            addCount(1L, binCount);
            return null;
        }

上面的代码中，我们没有具体介绍1.扩容 2.初始化 3. 链表旋转为红黑树,  
那么接下来我们详细了解一下

### spread ###

static final int spread(int h) { 
            return (h ^ (h >>> 16)) & HASH_BITS;
        }

h是原始的hash返回的值是int类型，int取值范围:-2147483648到2147483648,前后加起来大概四十亿的映射空间。只要hash函数映射的比较松散，一般是很难出现碰撞的。  
但是考虑到实际的内存的大小，很难放下这么大的数组。

所以为了空间上的考虑上述中的扰动函数，对原始计算出来的hash值（int 四个字节32位），右移16位，自己的高半区和低半区做异或，就是为了混合原始hash值的高位和地位，以此来加大低位的随机性。而且混合后的地位参杂了高位的部分特征，这样高位的信息也被变相的保留下来了。

### initTable ###

初始化数组

private final Node<K,V>[] initTable() { 
            Node<K,V>[] tab; int sc;//tab指向数组，sc（sizeCtl）数组初始化/扩容标志位
            while ((tab = table) == null || tab.length == 0) { //数组为null或长度为o的可以进行初始化，
                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;//默认数组大小为16
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//创建数组
                            table = tab = nt; //table是volatile的
                            sc = n - (n >>> 2);//n=16的话 sc = 16-4 =12 = 16*0.75
                            //sc的大小等于扩容阈值
                        }
                    } finally { 
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

/** * Table initialization and resizing control. When negative, the * table is being initialized or resized: -1 for initialization, * else -(1 + the number of active resizing threads). Otherwise, * when table is null, holds the initial table size to use upon * creation, or 0 for default. After initialization, holds the * next element count value upon which to resize the table. */
        private transient volatile int sizeCtl;
        //变量被volatile修饰，表示变量是同步的，等于-1表示数组正在初始化，扩容：-（1+扩容线程大小）

### get ###

get方法比较简单，需要注意桶的hash值如果小于0则表示正在扩容或者是红黑树

public V get(Object key) { 
            Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
            int h = spread(key.hashCode());//计算hash值
            if ((tab = table) != null && (n = tab.length) > 0 &&
                (e = tabAt(tab, (n - 1) & h)) != null) { //如果key是存在则继续在桶中查找，否则返回null表示没有
                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;
        }

## 总结 ##

主要目的是了解java8的ConcurrentHashMap是如何保证并发安全的。

[JAVA HASHMAP]: https://coolshell.cn/articles/9606.html
[HashMap02]: /images/20220519/42f61c9156bb4a9a9ddfb624a81ce08e.png
[https_bugs.java.com_bugdatabase_view_bug.do_bug_id_6423457]: https://bugs.java.com/bugdatabase/view_bug.do?bug_id=6423457
[123]: /images/20220519/f364aa40aa9243b0867b65191532c6ac.png
[456]: /images/20220519/05db02f90bab4afe90e5cfd3459f3dc8.png