ConcurrentHashMap是线程安全且高效的HashMap

1 为什么要使用ConcurrentHashMap

线程不安全的HashMap

HashMap是Java中最常用的一个Map类，性能好、速度快，但不能保证线程安全，它可用null作为key/value

HashMap的线程不安全主要体现在resize时的死循环及使用迭代器时的fast-fail

在多线程环境下,使用HashMap进行put会引起死循环,所以在并发情况下不能使用HashMap.例如,执行以下代码会引起死循环.

HashMap在并发执行put会引起死循环,是因为多线程会导致HashMap的Entry链表成环,一旦成环,Entry的next节点永远不为空,产生死循环

效率低下的HashTable

线程安全的Map类，其public方法均用synchronize修饰

这表示在多线程操作时，每个线程在操作之前都会锁住整个map，待操作完成后才释放.如线程1使用put进行元素添加,线程2不但不能使用put方法进行添加元素,也不能使用get方法来获取元素,所以竞争越激烈效率越低,这必然导致多线程时性能不佳。

Hashtable也是线程安全的，所以也是key和value也是不可以null。

锁分段技术可有效提升并发访问率

我们可以回想一下，Hashtable（或者替代方案 Collections.synchronizedMap）的可伸缩性的主要障碍是它使用了一个 map 范围的锁，为了保证插入、删除或者检索操作的完整性必须保持这样一个锁，而且有时候甚至还要为了保证迭代遍历操作的完整性保持这样一个锁。这样一来，只要锁被保持，就从根本上阻止了其他线程访问 Map，即使处理器有空闲也不能访问，这样大大地限制了并发性。

ConcurrentHashMap 摒弃了单一的 map 范围的锁，取而代之的是由 16 个锁组成的集合，其中每个锁负责保护 hash bucket 的一个子集。锁主要由变化性操作（put() 和 remove()）使用。具有 16个独立的锁意味着最多可以有 16 个线程可以同时修改 map。这并不一定是说在并发地对 map 进行写操作的线程数少于 16 时，另外的写操作不会被阻塞――16 对于写线程来说是理论上的并发限制数目，但是实际上可能达不到这个值。但是，16 依然比 1 要好得多，而且对于运行于目前这一代的计算机系统上的大多数应用程序来说已经足够了

首先将数据分成一段一段地存储
然后给每一段数据配一把锁
当一个线程占用锁访问其中一个段数据的时候,其他段的数据也能被其他线程访问

2 ConcurrentHashMap的结构

通过ConcurrentHashMap的类图来分析ConcurrentHashMap的结构

ConcurrentHashMap是由Segment数组和HashEntry数组组成

2.1 可重入锁 : Segment

static final class Segment<K,V> extends ReentrantLock implements Serializable { 
       /** 
        * 在本 segment 范围内，包含的 HashEntry 元素的个数
        * 该变量被声明为 volatile 型
        */ 
       transient volatile int count; 
       /** 
        * table 被更新的次数
        */ 
       transient int modCount; 
       /** 
        * 当 table 中包含的 HashEntry 元素的个数超过本变量值时，触发 table 的再散列
        */ 
       transient int threshold; 
       /** 
        * table 是由 HashEntry 对象组成的数组
        * 如果散列时发生碰撞，碰撞的 HashEntry 对象就以链表的形式链接成一个链表
        * table 数组的数组成员代表散列映射表的一个桶
        * 每个 table 守护整个 ConcurrentHashMap 包含桶总数的一部分
        * 如果并发级别为 16，table 则守护 ConcurrentHashMap 包含的桶总数的 1/16 
        */ 
       transient volatile HashEntry<K,V>[] table; 
       /** 
        * 装载因子
        */ 
       final float loadFactor; 
       Segment(int initialCapacity, float lf) { 
           loadFactor = lf; 
           setTable(HashEntry.<K,V>newArray(initialCapacity)); 
       } 
       /** 
        * 设置 table 引用到这个新生成的 HashEntry 数组
        * 只能在持有锁或构造函数中调用本方法
        */ 
       void setTable(HashEntry<K,V>[] newTable) { 
           // 计算临界阀值为新数组的长度与装载因子的乘积
           threshold = (int)(newTable.length * loadFactor); 
           table = newTable; 
       } 
       /** 
        * 根据 key 的散列值，找到 table 中对应的那个桶（table 数组的某个数组成员）
        */ 
       HashEntry<K,V> getFirst(int hash) { 
           HashEntry<K,V>[] tab = table; 
           // 把散列值与 table 数组长度减 1 的值相“与”，
// 得到散列值对应的 table 数组的下标
           // 然后返回 table 数组中此下标对应的 HashEntry 元素
           return tab[hash & (tab.length - 1)]; 
       } 
}

