ConcurrentHashMap 的工作原理及源码分析,如何统计所有的元素个数

  • Post author:
  • Post category:其他


1.ConcurrentHashMap(线程安全):

ConcurrentHashMap采用了非常精妙的”分段锁”策略,ConcurrentHashMap的主干是个Segment数组。

 final Segment<K,V>[] segments;

Segment继承了ReentrantLock(参照:

https://blog.csdn.net/striveb/article/details/83421107

),所以它就是一种可重入锁(ReentrantLock)。在ConcurrentHashMap,一个Segment就是一个子哈希表,Segment里维护了一个HashEntry数组,并发环境下,对于不同Segment的数据进行操作是不用考虑锁竞争的。(就按默认的ConcurrentLeve为16来讲,理论上就允许16个线程并发执行 )。

所以,对于同一个Segment的操作才需考虑线程同步,不同的Segment则无需考虑。

Segment类似于HashMap,一个Segment维护着一个HashEntry数组:

transient volatile HashEntry<K,V>[] table;

HashEntry是目前我们提到的最小的逻辑处理单元了。

一个ConcurrentHashMap维护一个Segment数组,一个Segment维护一个HashEntry数组。

static final class HashEntry<K,V> {
        final int hash;
        final K key;
        volatile V value;
        volatile HashEntry<K,V> next;

        HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
            this.hash = hash;
            this.key = key;
            this.value = value;
            this.next = next;
        }
    }

Segment构造方法:

Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
            this.loadFactor = lf;//负载因子
            this.threshold = threshold;//阈值
            this.table = tab;//主干数组即HashEntry数组
        }

Segment类似哈希表,那么一些属性就跟我们之前提到的HashMap差不离,比如负载因子loadFactor,比如阈值threshold等等。

  • 存储结构: ConcurrentHashMap 和 HashMap 实现上类似,最主要的差别是 ConcurrentHashMap 采用了分段锁(Segment),每个分段锁维护着几个桶(HashEntry),多个线程可以同时访问不同分段锁上的桶,从而使其并发度更高(并发度就是 Segment 的个数)。
  • 数据结构(JDK1.7):

如图所示,是由

Segment

数组、

HashEntry

数组组成,和

HashMap

一样,仍然是数组加链表组成。

ConcurrentHashMap

采用了分段锁技术,其中

Segment

继承于

ReentrantLock

。不会像

HashTable

那样不管是

put

还是

get

操作都需要做同步处理,理论上 ConcurrentHashMap 支持

CurrencyLevel

(Segment 数组数量)的线程并发。每当一个线程占用锁访问一个

Segment

时,不会影响到其他的

Segment

ConcurrentHashMap构造方法:

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

初始化方法有三个参数,如果用户不指定则会使用默认值,initialCapacity为16,loadFactor为0.75(负载因子,扩容时需要参考),concurrentLevel为16。


