图解ConcurrentHashMap的前世今生

Posted java叶新东老师

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了图解ConcurrentHashMap的前世今生相关的知识,希望对你有一定的参考价值。

前言

     首先呢,想要了解ConcurrentHashMap, 你得先了解HashMap,可以看我另一个帖子 : HashMap底层原理以及 LinkedHashMap、HashTable 、HashSet 四者区别

    为什么要先了解HashMap呢? 因为HashMap是线程不安全的类,只适合在单线程上使用,既然使用受限,那就意味着它的结构相对比较简单,所以呢,先学HashMap在来了解ConcurrentHashMap将会更好理解,达到锦上添花的作用,事实也是如此,因为ConcurrentHashMap也是基于HashMap发展而来的;如果你一上来就直接看ConcurrentHashMap的源码,会非常懵逼,源码的可读性都不太好,所以呀,学东西最好由浅入深,不然一下子太难了,学起来也会很费劲,甚至像打退堂鼓;

1、ConcurrentHashMap 是什么?

       ConcurrentHashMap 是JDK1.5之后新出一个在并发包里面类,包名叫 java.util.current;简称JUC,既然叫并发包,那肯定就意味着它是线程安全的,里面有一个概念:分而治之,这是ConcurrentHashMap的核心思想,并且在jdk7里面用到一个非常新颖且时髦的技术 :分段锁;由此可见,ConcurrentHashMap的出现就是为了高并发而准备的;并且使用方式和HashMap一样,用key-value方式存储数据;连方法名都一样;只不过区别是一个线程安全,一个线程不安全。

HashTable的实现

但是不对啊,线程的安全的map不是已经有HashTable了吗?为什么还要正处一个ConcurrentHashMap出来呢?这是个好问题,首先我们先来看看HashTable的实现;

HashTable 的每个修饰为 public 的方法都加上了 synchronized 的同步方法,也就是说,不管我对map的增删改查都会上锁,也正因为它的锁简单粗暴,不管你干嘛我都给你锁住,造成一个原因就是效率低下;高并发场景下,只要有一个线程对HashTable操作,其他线程都会进入阻塞状态,线程数量太多的情况下会造成响应时间缓慢,所以你会看到,现在几乎没人用HashTable来实现线程安全;

2、JDK1.7 中的ConcurrentHashMap实现原理

    在jdk1.7及其以下的版本中,结构是用Segments数组 + HashEntry数组 + 链表实现的,

final Segment<K,V>[] segments;

Segment 继承了ReentrantLock,所以它除了是一个独占锁之外,还是一种可重入锁(ReentrantLock),多个锁同时存在时会自动合并为一把锁来操作,ConcurrentHashMap 使用了分段锁技术来保证线程安全,它把数据分成一段一段的,也就是Segments [] 数组,Segments数组中每一个元素就是一个段,每个元素里面又存储了一个Enter数组,这个enter数组就相当于是一个HashMap,所以在高并发场景下,每次修改内容时只会锁住segment数组的每个元素,多个元素之间各自负责自己的锁,分段后,多个段(元素)之间的插入修改不会有任何影响,既做到了并发,又提升了效率;按照默认的并发级别 concurrentLevel 来说 ,默认是16,所以理论上支持同时16个线程并发操作,并且还互不冲突,是不是很牛掰!

2.1、构造函数

  • 当我们new一个ConcurrentHashMap后, 它会先计算出Segments数组的大小,segment数组的大小是根据 concurrentLevel 计算出来的,ssize 就是Segments数组的大小,并且一定是2的次幂;
  • 默认情况下concurrentLevel是16,则ssize为16;
  • 若concurrentLevel为14,ssize为16;
  • 若concurrentLevel为17,则ssize为32。

为什么Segment的数组大小一定是2的次幂?其实主要是便于通过按位与的散列算法来定位Segment的index