Segment 类继承于 ReentrantLock ，从而使得 Segment 对象能充当锁的角色;

Segment 对象用来守护其（成员对象 table 中）包含的若干个桶;

table 是一个由 HashEntry 对象组成的数组;

table 数组的每一个数组成员就是散列映射表的一个桶.

count 变量是一个计数器，它表示每个 Segment 对象管理的 table 数组（若干个 HashEntry 组成的链表）包含的 HashEntry 对象的个数;

每一个 Segment 对象都有一个 count 对象来表示本 Segment 中包含的 HashEntry 对象的总数;

之所以在每个 Segment 对象中包含一个计数器，而不是在 ConcurrentHashMap 中使用全局的计数器，是为了避免出现“热点域”而影响 ConcurrentHashMap 的并发性

依次插入 ABC 三个 HashEntry 节点后，Segment 的结构示意图

2.2 HashEntry : 存储键值对

static final class HashEntry<K,V> { 
       final K key;                         // 声明 key 为 final 型
       final int hash;                      // 声明 hash 值为 final 型 
       volatile V value;                    // 声明 value 为 volatile 型
       final HashEntry<K,V> next;           // 声明 next 为 final 型 
       HashEntry(K key, int hash, HashEntry<K,V> next, V value) { 
           this.key = key; 
           this.hash = hash; 
           this.next = next; 
           this.value = value; 
       } 
}

一个ConcurrentHashMap里包含一个Segment数组;

Segment的结构和HashMap类似,是一种数组和链表结构;

一个Segment里包含一个HashEntry数组,每个HashEntry是一个链表结构的元素,每个Segment守护着一个HashEntry数组里的元素;

当对HashEntry数组的数据进行修改时,必须首先获得与它对应的Segment锁.

在 ConcurrentHashMap 中，在散列时如果产生“碰撞”，将采用“分离链接法”来处理“碰撞”：把“碰撞”的 HashEntry 对象链接成一个链表.

由于 HashEntry 的 next 域为 final 型，所以新节点只能在链表的表头处插入

下图是在一个空桶中依次插入 A，B，C 三个 HashEntry 对象后的结构图

注意：由于只能在表头插入，所以链表中节点的顺序和插入的顺序相反。

3 ConcurrentHashMap的初始化

3.1 Segment详解

3.1.1 Segment的索引与读取

ConcurrentHashMap类中包含三个与Segment相关的成员变量：

/**
 * Mask value for indexing into segments. The upper bits of a
 * key's hash code are used to choose the segment.
 */ final int segmentMask;
/**
 * Shift value for indexing within segments.
 */ final int segmentShift;
/**
 * The segments, each of which is a specialized hash table.
 */ final Segment<K,V>[] segments;

其中segments是Segment的原生数组，此数组的长度可以在ConcurrentHashMap的构造函数中使用并发度参数指定，其默认值为16

segmentShift是用来计算segments数组索引的位移量
segmentMask则是用来计算索引的掩码值

例如并发度为16时（即segments数组长度为16）

segmentShift为32-4=28（因为2的4次幂为16）
segmentMask则为1111（二进制）
索引的计算式如下：

int j = (hash >>> segmentShift) & segmentMask;

3.1.2 寻址方式

在读写某个Key时，先取该Key的哈希值;

并将哈希值的高N位对Segment个数取模从而得到该Key应该属于哪个Segment，接着如同操作HashMap一样操作这个Segment;

为了保证不同的值均匀分布到不同的Segment，需要通过如下方法计算哈希值.

private int hash(Object k) {
  int h = hashSeed;
  if ((0 != h) && (k instanceof String)) {
    return sun.misc.Hashing.stringHash32((String) k);
  }
  h ^= k.hashCode();
  h += (h <<  15) ^ 0xffffcd7d;
  h ^= (h >>> 10);
  h += (h <<   3);
  h ^= (h >>>  6);
  h += (h <<   2) + (h << 14);
  return h ^ (h >>> 16);
}

同样为了提高取模运算效率，通过如下计算

ssize即为大于concurrencyLevel的最小的2的N次方
segmentMask为2^N-1

