数据结构算法 - ConcurrentHashMap 源码解析

Love The Way You Lie 2022-02-19 01:11 247阅读 0赞

五个线程同时往 HashMap 中 put 数据会发生什么？  
ConcurrentHashMap 是怎么保证线程安全的？

在分析 HashMap 源码时还遗留这两个问题，这次我们站在 Java 多线程内存模型和 synchronized 的实现原理，这两个角度来彻底分析一下。至于 JDK 1.8 的红黑树不是本文探讨的内容。

## 1. Java 多线程内存模型 ##

五个线程同时往 HashMap 中 put 数据会出现两种现象，大概率会出现数据丢失，小概率会出现死循环，我们不妨写个测试代码自己验证一下。那为什么会出现这两种现象，我们先来回顾一下之前的Java 多线程内存模型。请看图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70]

Java内存模型中规定了所有的变量都存储在主内存中，每条线程还有自己的工作内存，线程的工作内存中保存了该线程使用到的变量到主内存副本拷贝，线程对变量的所有操作（读取、赋值）都必须在工作内存中进行，而不能直接读写主内存中的变量。不同线程之间无法直接访问对方工作内存中的变量，线程间变量值的传递均需要在主内存来完成，线程、主内存和工作内存的交互关系如上图所示。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 1]  
1、Segment默认是16，理论上，最多同时支持16个线程并发读写，但是是操作不同的Segment  
2、初始化时可以指定Segment数量  
3、Segment继承自ReentrantLock,每一个Segment都会有一把锁，保证线程安全  
4、采用链表方式解决hash冲突

**原理上来说：**  
和 HashMap 非常类似，唯一的区别就是其中的核心数据如 value ，以及链表都是 volatile 修饰的，保证了获取时的可见性。

ConcurrentHashMap 采用了分段锁技术，其中 Segment 继承于 ReentrantLock。不会像 HashTable 那样不管是 put 还是 get 操作都需要做同步处理，理论上 ConcurrentHashMap 支持 CurrencyLevel (Segment 数组数量)的线程并发。每当一个线程占用锁访问一个 Segment 时，不会影响到其他的 Segment。  
**思考？**  
超过16个线程是否就会出现线程不安全的情况？不会，只是会出现线程的等待竞争的情况。

**采用链表解决hash冲突的缺点：那就是查询遍历链表效率太低。**

**源码分析：**  
1、首先是通过 key 定位到 Segment，之后在对应的 Segment 中进行具体的 put。

public V put(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          // nonvolatile; recheck
                 (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
                s = ensureSegment(j);
            return s.put(key, hash, value, false);
        }

2、在Segment中通过**scanAndLockForPut**加锁后，进行添加操作，然后解锁

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

将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry。  
遍历该 HashEntry，如果不为空则判断传入的 key 和当前遍历的 key 是否相等，相等则覆盖旧的 value。  
不为空则需要新建一个 HashEntry 并加入到 Segment 中，同时会先判断是否需要扩容。  
最后会解除在 1 中所获取当前 Segment 的锁。

**分析Segment加锁代码：**  
虽然 HashEntry 中的 value 是用 volatile 关键词修饰的，但是并不能保证并发的原子性，所以 put 操作时仍然需要加锁处理。  
首先第一步的时候会尝试获取锁，如果获取失败肯定就有其他线程存在竞争，则利用 scanAndLockForPut() 自旋获取锁。  
**其中 Segment 继承于 ReentrantLock**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 2]  
1、尝试自旋获取锁。  
2、如果重试的次数达到了 MAX\_SCAN\_RETRIES 则改为**阻塞锁获取**，保证能获取成功。

**jdk1.8开始， ConcurrentHashMap的结构：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 3]  
1、取消了Segment 分段锁，采用 synchronized 和 CAS 来实现线程安全  
2、当链表长度大于8时，转为红黑色结构。