public ConcurrentHashMap(int initialCapacity,
                               float loadFactor, int concurrencyLevel) {
          if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
              throw new IllegalArgumentException();
          //MAX_SEGMENTS 为1<<16=65536,也就是最大并发数为65536
          if (concurrencyLevel > MAX_SEGMENTS)
              concurrencyLevel = MAX_SEGMENTS;
          //2的sshif次方等于ssize,例:ssize=16,sshift=4;ssize=32,sshif=5
         int sshift = 0;
         //ssize 为segments数组长度,根据concurrentLevel计算得出
         int ssize = 1;
         while (ssize < concurrencyLevel) {
             ++sshift;
             ssize <<= 1;
         }
         //segmentShift和segmentMask这两个变量在定位segment时会用到,后面会详细讲
         this.segmentShift = 32 - sshift;
         this.segmentMask = ssize - 1;
         if (initialCapacity > MAXIMUM_CAPACITY)
             initialCapacity = MAXIMUM_CAPACITY;
         //计算cap的大小,即Segment中HashEntry的数组长度,cap也一定为2的n次方.
         int c = initialCapacity / ssize;
         if (c * ssize < initialCapacity)
             ++c;
         int cap = MIN_SEGMENT_TABLE_CAPACITY;
         while (cap < c)
             cap <<= 1;
         //创建segments数组并初始化第一个Segment,其余的Segment延迟初始化
         Segment<K,V> s0 =
             new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
                              (HashEntry<K,V>[])new HashEntry[cap]);
         Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
         UNSAFE.putOrderedObject(ss, SBASE, s0);
         this.segments = ss;
     }

2.2、put方法

通过源码可以看到,在进入put方法后,会先判断key和value是否为空,ConcurrentHashMap是不允许key/value为空的;下一步就是定位segment数组的下标位置,通过hash按位与算法得出,并确保segments下标位置的HashEnter数组已初始化;