对于某一个Key的哈希值，只需要向右移segmentShift位以取高sshift位，再与segmentMask取与操作即可得到它在Segment数组上的索引

int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
  ++sshift;
  ssize <<= 1;
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];

3.1.3 多线程并发访问一个共享变量

为了保证逻辑的正确，可以采用以下方法：

加锁，性能最低，能保证原子性、可见性，防止指令重排
volatile修饰，性能中等，能保证可见性，防止指令重排
使用getObjectVolatile，性能最好，可防止指令重排

因此ConcurrentHashMap选择了使用Unsafe的getObjectVolatile来读取segments中的元素

// Unsafe mechanics private static final sun.misc.Unsafe UNSAFE;
private static final long SBASE;
private static final int SSHIFT;
private static final long TBASE;
private static final int TSHIFT;
private static final long HASHSEED_OFFSET;
private static final long SEGSHIFT_OFFSET;
private static final long SEGMASK_OFFSET;
private static final long SEGMENTS_OFFSET;
static {
    int ss, ts;
    try {
        UNSAFE = sun.misc.Unsafe.getUnsafe();
        Class tc = HashEntry[].class;
        Class sc = Segment[].class;
        TBASE = UNSAFE.arrayBaseOffset(tc);
        SBASE = UNSAFE.arrayBaseOffset(sc);
        ts = UNSAFE.arrayIndexScale(tc);
        ss = UNSAFE.arrayIndexScale(sc);
        HASHSEED_OFFSET = UNSAFE.objectFieldOffset(
            ConcurrentHashMap.class.getDeclaredField("hashSeed"));
        SEGSHIFT_OFFSET = UNSAFE.objectFieldOffset(
            ConcurrentHashMap.class.getDeclaredField("segmentShift"));
        SEGMASK_OFFSET = UNSAFE.objectFieldOffset(
            ConcurrentHashMap.class.getDeclaredField("segmentMask"));
        SEGMENTS_OFFSET = UNSAFE.objectFieldOffset(
            ConcurrentHashMap.class.getDeclaredField("segments"));
    } catch (Exception e) {
        throw new Error(e);
    }
    if ((ss & (ss-1)) != 0 || (ts & (ts-1)) != 0)
        throw new Error("data type scale not a power of two");
    SSHIFT = 31 - Integer.numberOfLeadingZeros(ss);
    TSHIFT = 31 - Integer.numberOfLeadingZeros(ts);
}
private Segment<K,V> segmentForHash(int h) {
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    return (Segment<K,V>) UNSAFE.getObjectVolatile(segments, u);
}

观察segmentForHash(int h)方法可知

首先使用(h >>> segmentShift) & segmentMask
计算出该h对应的segments索引值（假设为x）
然后使用索引值(x<<SSHIFT) + SBASE计算出segments中相应Segment的地址
最后使用UNSAFE.getObjectVolatile(segments,u)取出相应的Segment，并保持volatile读的效果

3.1.4 Segment的锁

Segment继承了ReentrantLock，因此它实际上是一把锁.

在进行put、remove、replace、clear等需要改动内部内容的操作时，都要进行加锁操作，其代码一般是这样的：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
  //实际代码……
        }
    } finally {
        unlock();
    }
    return oldValue;
}

首先调用tryLock，如果加锁失败，则进入scanAndLockForPut(key, hash, value)

该方法实际上是先自旋等待其他线程解锁，直至指定的次数MAX_SCAN_RETRIES;

若自旋过程中，其他线程释放了锁，导致本线程直接获得了锁，就避免了本线程进入等待锁的场景，提高了效率;

若自旋一定次数后，仍未获取锁，则调用lock方法进入等待锁的场景.

采用这种自旋锁和独占锁结合的方法，在很多场景下能够提高Segment并发操作数据的效率.

初始化方法是通过initialCapacity、loadFactor和concurrencyLevel等几个

参数来初始化segment数组、段偏移量segmentShift、段掩码segmentMask和每个segment里的HashEntry数组来实现的.

初始化segments数组

     if (concurrencyLevel > MAX_SEGMENTS)
          concurrencyLevel = MAX_SEGMENTS;
          int sshift = 0;
          int ssize = 1;
          while (ssize < concurrencyLevel) {
              ++sshift;
              ssize <<= 1;
     }
       segmentShift = 32 - sshift;
       segmentMask = ssize - 1;
       this.segments = Segment.newArray(ssize);

segments数组的长度ssize是通过concurrencyLevel计算得出的

