java并发编程（二十二）----(JUC集合)ConcurrentHashMap介绍

这一节我们来看一下并发的Map，ConcurrentHashMap和ConcurrentSkipListMap。ConcurrentHashMap通常只被看做并发效率更高的Map，用来替换其他线程安全的Map容器，比如Hashtable和Collections.synchronizedMap。ConcurrentSkipListMap提供了一种线程安全的并发访问的排序映射表。内部是SkipList（跳表）结构实现，在理论上能够在O(log(n))时间内完成查找、插入、删除操作。

ConcurrentHashMap简介

ConcurrentHashMap是一个线程安全的HashTable，它的主要功能是提供了一组和HashTable功能相同但是线程安全的方法。ConcurrentHashMap可以做到读取数据不加锁，并且其内部的结构可以让其在进行写操作的时候能够将锁的粒度保持地尽量地小，不用对整个ConcurrentHashMap加锁。

为了更好的理解 ConcurrentHashMap 高并发的具体实现，让我们先探索它的结构模型。

ConcurrentHashMap 类中包含两个静态内部类 HashEntry 和 Segment。HashEntry 用来封装映射表的键 / 值对；Segment 用来充当锁的角色，每个 Segment 对象守护整个散列映射表的若干个桶。每个桶是由若干个 HashEntry 对象链接起来的链表。一个 ConcurrentHashMap 实例中包含由若干个 Segment 对象组成的数组。

HashEntry类：

HashEntry 用来封装散列映射表中的键值对。在 HashEntry 类中，key，hash 和 next 域都被声明为 final 型，value 域被声明为 volatile 型 ：

static final class HashEntry<K, V> {
    final int hash;
     final K key;
     volatile V value;
     volatile ConcurrentHashMap.HashEntry<K, V> next;
     static final Unsafe UNSAFE;
     static final long nextOffset;

     HashEntry(int var1, K var2, V var3, ConcurrentHashMap.HashEntry<K, V> var4) {
         this.hash = var1;
         this.key = var2;
         this.value = var3;
         this.next = var4;
     }

     final void setNext(ConcurrentHashMap.HashEntry<K, V> var1) {
         UNSAFE.putOrderedObject(this, nextOffset, var1);
     }

     static {
         try {
             UNSAFE = Unsafe.getUnsafe();
             Class var0 = ConcurrentHashMap.HashEntry.class;
             nextOffset = UNSAFE.objectFieldOffset(var0.getDeclaredField("next"));
         } catch (Exception var1) {
             throw new Error(var1);
         }
     }
 }

Segment类：

Segment继承了ReentrantLock，表明每个segment都可以当做一个锁。Segment 中包含HashEntry 的数组，其可以守护其包含的若干个桶（HashEntry的数组）。Segment 在某些意义上有点类似于 HashMap了，都是包含了一个数组，而数组中的元素可以是一个链表。

static final class Segment<K, V> extends ReentrantLock implements Serializable {
    private static final long serialVersionUID = 2249069246763182397L;
    static final int MAX_SCAN_RETRIES = Runtime.getRuntime().availableProcessors() > 1?64:1;
/** 
     * table 是由 HashEntry 对象组成的数组
     * 如果散列时发生碰撞，碰撞的 HashEntry 对象就以链表的形式链接成一个链表
     * table 数组的数组成员代表散列映射表的一个桶
     * 每个 table 守护整个 ConcurrentHashMap 包含桶总数的一部分
     * 如果并发级别为 16，table 则守护 ConcurrentHashMap 包含的桶总数的 1/16 
     */ 
    transient volatile ConcurrentHashMap.HashEntry<K, V>[] table;
    transient int count;  //Segment中元素的数量
    transient int modCount; //对table的大小造成影响的操作的数量（比如put或者remove操作）
    transient int threshold; //阈值，Segment里面元素的数量超过这个值依旧就会对Segment进行扩容
    final float loadFactor; //负载因子，用于确定threshold

    Segment(float var1, int var2, ConcurrentHashMap.HashEntry<K, V>[] var3) {
        this.loadFactor = var1;
        this.threshold = var2;
        this.table = var3;
    }       
}

ConcurrentHashMap 的成员变量中，包含了一个 Segment 的数组（final Segment

ConcurrentHashMap结构图

