线程安全的并发容器1.7版本的ConcurrentHashMap原理分析（二十二）

秒速五厘米 2022-12-30 15:51 40阅读 0赞

线程安全的并发容器1.7版本的ConcurrentHashMap原理分析：

**ConcurrentHashMap**  **使用**

除了  Map  系列应该有的线程安全的  get ， put  等方法外， ConcurrentHashMap 还提供了一个在并发下比较有用的方法 putIfAbsent ，如果传入  key  对应的  value 已经存在，就返回存在的 value ，不进行替换。如果不存在，就添加  key  和  value ， 返回 null 。在代码层面它的作用类似于：

synchronized(map){
    if (map.get(key) == null){
        return map.put(key, value);
    } else{
        return map.get(key);
        }
    }

源码如下：

public V putIfAbsent(K key, V value) {
            Segment<K,V> s;
            if (value == null)
                throw new NullPointerException();
            int hash = hash(key);
            int j = (hash >>> segmentShift) & segmentMask;
            if ((s = (Segment<K,V>)UNSAFE.getObject
                 (segments, (j << SSHIFT) + SBASE)) == null)
                s = ensureSegment(j);
            return s.put(key, hash, value, true);
        }

它让上述代码的整个操作是线程安全的。

***在***  ***1.7***  ***下的实现***

***1，***

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

桶：

static final class Segment<K,V> extends ReentrantLock implements Serializable {

ConcurrentHashMap  是由  Segment  数组结构和  HashEntry  数组结构组成。 Segment 是一种可重入锁（ ReentrantLock ），在  ConcurrentHashMap  里扮演锁的 角色；HashEntry  则用于存储键值对数据。一个  ConcurrentHashMap  里包含一个 Segment 数组。 Segment  的结构和  HashMap  类似，是一种数组和链表结构。一个 Segment 里包含一个  HashEntry  数组，每个  HashEntry  是一个链表结构的元素， 每个 Segment  守护着一个  HashEntry  数组里的元素，当对  HashEntry  数组的数据 进行修改时，必须首先获得与它对应的 Segment  锁。

2、 构造方法和初始化

public ConcurrentHashMap17(int initialCapacity,
                                 float loadFactor, int concurrencyLevel) {
            if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
                throw new IllegalArgumentException();
            if (concurrencyLevel > MAX_SEGMENTS)
                concurrencyLevel = MAX_SEGMENTS;
            // Find power-of-two sizes best matching arguments
            int sshift = 0;
            int ssize = 1;
            while (ssize < concurrencyLevel) {
                ++sshift;
                ssize <<= 1;
            }
            this.segmentShift = 32 - sshift;
            this.segmentMask = ssize - 1;
            if (initialCapacity > MAXIMUM_CAPACITY)
                initialCapacity = MAXIMUM_CAPACITY;
            int c = initialCapacity / ssize;
            if (c * ssize < initialCapacity)
                ++c;
            int cap = MIN_SEGMENT_TABLE_CAPACITY;
            while (cap < c)
                cap <<= 1;
            // create segments and segments[0]
            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); // ordered write of segments[0]
            this.segments = ss;
        }

ConcurrentHashMap  初始化方法是通过  initialCapacity 、 loadFactor  和 concurrencyLevel(参数  concurrencyLevel  是用户估计的并发级别，就是说你觉得最 多有多少线程共同修改这个 map ，根据这个来确定  Segment  数组的大小 concurrencyLevel 默认是  DEFAULT\_CONCURRENCY\_LEVEL = 16;) 等几个参数来初始 化 segment  数组、段偏移量  segmentShift 、段掩码  segmentMask  和每个  segment 里的 HashEntry  数组来实现的。并发级别可以理解为程序运行时能够同时更新 ConccurentHashMap 且不产 生锁竞争的最大线程数，实际上就是 ConcurrentHashMap  中的分段锁个数，即 Segment\[\]的数组长度。 ConcurrentHashMap  默认的并发度为  16 ，但用户也可以 在构造函数中设置并发度。当用户设置并发度时，ConcurrentHashMap  会使用大 于等于该值的最小 2  幂指数作为实际并发度（假如用户设置并发度为  17 ，实际 并发度则为 32 ）。 如果并发度设置的过小，会带来严重的锁竞争问题；如果并发度设置的过大， 原本位于同一个 Segment  内的访问会扩散到不同的  Segment  中， CPU cache  命中 率会下降，从而引起程序性能下降。（文档的说法是根据你并发的线程数量决定， 太多会导性能降低） segments 数组的长度  ssize  是通过  concurrencyLevel  计算得出的。为了能通 过按位与的散列算法来定位 segments  数组的索引，必须保证  segments  数组的长 度是 2  的  N  次方（ power-of-two size ），所以必须计算出一个大于或等于 concurrencyLevel 的最小的  2  的  N  次方值来作为  segments  数组的长度。假如 concurrencyLevel 等于  14 、 15  或  16 ， ssize  都会等于  16 ，即容器里锁的个数也是 16。