为了能通过按位与的散列算法来定位segments数组的索引,必须保证segments数组的长度是2的N次方,所以必须计算出一个大于或等于concurrencyLevel的最小的2的N次方值来作为segments数组的长度

concurrencyLevel的最大值是65535,这意味着segments数组的长度最大为65536,对应的二进制是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

初始化每个segment
输入参数initialCapacity是ConcurrentHashMap的初始化容量,loadfactor是每个segment的负载因子,在构造方法里需要通过这两个参数来初始化数组中的每个segment.

if (initialCapacity > MAXIMUM_CAPACITY)
     initialCapacity = MAXIMUM_CAPACITY;
 int c = initialCapacity / ssize;
 if (c * ssize < initialCapacity)
     ++c;
 int cap = 1;
 while (cap < c)
     cap <<= 1;
 for (int i = 0; i < this.segments.length; ++i)
     this.segments[i] = new Segment<K, V>(cap, loadFactor);

cap就是segment里HashEntry数组的长度，它等于。若c大于1，就会取≥c的2的N次方值。所以cap就是2的N（N≥0）次方。

segment的容量

threshold＝（int）cap*loadFactor

默认initialCapacity等于16，loadfactor等于0.75，通过运算cap等于1，threshold等于零。

定位Segment
ConcurrentHashMap使用分段锁Segment来保护不同段的数据,在插入和获取元素时,先通过散列算法定位到Segment

private static int hash(int h) {
     h += (h << 15) ^ 0xffffcd7d;
     h ^= (h >>> 10);
     h += (h << 3);
     h ^= (h >>> 6);
     h += (h << 2) + (h << 14);
     return h ^ (h >>> 16);
 }

进行再散列，是为了减少散列冲突，使元素能够均匀地分布在不同的Segment上，从而提高容器的存取效率。

假如散列的质量差到极点，那么所有的元素都在一个Segment中，不仅存取元素缓慢，分段锁也会失去意义。

ConcurrentHashMap通过以下散列算法定位segment

final Segment<K,V> segmentFor(int hash) {
      return segments[(hash >>> segmentShift) & segmentMask];
}

默认情况下segmentShift为28,segmentMask为15,再散列后的数最大是32位二进制数据,向右无符号移动28位,即让高4位参与到散列运算中,(hash>>>segmentShift)&segmentMask的运算结果分别是4、15、7和8,可以看到散列值没有发生冲突.

3.1.5 HashEntry

如果说ConcurrentHashMap中的segments数组是第一层hash表，则每个Segment中的HashEntry数组（transient volatile HashEntry<K,V>[] table）是第二层hash表;

每个HashEntry有一个next属性，因此它们能够组成一个单向链表.

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);
        }
    }
}
/**
 * Gets the ith element of given table (if nonnull) with volatile
 * read semantics. Note: This is manually integrated into a few
 * performance-sensitive methods to reduce call overhead.
 */ @SuppressWarnings("unchecked")
static final <K,V> HashEntry<K,V> entryAt(HashEntry<K,V>[] tab, int i) {
    return (tab == null) ? null :
        (HashEntry<K,V>) UNSAFE.getObjectVolatile
        (tab, ((long)i << TSHIFT) + TBASE);
}
/**
 * Sets the ith element of given table, with volatile write
 * semantics. (See above about use of putOrderedObject.)
 */ static final <K,V> void setEntryAt(HashEntry<K,V>[] tab, int i,
                                   HashEntry<K,V> e) {
    UNSAFE.putOrderedObject(tab, ((long)i << TSHIFT) + TBASE, e);
}

与Segment类似，HashEntry使用UNSAFE.putOrderedObject来设置它的next成员变量，这样既可以提高性能，又能保持并发可见性;

同时，entryAt方法和setEntryAt方法也使用了UNSAFE.getObjectVolatile和UNSAFE.putOrderedObject来读取和写入指定索引的HashEntry.

总之，Segment数组和HashEntry数组的读取写入一般都是使用UNSAFE.

3.2 用 HashEntery 对象的不变性来降低读操作对加锁的需求

在HashEntry 类的定义中我们可以看到，HashEntry 中的 key，hash，next 都声明为final型;

这意味着，不能把节点添加到链接的中间和尾部，也不能在链接的中间和尾部删除节点;

这个特性可以保证：在访问某个节点时，这个节点之后的链接不会被改变;

可以大大降低处理链表时的复杂性