ConcurrentHashMap引入了分割，并提供了HashTable支持的所有的功能。在ConcurrentHashMap中，支持多线程对Map做读操作，并且不需要任何的blocking。这得益于CHM将Map分割成了不同的部分，在执行更新操作时只锁住一部分。根据默认的并发级别(concurrency level)，Map被分割成16个部分，并且由不同的锁控制。这意味着，同时最多可以有16个写线程操作Map。试想一下，由只能一个线程进入变成同时可由16个写线程同时进入(读线程几乎不受限制)，性能的提升是显而易见的。但由于一些更新操作，如put(),remove(),putAll(),clear()只锁住操作的部分，所以在检索操作不能保证返回的是最新的结果。

ConcurrentHashMap默认的并发级别是16，但可以在创建CHM时通过构造函数改变。毫无疑问，并发级别代表着并发执行更新操作的数目，所以如果只有很少的线程会更新Map，那么建议设置一个低的并发级别。另外，ConcurrentHashMap还使用了ReentrantLock来对segments加锁。

经过前面的铺垫我们来正式对ConcurrentHashMap的使用进行剖析，重点关注get、put、remove这三个操作。

首先来看一下get的操作：

public V get(Object var1) {
   int var4 = this.hash(var1);
    long var5 = (long)((var4 >>> this.segmentShift & this.segmentMask) << SSHIFT) + SBASE;
    ConcurrentHashMap.Segment var2;
    if((var2 = (ConcurrentHashMap.Segment)UNSAFE.getObjectVolatile(this.segments, var5)) != null) {
        ConcurrentHashMap.HashEntry[] var3 = var2.table;
        if(var2.table != null) {
            for(ConcurrentHashMap.HashEntry var7 = (ConcurrentHashMap.HashEntry)UNSAFE.getObjectVolatile(var3, ((long)(var3.length - 1 & var4) << TSHIFT) + TBASE); var7 != null; var7 = var7.next) {
                Object var8 = var7.key;
                if(var7.key == var1 || var7.hash == var4 && var1.equals(var8)) {
                    return var7.value;
                }
            }
        }
    }

    return null;
}

1. 根据key，计算出hashCode；

2. 根据步骤1计算出的hashCode定位segment，如果segment不为null && segment.table也不为null，跳转到步骤3，否则，返回null，该key所对应的value不存在；

3. 根据hashCode定位table中对应的hashEntry，遍历hashEntry，如果key存在，返回key对应的value；

4. 步骤3结束仍未找到key所对应的value，返回null，该key锁对应的value不存在。

ConcurrentHashMap的get操作高效之处在于整个get操作不需要加锁。如果不加锁，ConcurrentHashMap的get操作是如何做到线程安全的呢？原因是volatile，所有的value都定义成了volatile类型，（上面介绍HashEntry类源码中提到：volatile V value）volatile可以保证线程之间的可见性，这也是用volatile替换锁的经典应用场景。

再来看一下put操作：

public V put(K var1, V var2) {
    if(var2 == null) {
        throw new NullPointerException();
    } else {
        int var4 = this.hash(var1);
        int var5 = var4 >>> this.segmentShift & this.segmentMask;
        ConcurrentHashMap.Segment var3;
        if((var3 = (ConcurrentHashMap.Segment)UNSAFE.getObject(this.segments, (long)(var5 << SSHIFT) + SBASE)) == null) {
            var3 = this.ensureSegment(var5);
        }

        return var3.put(var1, var4, var2, false);
    }
}

我们看到在第7行定义了一个Segment类型的 var3，然后调用了Segment的put方法存入map，我们不妨来看一下Segment的put方法：

final V put(K var1, int var2, V var3, boolean var4) {
        //1.获取锁，保证put操作的线程安全；
      ConcurrentHashMap.HashEntry var5 = this.tryLock()?null:this.scanAndLockForPut(var1, var2, var3);

      Object var6;
      try {
          ConcurrentHashMap.HashEntry[] var7 = this.table;
          int var8 = var7.length - 1 & var2;
//2.定位到HashEntry数组中具体的HashEntry
          ConcurrentHashMap.HashEntry var9 = ConcurrentHashMap.entryAt(var7, var8);
          ConcurrentHashMap.HashEntry var10 = var9;
//3.遍历HashEntry链表，假若待插入key已存在：
需要更新key所对应value（!onlyIfAbsent），更新oldValue -> newValue，跳转到步骤5；
否则，直接跳转到步骤5；
          while(true) {
              if(var10 == null) {
                  if(var5 != null) {
                      var5.setNext(var9);
                  } else {
                      var5 = new ConcurrentHashMap.HashEntry(var2, var1, var3, var9);
                  }

                  int var15 = this.count + 1;
                  if(var15 > this.threshold && var7.length < 1073741824) {
                      this.rehash(var5);
                  } else {
                      ConcurrentHashMap.setEntryAt(var7, var8, var5);
                  }

                  ++this.modCount;
                  this.count = var15;
                  var6 = null;
                  break;
              }

//4.遍历完HashEntry链表，key不存在，插入HashEntry节点，oldValue = null，跳转到步骤5
              Object var11 = var10.key;
              if(var10.key == var1 || var10.hash == var2 && var1.equals(var11)) {
                  var6 = var10.value;
                  if(!var4) {
                      var10.value = var3;
                      ++this.modCount;
                  }
                  break;
              }

              var10 = var10.next;
          }
      } finally {
//5.释放锁，返回oldValue
          this.unlock();
      }

      return var6;
  }