public V put(K key, V value) {
        Segment<K,V> s;
        //concurrentHashMap不允许key/value为空
        if (value == null)
            throw new NullPointerException();
        //hash函数对key的hashCode重新散列,避免差劲的不合理的hashcode,保证散列均匀
        int hash = hash(key);
        //返回的hash值无符号右移segmentShift位与段掩码进行位运算,定位segment
        int j = (hash >>> segmentShift) & segmentMask;
        if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
             (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
            s = ensureSegment(j);
        return s.put(key, hash, value, false);
    }

关于segmentShift和segmentMask

  segmentShift和segmentMask这两个全局变量的主要作用是用来定位Segment,int j =(hash >>> segmentShift) & segmentMask。

  segmentMask:段掩码,假如segments数组长度为16,则段掩码为16-1=15;segments长度为32,段掩码为32-1=31。这样得到的所有bit位都为1,可以更好地保证散列的均匀性

  segmentShift:2的sshift次方等于ssize,segmentShift=32-sshift。若segments长度为16,segmentShift=32-4=28;若segments长度为32,segmentShift=32-5=27。而计算得出的hash值最大为32位,无符号右移segmentShift,则意味着只保留高几位(其余位是没用的),然后与段掩码segmentMask位运算来定位Segment。

因为Segments是个数组,里面每个元素里面又套了一个数组,所以每次put一个元素,需要计算2次hash;第一次计算Segment[] 数组的下标,第二次计算HashEnter数组的下标,

static final class HashEntry<K,V> {
        final int hash;
        final K key;
        volatile V value;
        volatile HashEntry<K,V> next;
        //其他省略
}

2.3、get方法

get方法无需加锁,由于其中涉及到的共享变量都使用volatile修饰,volatile可以保证内存可见性,且防止了指令重排序,所以不会读取到过期数据。

2.4、size方法

size操作是先统计2次,如果2次的结果不一样,就代表着有线程正在修改数据,然后对put、remove、clean进行加锁后,在统计一次,之后unlock。所以最多会统计3次,

3、JDK1.8中的ConcurrentHashMap

    在JDK1.8中,对ConcurrentHashMap的结构做了一些改进,其中最大的区别就是jdk1.8抛弃了Segments数组,摒弃了分段锁的方案,而是改用了和HashMap一样的结构操作,也就是数组 + 链表 + 红黑树结构,比jdk1.7中的ConcurrentHashMap提高了效率,在并发方面,使用了cas + synchronized的方式保证数据的一致性;因为去掉了分段锁,所以在高并发时锁住的就是数组的节点了,使得结构更加简单了;

3.1 链表转红黑树条件

    要知道,链表在遍历的时候一定是从头遍历到尾的,如果很不巧,get方法中我们要找的元素恰好在尾部,那每次获取元素的时候都得遍历一次链表,所以为了避免链表过长的情况发生,在jdk1.8中,在map的结构达到一定条件之后,将会把链表自动转为红黑树的结构,这2个条件分别是:

  1. 数组中任意个链表的长度超过8个
  2. 数组长度大于64个时

 转为红黑树后的结构如下图

3.2、put方法

 jdk8中的ConcurrentHashMap 的结构看起来虽然简单了,但是源代码却不那么容易读懂,我们先来看看put方法都做了哪些逻辑

通过上图可以看到,首次添加元素和二次添加元素所做的事情也不相同,第一次添加元素时会先初始化node数组,然后再初始化链表的头部,如果该map正在扩容,则会协助其他线程扩容,最后才是将元素插入链表中,需要注意的是,在统计元素数量时会判断是否需要扩容,是下一节就是我们要讲到的并发扩容了;

3.3、并发扩容

jdk8中的ConcurrentHashMap最复杂的就是扩容机制了,因为它不是一个个地扩容,它可以并发扩容,也就是同时进行多个节点的扩容,在默认情况下,每个cpu可以负责16个元素的长度进行扩容,比如node数组的长度为32,那么线程A负责0-16下标的数组扩容, 线程B负责17-31下标的扩容,并发扩容在transfer方法中进行,这样,2个线程分别负责高16位和低16位的扩容,不管怎样都不会产生冲突,提升了效率;

3.4、计算数组索引公式

     当我们put一个元素之后,把这个元素放到数组的哪个元素下是需要通过计算得出的,在此之前,先通过 spread() 方法计算出hash值,

计算Hash值公式: key的hashcode的低十六位 异或 高十六位,然后与Hash_bits相与(与hashmap唯一不同)

    // 用16进制表示,转为数字后数值为:2147483647   
    static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
   
    //计算hash值
    static final int spread(int h) {
        return (h ^ (h >>> 16)) & HASH_BITS;
    }

计算hash值后,在和数组的长度n进行与运算,因为下标是从0开始的,所以需要 -  1,

计算数组索引公式:hash & (n-1)    // n表示数组的长度

接下来我们手动计算一下,比如我调用了put方法

  ConcurrentHashMap<Object, Object> map = new ConcurrentHashMap<>();
  map.put("1",1);

其中 key为字符串"1",得出hashCode 为 49,得出总公式代码如下

        // key 值的hashCode ,通过key.hashCode() 方法获取
        int h = 49;
        
        // 数组长度,默认为16
        int n =  16;
        
        // hash字节,在源码中用0x7fffffff表示,在这里转为数字,方便调试
        int HASH_BITS = 2147483647;
        
        // 计算hash值
        int hash = (h ^ (h >>> 16)) & HASH_BITS;
        
        // 计算数组下标
        int arrayIndex =  hash & (n - 1);
        System.out.println(arrayIndex);

运行后,结果为1,通过以上结果,数组的下标值就计算出来了;

     了解代码的实现原理有什么用呢?其实我们在实现业务场景的时候,有很多的算法和机制是可以复制的,我们在编码的时候大部分情况下都要去看别人写的代码,如果有注释还好,没有注释的话就只能去猜作者为什么要这样写,然后小心验证,所以熟悉源码的作用也是如此,只有看多了别人写的代码,才有经验去读懂别人写的代码,让我们自己实现起来的话也可以起到一个借鉴的作用,就像设计模式那样,通过复用的方式减少了冗余的代码,也提高了可读性;

以上是关于图解ConcurrentHashMap的前世今生的主要内容,如果未能解决你的问题,请参考以下文章

husky的配置及前世今生

云计算的前世今生

Lambda的前世今生

原创 | 函数 scanf 的前世今生

GIT前世今生

.Net Core001.NetCore前世今生