以下知识了解即可：

*初始化*  *segmentShift*  *和*  *segmentMask* *这两个全局变量需要在定位 segment*  *时的散列算法里使用，* *sshift*  *等于*  *ssize* *从*  *1*  *向左移位的次数，在默认情况下*  *concurrencyLevel*  *等于*  *16* *，* *1*  *需要向左移位*  *移动*  *4*  *次，所以*  *sshift*  *等于*  *4* *。* *segmentShift*  *用于定位参与散列运算的位数，*  *segmentShift*  *等于*  *32*  *减*  *sshift* *，所以等于*  *28* *，这里之所以用*  *32*  *是因为*  *ConcurrentHashMap*  *里的*  *hash()* *方法输出的最大数是*  *32*  *位的。* *segmentMask*  *是散*  *列运算的掩码，等于*  *ssize*  *减*  *1* *，即*  *15* *，掩码的二进制各个位的值都是*  *1* *。因为*

*ssize*  *的最大长度是*  *65536* *，所以*  *segmentShift*  *最大值是*  *16* *，* *segmentMask*  *最大值*  *是*  *65535* *，对应的二进制是*  *16*  *位，每个位都是*  *1* *。*

输入参数 initialCapacity 是 ConcurrentHashMap 的初始化容量，loadfactor 是 每个 segment  的负载因子，在构造方法里需要通过这两个参数来初始化数组中的 每个 segment 。上面代码中的变量  cap  就是  segment  里  HashEntry  数组的长度， 它等于 initialCapacity  除以  ssize  的倍数  c ，如果  c  大于  1 ，就会取大于等于  c  的  2 的 N  次方值，所以  segment  里  HashEntry  数组的长度不是  1 ，就是  2  的  N  次方。 在默认情况下， ssize = 16 ， initialCapacity = 16 ， loadFactor = 0.75f ，那么  cap = 1， threshold = (int) cap \* loadFactor = 0 。既然 ConcurrentHashMap 使用分段锁 Segment 来保护不同段的数据，那么在 插入和获取元素的时候，必须先通过散列算法定位到 Segment 。 ConcurrentHashMap 会首先使用  Wang/Jenkins hash  的变种算法对元素的 hashCode 进行一次再散列。 ConcurrentHashMap 完全允许多个读操作并发进行，读操作并不需要加锁。 ConcurrentHashMap 实现技术是保证  HashEntry  几乎是不可变的以及  volatile  关键 字。

3、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;
            }
    
            /**
             * Sets next field with volatile write semantics.  (See above
             * about use of putOrderedObject.)
             */
            final void setNext(HashEntry<K,V> n) {
                UNSAFE.putOrderedObject(this, nextOffset, n);
            }
    
            // Unsafe mechanics
            static final sun.misc.Unsafe UNSAFE;
            static final long nextOffset;
            static {
                try {
                    UNSAFE = sun.misc.Unsafe.getUnsafe();
                    Class k = HashEntry.class;
                    nextOffset = UNSAFE.objectFieldOffset
                        (k.getDeclaredField("next"));
                } catch (Exception e) {
                    throw new Error(e);
                }
            }
        }

4、 get  操作

public V get(Object key) {
            Segment<K,V> s; // manually integrate access methods to reduce overhead
            HashEntry<K,V>[] tab;
            int h = hash(key);//准备定位的hash值
            long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
            if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
                (tab = s.table) != null) {//拿到segment下的table
                for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                         (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
                     e != null; e = e.next) {//遍历table下指定的HashEntry列表
                    K k;
                    if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                        return e.value;
                }
            }
            return null;
        }

get  操作先经过一次再散列，然后使用这个散列值通过散列运算定位到 Segment(使用了散列值的高位部分 ) ，再通过散列算法定位到  table( 使用了散列值 的全部) 。整个  get  过程，没有加锁，而是通过  volatile  保证  get  总是可以拿到最 新值。

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

5、 put  操作