上面代码中已经做出解析，需要知道的是Segment的HashEntry数组采用开链法来处理冲突，我们知道散列最大的局限性就是空间利用率低，例如载荷因子为0.7，那么仍有0.3的空间未被利用。使用开链法可以使载荷因子为1，每个链上都挂常数个数据，对于哈希表的开链法来说，其开的空间都是按素数个依次往后开的空间，所以put操作的效率很高。

再来看一下remove操作：

public V remove(Object var1) {
     int var2 = this.hash(var1);
     ConcurrentHashMap.Segment var3 = this.segmentForHash(var2);
     return var3 == null?null:var3.remove(var1, var2, (Object)null);
 }

仍旧是调用了Segment的remove方法：

final V remove(Object var1, int var2, Object var3) {
    //获取锁
     if(!this.tryLock()) {
         this.scanAndLock(var1, var2);
     }

     Object var4 = null;

     try {
         ConcurrentHashMap.HashEntry[] var5 = this.table;
         int var6 = var5.length - 1 & var2;
         ConcurrentHashMap.HashEntry var7 = ConcurrentHashMap.entryAt(var5, var6);

         ConcurrentHashMap.HashEntry var10;
         for(ConcurrentHashMap.HashEntry var8 = null; var7 != null; var7 = var10) {
// 所有处于待删除节点之后的节点原样保留在链表中                
             var10 = var7.next;
             Object var9 = var7.key;
//找到要删除的节点
             if(var7.key == var1 || var7.hash == var2 && var1.equals(var9)) {
// 所有处于待删除节点之前的节点被克隆到新链表中
                 Object var11 = var7.value;
                 if(var3 != null && var3 != var11 && !var3.equals(var11)) {
                     break;
                 }

                 if(var8 == null) {
                     ConcurrentHashMap.setEntryAt(var5, var6, var10);
                 } else {
                     var8.setNext(var10);
                 }

                 ++this.modCount;
                 --this.count;
// 把桶链接到新的头结点
                 // 新的头结点是原链表中，删除节点之前的那个节点
                 var4 = var11;
                 break;
             }

             var8 = var7;
         }
     } finally {
         this.unlock();
     }

     return var4;
 }

我们来看两张图，执行删除前的原链表：

删除之后的链表：

从上图可以看出，删除节点 C 之后的所有节点原样保留到新链表中；删除节点 C 之前的每个节点被克隆到新链表中，注意：它们在新链表中的链接顺序被反转了。

在执行 remove 操作时，原始链表并没有被修改，也就是说：读线程不会受同时执行 remove 操作的并发写线程的干扰。

综合上面的分析我们可以看出，写线程对某个链表的结构性修改不会影响其他的并发读线程对这个链表的遍历访问。

总结

• ConcurrentHashMap 允许并发的读和线程安全的更新操作
• 在执行写操作时，ConcurrentHashMap 只锁住部分的Map
• 并发的更新是通过内部根据并发级别将Map分割成小部分实现的
• 高的并发级别会造成时间和空间的浪费，低的并发级别在写线程多时会引起线程间的竞争
• ConcurrentHashMap 的所有操作都是线程安全
• ConcurrentHashMap 返回的迭代器是弱一致性，fail-safe并且不会抛出ConcurrentModificationException异常
• ConcurrentHashMap 不允许null的键值

ConcurrentHashMap 是一个并发散列映射表的实现，它允许完全并发的读取，并且支持给定数量的并发更新。相比于 HashTable 和用同步包装器包装的 HashMap（Collections.synchronizedMap(new HashMap())），ConcurrentHashMap 拥有更高的并发性。在 HashTable 和由同步包装器包装的 HashMap 中，使用一个全局的锁来同步不同线程间的并发访问。同一时间点，只能有一个线程持有锁，也就是说在同一时间点，只能有一个线程能访问容器。这虽然保证多线程间的安全并发访问，但同时也导致对容器的访问变成串行化的了。