**线程不安全的原因分析：**  
现在我们来想象一下，假设线程 1 把数据读到了自己的工作内存中，在 tab 角标为 1 的链表头插入了一条新的数据，倘若这时还没来得及将新增的数据刷新到主内中。接着线程 2 就把数据读到了自己的工作内存中，在 tab 角标为 1 的链表头插入了一条新的数据。接着线程 1 把新增数据刷新到主内存中，线程 2 也把数据新增数据刷新到主内存中，那么线程 2 就会覆盖线程 1 的新增数据，从而导致数据丢失的情况。这里需要注意的是，只有两个线程都是操作 tab 的同一个 index 链表才会导致数据丢失的情况，如果不是同一个 index 链表就不会有覆盖和丢失这一说。  
**简单的说，多个线程进行put操作时，由于key值的hashcode值相等，导致向同一个tab里面的链表或者树节点上的空位存入值，这样，就会出现相互覆盖的情况。**

## 2. synchronized 的底层实现原理 ##

关于 HashMap 的线程不安全问题，Java 给我们提供了三种方案：

**1. 第一种是 HashTable  
2. 第二种是 Collections.synchronizedMap()  
3. 第三种是 ConcurrentHashMap**

而第一种和第二种都是通过用 synchronized 同步方法来保证线程安全，性能上有所欠缺不推荐大家使用。**ConcurrentHashMap 在 JDK 1.8 之前采用的是 Segment 分段锁来实现的，而 JDK 1.8 之后则采用 synchronized 和 CAS 来实现。**

HashTable 通过锁住整个 put 和 get 方法来实现线程安全并不是很合理，因为一个线程在 put 的时候，另外一个线程不能再 put 和 get 必须进入等待状态。同理一个线程在 get 的时候，另外一个线程也不能再 get 和 put 。上面通过分析只有两个线程都是操作 tab 的同一个 index 链表才会导致数据丢失的情况，如果不是同一个 index 链表就不会有覆盖和丢失这一说。因此也没必要锁住整个方法，只需要锁住每个 tab 的 index 链即可。

ConcurrentHashMap 在 JDK 1.8 之前采用的是 Segment 继承自 ReentrantLock 来锁住 tab 的 index 链，而 JDK 1.8 之后则采用 synchronized 来实现，这两者又有什么区别？我们首先看下 synchronized 的底层是怎么实现线程安全的。Java中的每一个对象都可以作为锁。具体表现有以下3种形式。

// 1.对于普通同步方法，锁是当前实例对象。this
    public synchronized void method(){
    
    }
    
    // 2.对于静态同步方法，锁是当前类的Class对象。this.class
    public static synchronized void method(){
    
    }
    
    // 3.对于同步方法块，锁是Synchonized括号里配置的对象。object
    public static synchronized void method(){
      synchronized(object){
    
      }
    }

--------------------

我们可能会想锁到底存在哪里呢？锁里面会存储什么信息呢？**其实 synchronized 同步的代码块，虚拟机在同步代码块开始前会插入一条 monitorenter 指令，在代码块的末尾会插入一条 monitorexit 指令**。而每个对象的 Mark Word 头信息里都会存储 Monitor 信息，也就是当前对象的锁信息，当然 Mark Word 头信息还包含对象的 hashCode 和 GC 的分代年龄，具体请看下表：

Lock 的实现原理和 synchronized 有些类似，都是通过线程的原子性来保证线程同步，具体的实现的方式大家可以去看下 ReentrantLock 的源码实现。  
**那为什么在 JDK 1.8 之后要采用 synchronized 和 CAS 来实现？**  
在 JDK 1.6 为了减少获得锁和释放锁带来的性能消耗，引入了“偏向锁”和“轻量级锁”，**级别从低到高依次是：无锁状态、偏向锁状态、轻量级锁状态和重量级锁状态**，这几个状态会随着竞争情况逐渐升级。锁可以升级但不能降级，意味着偏向锁升级成轻量级锁后不能降级成偏向锁。这种锁升级却不能降级的策略，目的是为了提高获得锁和释放锁的效率。  
当线程 1 进入同步代码块遇到 monitorenter 指令，首先判断锁的状态发现是 0 ，采用 CAS 将锁的状态设置为 1，偏向锁设置为 1，锁的标致位设置为 1 ，继续执行同步代码块里面的指令。这时若线程 2 也来到了同步代码块，也会遇到 monitorenter 指令，首先判断锁的状态发现是 1 进入等待中，等线程 1 执行完同步代码块遇到 monitorenter 指令，首先会清空锁的状态然后唤醒线程 2 。如此反复即可保证线程安全。