public V put(K key, V value) {
            Segment<K,V> s;
            if (value == null)
                throw new NullPointerException();
            int hash = hash(key);//定位所需的hash值
            int j = (hash >>> segmentShift) & segmentMask;
            if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
                 (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
                s = ensureSegment(j);//初始化segment[j],因为整个map初始化时，只初始化了segment[0]
            return s.put(key, hash, value, false);
        }

ConcurrentHashMap  初始化的时候会初始化第一个槽  segment\[0\] ，对于其他 槽，在插入第一个值的时候再进行初始化。 ensureSegment 方法考虑了并发情况，多个线程同时进入初始化同一个槽 segment\[k\]，但只要有一个成功就可以了。 具体实现是

private Segment<K,V> ensureSegment(int k) {
            final Segment<K,V>[] ss = this.segments;
            long u = (k << SSHIFT) + SBASE; // raw offset
            Segment<K,V> seg;
            if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
                Segment<K,V> proto = ss[0]; // use segment 0 as prototype
                int cap = proto.table.length;
                float lf = proto.loadFactor;
                int threshold = (int)(cap * lf);
                HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
                if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                    == null) { // recheck
                    Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
                    //多次检查，循环CAS操作，保证多线程下只有一个线程可以成功
                    while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                           == null) {
                        if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                            break;
                    }
                }
            }
            return seg;
        }

最终调用s.put(key, hash, value, false);方法：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
                HashEntry<K,V> node = tryLock() ? null :
                    scanAndLockForPut(key, hash, value);
                V oldValue;
                try {
                    HashEntry<K,V>[] tab = table;
                    int index = (tab.length - 1) & hash;
                    HashEntry<K,V> first = entryAt(tab, index);
                    for (HashEntry<K,V> e = first;;) {
                        if (e != null) {
                            K k;
                            if ((k = e.key) == key ||
                                (e.hash == hash && key.equals(k))) {
                                oldValue = e.value;
                                if (!onlyIfAbsent) {
                                    e.value = value;
                                    ++modCount;
                                }
                                break;
                            }
                            e = e.next;
                        }
                        else {
                            if (node != null)
                                node.setNext(first);
                            else
                                node = new HashEntry<K,V>(hash, key, value, first);
                            int c = count + 1;
                            if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                                rehash(node);
                            else
                                setEntryAt(tab, index, node);
                            ++modCount;
                            count = c;
                            oldValue = null;
                            break;
                        }
                    }
                } finally {
                    unlock();
                }
                return oldValue;
            }

put  方法会通过  tryLock() 方法尝试获得锁，获得了锁， node  为  null  进入  try 语句块，没有获得锁，调用 scanAndLockForPut  方法自旋等待获得锁。

6、调用  scanAndLockForPut(key, hash, value)方法：

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
                HashEntry<K,V> first = entryForHash(this, hash);
                HashEntry<K,V> e = first;
                HashEntry<K,V> node = null;
                int retries = -1; // negative while locating node
                while (!tryLock()) {
                    HashEntry<K,V> f; // to recheck first below
                    if (retries < 0) {
                        if (e == null) {
                            if (node == null) // speculatively create node
                                node = new HashEntry<K,V>(hash, key, value, null);
                            retries = 0;
                        }
                        else if (key.equals(e.key))
                            retries = 0;
                        else
                            e = e.next;
                    }
                    else if (++retries > MAX_SCAN_RETRIES) {
                        lock();
                        break;
                    }
                    else if ((retries & 1) == 0 &&
                             (f = entryForHash(this, hash)) != first) {
                        e = first = f; // re-traverse if entry changed
                        retries = -1;
                    }
                }
                return node;
            }

scanAndLockForPut  方法里在尝试获得锁的过程中会对对应  hashcode  的链表 进行遍历，如果遍历完毕仍然找不到与 key  相同的  HashEntry  节点，则为后续的 put 操作提前创建一个  HashEntry 。当  tryLock  一定次数后仍无法获得锁，则通过 lock 申请锁。在获得锁之后，Segment 对链表进行遍历，如果某个 HashEntry 节点具有相 同的 key ，则更新该  HashEntry  的  value  值，否则新建一个 HashEntry 节点，采用头插法，将它设置为链表的新 head 节 点并将原头节点设为新 head  的下一个节点。新建过程中如果节点总数（含新建 的 HashEntry ）超过  threshold ，则调用  rehash() 方法对  Segment  进行扩容，最后 将新建 HashEntry  写入到数组中。

7、 rehash  操作

put方法里 rehash(node);