同时，HashEntry 类的 value 域被声明为 Volatile ;

JMM可以保证：某个写线程对 value 域的写入马上可以被后续的某个读线程“看”到.

在 ConcurrentHashMap 中，不允许用 null 作为键/值

ConcurrentMaps（ConcurrentHashMaps，ConcurrentSkipListMaps）不允许使用null的主要原因是，无法容纳在非并行映射中几乎无法容忍的歧义。最主要的是，如果map.get(key)return null，则无法检测到该键是否显式映射到null该键。在非并行映射中，可以通过进行检查 map.contains(key)，但在并行映射中，两次调用之间的映射可能已更改。

当读线程读到某个 HashEntry 的 value 域的值为 null 时，便知道产生了冲突——发生了重排序现象，需要加锁后重新读入这个 value 值

这些特性互相配合，使得读线程即使在不加锁状态下，也能正确访问 ConcurrentHashMap

下面我们分别来分析线程写入的两种情形：对散列表做非结构性修改的操作和对散列表做结构性修改的操作。

3.2.1 非结构性修改操作

只是更改某个 HashEntry 的 value 域值;

由于对 Volatile 变量的写操作将与随后对这个变量的读操作进行同步;

当一个写线程修改了某个 HashEntry 的 value 域后,另一个读线程读这个值域,JMM能够保证读线程读取的一定是更新后的值.

所以，写线程对链表的非结构性修改能够被后续不加锁的读线程“看到”

3.2.3 结构性修改

实质上是对某个桶指向的链表做结构性修改;

如果能够确保：在读线程遍历一个链表期间，写线程对这个链表所做的结构性修改不影响读线程继续正常遍历这个链表;

那么读 / 写线程之间就可以安全并发访问这个 ConcurrentHashMap

结构性修改操作包括 put，remove，clear

下面我们分别分析这三个操作

clear 操作只是把 ConcurrentHashMap 中所有的桶“置空”;
每个桶之前引用的链表依然存在，只是桶不再引用到这些链表（所有链表的结构并没有被修改）;
正在遍历某个链表的读线程依然可以正常执行对该链表的遍历.
从下面的代码put 操作中，我们可以看出：put 操作如果需要插入一个新节点到链表中时 , 会在链表头部插入这个新节点;
此时，链表中的原有节点的连接并没有被修改;
也就是说：插入新键 / 值对到链表中的操作不会影响读线程正常遍历这个链表.

remove

V remove(Object key, int hash, Object value) { 
           lock();         // 加锁
           try{ 
               int c = count - 1; 
               HashEntry<K,V>[] tab = table; 
               // 根据散列码找到 table 的下标值
               int index = hash & (tab.length - 1); 
               // 找到散列码对应的那个桶
               HashEntry<K,V> first = tab[index]; 
               HashEntry<K,V> e = first; 
               while(e != null&& (e.hash != hash || !key.equals(e.key))) 
                   e = e.next; 
               V oldValue = null; 
               if(e != null) { 
                   V v = e.value; 
                   if(value == null|| value.equals(v)) { // 找到要删除的节点
                       oldValue = v; 
                       ++modCount; 
                       // 所有处于待删除节点之后的节点原样保留在链表中
                       // 所有处于待删除节点之前的节点被克隆到新链表中
                       HashEntry<K,V> newFirst = e.next;// 待删节点的后继结点
                       for(HashEntry<K,V> p = first; p != e; p = p.next) 
                           newFirst = new HashEntry<K,V>(p.key, p.hash, 
                                                         newFirst, p.value); 
                       // 把桶链接到新的头结点
                       // 新的头结点是原链表中，删除节点之前的那个节点
                       tab[index] = newFirst; 
                       count = c;      // 写 count 变量
                   } 
               } 
               return oldValue; 
           } finally{ 
               unlock();               // 解锁
           } 
       }

和get一样，先根据散列码找到具体的链表；

然后遍历这个链表找到要删除的节点；

最后把待删除节点之后的所有节点原样保留在新链表中,把待删除节点之前的每个节点克隆到新链表中.

假设写线程执行 remove 操作，要删除链表的 C 节点，另一个读线程同时正在遍历这个链表

所待删除节点 C 后的所有节点原样保留到新链表中;

所待删除节点 C 前的每个节点被克隆到新链表中

注意：它们在新链表中的链接顺序被反转了

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

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

Java 集合源码解析 - ConcurrentHashMap（JDK7）（上）