**偏向锁**  
大多数情况下，锁不仅不存在多线程竞争，而且总是由同一线程多次获得，为了让线程获得锁的代价更低而引入了偏向锁。当一个线程访问同步块并获取锁时，会在对象头和栈帧中的锁记录里存储锁偏向的线程 ID，以后该线程在进入和退出同步块时不需要进行 CAS 操作来加锁和解锁，只需简单地测试一下对象头的 Mark Word 里是否存储着指向当前线程的偏向锁。如果测试成功，表示线程已经获得了锁。如果测试失败，则需要再测试一下 Mark Word 中偏向锁的标识是否设置成1（表示当前是偏向锁）：如果没有设置，则使用 CAS 竞争锁；如果设置了，则尝试使用CAS将对象头的偏向锁指向当前线程。

**轻量级锁**  
线程在执行同步块之前，JVM 会先在当前线程的栈桢中创建用于存储锁记录的空间，并将对象头中的 Mark Word 复制到锁记录中。然后线程尝试使用 CAS 将对象头中的 Mark Word 替换为指向锁记录的针。如果成功，当前线程获得锁，如果失败，表示其他线程竞争锁，当前线程便尝试使用自旋来获取锁。

**重量级锁**  
轻量级锁采用自旋的方式不断的尝试获取锁，如果长时间获取不到锁势必会不断消耗 CPU 的资源。所以当线程竞争比较激烈或者线程迟迟获取不到锁，就会升级为重量级的锁状态，此时线程是阻塞的，且响应时间缓慢。

## 3. ConcurrentHashMap 源码分析 ##

其中抛弃了原有的 Segment 分段锁，而采用了 CAS + synchronized 来保证并发安全性。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 5]  
也将 1.7 中存放数据的 HashEntry 改为 Node，但作用都是相同的。

其中的 val next 都用了 volatile 修饰，保证了可见性。

put 方法  
重点来看看 put 函数：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 6]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 7]

根据 key 计算出 hashcode 。  
判断是否需要进行初始化。  
f 即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。  
如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。  
如果都不满足，则利用 synchronized 锁写入数据。  
如果数量大于 TREEIFY\_THRESHOLD 则要转换为红黑树。

get 方法：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 8]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 9]  
根据计算出来的 hashcode 寻址，如果就在桶上那么直接返回值。  
如果是红黑树那就按照树的方式获取值。  
就不满足那就按照链表的方式遍历获取值。  
1.8 在 1.7 的数据结构上做了大的改动，采用红黑树之后可以保证查询效率（O(logn)），甚至取消了 ReentrantLock 改为了 synchronized，这样可以看出在新版的 JDK 中对 synchronized 优化是很到位的。

总结：  
本文主要介绍了一下几点：  
**1、ConcurrentHashMap 的数据结构  
2、ConcurrentHashMap 是如何解决hash冲突的，这部分其实和hashmap处理方式是一致的  
3、分析了ConcurrentHashMap 是如何实现线程安全的**

> [深入理解HashMap和CurrentHashMap][HashMap_CurrentHashMap]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70]: /images/20220219/853dc07fdd88436ebfb60a631636bd4d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 1]: /images/20220219/d57f43a0835d402695ac21d8ae8157cb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 2]: /images/20220219/29aa2e1c764743b9a68bea1aab7c8ad5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 3]: /images/20220219/0ecb8cdc6f434d70abd97ec233db1903.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 4]: /images/20220219/0f3195682f7c4574a1973443780d7a4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 5]: /images/20220219/18497a65d63b4bd5aa16f41bd14a2ee6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 6]: /images/20220219/5ab34ef725bc445084dfbe2f7d397c0f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 7]: /images/20220219/2f8708aa209049c0bf5b93d6fe0a2789.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 8]: /images/20220219/392a8191cc764ea19b5550be710bda9e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3cxMDE0MDc0Nzk0_size_16_color_FFFFFF_t_70 9]: /images/20220219/f6c3065f4a234fde87b8e9c2b5b1910e.png
[HashMap_CurrentHashMap]: https://www.cnblogs.com/fsychen/p/9361858.html