private void rehash(HashEntry<K,V> node) {
              
                HashEntry<K,V>[] oldTable = table;
                int oldCapacity = oldTable.length;
                int newCapacity = oldCapacity << 1;/* oldCapacity*2 */
                threshold = (int)(newCapacity * loadFactor);
                HashEntry<K,V>[] newTable =
                    (HashEntry<K,V>[]) new HashEntry[newCapacity];
                int sizeMask = newCapacity - 1;
                for (int i = 0; i < oldCapacity ; i++) {
                    HashEntry<K,V> e = oldTable[i];
                    if (e != null) {
                        HashEntry<K,V> next = e.next;
                        int idx = e.hash & sizeMask;
                        if (next == null)   //  Single node on list
                            newTable[idx] = e;
                        else { // Reuse consecutive sequence at same slot
                            HashEntry<K,V> lastRun = e;
                            int lastIdx = idx;
                            for (HashEntry<K,V> last = next;
                                 last != null;
                                 last = last.next) {
                                int k = last.hash & sizeMask;
                                if (k != lastIdx) {
                                    lastIdx = k;
                                    lastRun = last;
                                }
                            }
                            newTable[lastIdx] = lastRun;
                            // Clone remaining nodes
                            for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
                                V v = p.value;
                                int h = p.hash;
                                int k = h & sizeMask;
                                HashEntry<K,V> n = newTable[k];
                                newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
                            }
                        }
                    }
                }
                int nodeIndex = node.hash & sizeMask; // add the new node
                node.setNext(newTable[nodeIndex]);
                newTable[nodeIndex] = node;
                table = newTable;
            }

扩容是新创建了数组，然后进行迁移数据，最后再将 newTable 设置给属性table。为了避免让所有的节点都进行复制操作：由于扩容是基于 2 的幂指来操作， 假设扩容前某 HashEntry 对应到 Segment 中数组的 index 为 i，数组的容量为 capacity，那么扩容后该 HashEntry 对应到新数组中的 index 只可能为 i 或者 i+capacity，因此很多HashEntry 节点在扩容前后 index 可以保持不变。

假设原来  table  长度为  4 ，那么元素在  table  中的分布是这样的

![20201227210800583.png][]

扩容后  table  长度变为  8 ，那么元素在  table  中的分布变成：

![20201227210823783.png][]

可以看见  hash  值为  34  和  56  的下标保持不变，而  15,23,77  的下标都是在原 来下标的基础上+4  即可，可以快速定位和减少重排次数。 该方法没有考虑并发，因为执行该方法之前已经获取了锁。

8、 remove  操作 :key做参数

public V remove(Object key) {
            int hash = hash(key);
            Segment<K,V> s = segmentForHash(hash);
            return s == null ? null : s.remove(key, hash, null);
        }

或：用key和value做参数

public boolean remove(Object key, Object value) {
            int hash = hash(key);
            Segment<K,V> s;
            return value != null && (s = segmentForHash(hash)) != null &&
                s.remove(key, hash, value) != null;
        }

与 put 方法类似，都是在操作前需要拿到锁，以保证操作的线程安全性。

9、 ConcurrentHashMap  的弱一致性

然后对链表遍历判断是否存在  key  相同的节点以及获得该节点的  value 。但 由于遍历过程中其他线程可能对链表结构做了调整，因此 get  和  containsKey  返 回的可能是过时的数据，这一点是 ConcurrentHashMap  在弱一致性上的体现。如 果要求强一致性，那么必须使用 Collections.synchronizedMap() 方法。

size 、 containsValue

这些方法都是基于整个  ConcurrentHashMap  来进行操作的，他们的原理也基 本类似：首先不加锁循环执行以下操作：循环所有的 Segment ，获得对应的值以 及所有 Segment  的  modcount  之和。在  put 、 remove  和  clean  方法里操作元素 前都会将变量 modCount  进行变动，如果连续两次所有  Segment  的  modcount 和相等，则过程中没有发生其他线程修改 ConcurrentHashMap  的情况，返回获得 的值。当循环次数超过预定义的值时，这时需要对所有的 Segment  依次进行加锁， 获取返回值后再依次解锁。所以一般来说，应该避免在多线程环境下使用 size 和 containsValue  方法。

今天主要分享1.7版本的 ConcurrentHashMap，下篇我们分享1.8版本的原理，敬请期待！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221120/250cded579064d388872f58a6cf3a00d.png
[20201227210800583.png]: /images/20221120/9ec90494dcd8409a88377a72d71335d4.png
[20201227210823783.png]: https://img-blog.csdnimg.cn/20201227210823783.png