从上面的代码可以看出来,Segment数组的大小ssize是由concurrentLevel来决定的,但是却不一定等于concurrentLevel,ssize一定是大于或等于concurrentLevel的最小的2的次幂。比如:默认情况下concurrentLevel是16,则ssize为16;若concurrentLevel为14,ssize为16;若concurrentLevel为17,则ssize为32。为什么Segment的数组大小一定是2的次幂?其实主要是便于通过按位与的散列算法来定位Segment的index。

  • 默认的并发级别为 16,也就是说默认创建 16 个 Segment。
  • get 方法:ConcurrentHashMap的 Segment的get操作实现非常简单和高效。,因为整个过程都不需要加锁。

    public V get(Object key) {
            Segment<K,V> s; 
            HashEntry<K,V>[] tab;
            int h = hash(key);
            long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
            //先定位Segment,再定位HashEntry
            if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
                (tab = s.table) != null) {
                for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                         (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
                     e != null; e = e.next) {
                    K k;
                    if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                        return e.value;
                }
            }
            return null;
        }

    只需要将

    Key

    通过

    Hash

    之后定位到具体的

    Segment

    ,再通过一次

    Hash

    定位到具体的元素上。

    由于

    HashEntry

    中的

    value

    属性是用

    volatile

    关键词修饰的,保证了内存可见性,所以每次获取时都是最新值,但不能保证多个线程写。

    根据Java内存模型的happen before原则,对volatile字段的写入操作先于读操作,即使两个线程同时修改和获取volatile变量,get操作也能拿到最新的值,这是用volatile替换锁的经典应用场景。

  • put 方法:

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

    虽然 HashEntry 中的 value 是用

    volatile

    关键词修饰的,但是并不能保证并发的原子性,所以 put 操作时仍然需要加锁处理。首先也是通过 Key 的 Hash 定位到具体的 Segment,在 put 之前会进行一次扩容校验。这里比 HashMap 要好的一点是:HashMap 是插入元素之后再看是否需要扩容,有可能扩容之后后续就没有插入就浪费了本次扩容(扩容非常消耗性能)。而 ConcurrentHashMap 不一样,它是在将数据插入之前检查是否需要扩容,之后再做插入操作。在扩容的时,首先会创建一个容量是原来容量两倍的数组,然后将原数组里的元素进行再散列后插入到新的数组里。为了高效,ConcurrentHashMap不会对整个容器进行扩容,而只对某个segment进行扩容。主要就分为两步:

    1.定位segment并确保定位的Segment已初始化 2.调用Segment的put方法。

  • size 方法:每个 Segment 维护了一个 count 变量来统计该 Segment 中的键值对个数。

    /**
     * The number of elements. Accessed only either within locks
     * or among other volatile reads that maintain visibility.
     */
    transient int count;

    如果要统计整个ConcurrentHashMap里元素的大小,就必须统计所有Segment里元素的大小后求和。Segment里的全局变量count是一个volatile变量,在多线程场景下,是不是直接把所有Segment的count相加就可以得到整个ConcurrentHashMap大小了呢?不是的,虽然相加时可以获取每个Segment的count的最新值,但是可能累加前使用的count发生了变化,那么统计结果就不准了。所以,最安全的做法是在统计size的时候把所有Segment的put、remove和clean方法全部锁住,但是这种做法显然非常低效。

    因为在累加count操作过程中,之前累加过的count发生变化的几率非常小,所以ConcurrentHashMap的做法是先尝试2次通过不锁住Segment的方式来统计各个Segment大小,如果统计的过程中,容器的count发生了变化,则再采用加锁的方式来统计所有Segment的大小。尝试次数使用 RETRIES_BEFORE_LOCK 定义,该值为 2,retries 初始值为 -1,因此尝试次数为 3。如果尝试的次数超过 3 次,就需要对每个 Segment 加锁。

    /**
     * Number of unsynchronized retries in size and containsValue
     * methods before resorting to locking. This is used to avoid
     * unbounded retries if tables undergo continuous modification
     * which would make it impossible to obtain an accurate result.
     */
    static final int RETRIES_BEFORE_LOCK = 2;
    
    public int size() {
        // Try a few times to get accurate count. On failure due to
        // continuous async changes in table, resort to locking.
        final Segment<K,V>[] segments = this.segments;
        int size;
        boolean overflow; // true if size overflows 32 bits
        long sum;         // sum of modCounts
        long last = 0L;   // previous sum
        int retries = -1; // first iteration isn't retry
        try {
            for (;;) {
                // 超过尝试次数,则对每个 Segment 加锁
                if (retries++ == RETRIES_BEFORE_LOCK) {
                    for (int j = 0; j < segments.length; ++j)
                        ensureSegment(j).lock(); // force creation
                }
                sum = 0L;
                size = 0;
                overflow = false;
                for (int j = 0; j < segments.length; ++j) {
                    Segment<K,V> seg = segmentAt(segments, j);
                    if (seg != null) {
                        sum += seg.modCount;
                        int c = seg.count;
                        if (c < 0 || (size += c) < 0)
                            overflow = true;
                    }
                }
                // 连续两次得到的结果一致,则认为这个结果是正确的
                if (sum == last)
                    break;
                last = sum;
            }
        } finally {
            if (retries > RETRIES_BEFORE_LOCK) {
                for (int j = 0; j < segments.length; ++j)
                    segmentAt(segments, j).unlock();
            }
        }
        return overflow ? Integer.MAX_VALUE : size;
    }

    每个

    Segment

    都有一个

    modCount

    变量,每当进行一次

    put、remove和clean

    等操作,

    modCount

    将会 +1。只要

    modCount

    发生了变化就认为容器的大小也在发生变化。

  • JDK1.8的实现:

抛弃了原有的 Segment 分段锁,而采用了

CAS + synchronized

来保证并发安全性。

也将 1.7 中存放数据的 HashEntry 改为 Node,但作用都是相同的。其中的

val next

都用了 volatile 修饰,保证了可见性。

put方法:

  • 根据 key 计算出 hashcode 。
  • 判断是否需要进行初始化。

  • f

    即为当前 key 定位出的 Node,如果为空表示当前位置可以写入数据,利用 CAS 尝试写入,失败则自旋保证成功。
  • 如果当前位置的

    hashcode == MOVED == -1

    ,则需要进行扩容。
  • 如果都不满足,则利用 synchronized 锁写入数据。
  • 如果数量大于

    TREEIFY_THRESHOLD

    则要转换为红黑树。

get方法:

  • 根据计算出来的 hashcode 寻址,如果就在桶上那么直接返回值。
  • 如果是红黑树那就按照树的方式获取值。
  • 都不满足那就按照链表的方式遍历获取值。

总结

1.8 在 1.7 的数据结构上做了大的改动,采用红黑树之后可以保证查询效率(

O(logn)

),甚至取消了 ReentrantLock 改为了 synchronized,这样可以看出在新版的 JDK 中对 synchronized 优化是很到位的。

推荐几篇不错的文章:


https://blog.csdn.net/fjse51/article/details/55260493


https://crossoverjie.top/2018/07/23/java-senior/ConcurrentHashMap/


http://www.cnblogs.com/chengxiao/p/6842045.html

觉得有收获的,可以来一波赞赏!




版权声明:本文为striveb原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。