1. ConcurrentHashMap的简介

JDK1.7中ConcurrentHashMap采用数组+链表的数据结构，哈希表整体上是一个Segment的数组，而每个分段Segment又是一个HashEntry的数组，每个HashEntry是一个链表。

Java7 ConcurrentHashMap结构

• JDK1.8的实现降低了锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）

• JDK1.8版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用synchronized来进行同步，所以不需要分段锁的概念，也就不需要Segment这种数据结构了，由于粒度的降低，实现的复杂度也增加了

• JDK1.8使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档

Java8 ConcurrentHashMap结构

2. JDK1.7 ConcurrentHashMap原理

2.1 ConcurrentHashMap锁分段

锁粗化是锁优化的一种重要措施，而锁粗化又包含"lock-splitting"（锁定拆分）和"lock-stripping"（锁条带化）。读写锁分离是一种典型的锁定拆分方式，JUC中的ReentrantReadWriteLock就是一种读写分离锁，锁定条带化是指将一把“大锁”拆分成若干个“小锁”来降低锁的竞争。ConcurrentHashMap就是通过锁条带化来做锁的优化。我们都知道ConcurrentHashMap是分段的，它的表是一个Segment数组：

/**
 * The segments, each of which is a specialized hash table
 */
final Segment<K,V>[] segments;

而每个Segment都是继承了一个ReentrantLock：

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

所以ConcurrentHashMap的每个Segment都是一把锁，不同的Segment之间的读写不构成竞争，大大降低了锁的竞争。既然每个Segment都是一把锁，那么这个segment数组的长度是多少呢？也就是说整个表我们需要多少把锁呢？

public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();

    if (concurrencyLevel > MAX_SEGMENTS) // MAX_SEGMENTS是指整个表最多能分成多少个segment，也即是多少把锁
        concurrencyLevel = MAX_SEGMENTS;

    // Find power-of-two sizes best matching arguments
    int sshift = 0;
    int ssize = 1;
    while (ssize < concurrencyLevel) { // 找到不小于我们指定的concurrencyLevel的2的幂次方的一个数作为segment数组的长度
        ++sshift;
        ssize <<= 1;
    }
    segmentShift = 32 - sshift;
    segmentMask = ssize - 1;
    this.segments = Segment.newArray(ssize);

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

在ConcurrentHashMap的构造函数中我们指定了concurrencyLevel，也即是多少把锁。这个数量不能超过上限:MAX_SEGMENTS(1 << 16)，锁的个数必须是2的幂次方，如果我们指定的concurrencyLevel不是2的幂次方，构造函数会找到最接近的一个不小于我们指定的值的一个2的幂次方数作为segment数组长度。例如：我们指定concurrencyLevel为15，则最终segment数组长度为16，也即是表一共有16把锁。设想两个线程同时向表中插入元素，线程1插入的第0个segment，线程2插入的是第1个segment，线程1和线程2互不影响，能够同时并行。但是HashTable就做不到这一点。

2.2 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; // 通过位运算得到segment的索引位置
    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);
}

ConcurrentHashMap不支持插入null的值，因此首先校验value是否为null。如果value是null则抛出异常。
注意这里计算segment索引方式是: (hash >>> segmentShift) & segmentMask; 而不是hash % segment数组长度。<font color="#FF0000">这儿是一个优化：因为取模"%"操作相对位运算来说是很慢的，因此这里是用位运算来得到segment索引。而当segment数组长度是2的幂次方记为segmentSize时:hash % segmentSize == hash & (segmentSize - 1)。</font>这里不做证明。因此segmentSize必须是2的幂次方。来看看Segment中的put()方法：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry<K,V> node = tryLock() ? null :  // 获取segment的锁，这里会有一个优化：获取锁的时候首先会通过 `tryLock()` 尝试若干次
        scanAndLockForPut(key, hash, value);  // 如果若干次之后还没有获取锁，则用 `lock()` 方法阻塞等待，直到获取锁
    V oldValue;
    try {
        HashEntry<K,V>[] tab = table;
        int index = (tab.length - 1) & hash; // 得到segment的table的索引，也是通过位运算
        HashEntry<K,V> first = entryAt(tab, index); // table中index位置的first节点
        for (HashEntry<K,V> e = first;;) {
            if (e != null) {
                K k;
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) { // 对应的key已经有了value
                    oldValue = e.value;
                    if (!onlyIfAbsent) { // 是否覆盖原来的value
                        e.value = value; // 覆盖原来的value
                        ++modCount;
                    }
                    break;
                }
                e = e.next;  // 遍历
            }
            else {
                if (node != null)
                    node.setNext(first); // 如果node已经在scanAndLockForPut()方法中初始化过
                else
                    node = new HashEntry<K,V>(hash, key, value, first); // 如果node为null，则初始化
                int c = count + 1;
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    rehash(node); // 如果超过阈值，则扩容
                else
                    setEntryAt(tab, index, node); // 通过UNSAFE设置table数组的index位置的元素为node
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

首先，会获取segment的锁，然后判断添加元素后是否需要扩容。<font color="#FF000000">注意这里的扩容是指Segment中的HashEntry[] table表数组扩容，而不是最外层的segment[]数组扩容。segment[]数组是不可扩展的，在构造函数中已经确定了segment[]数组的长度。</font>
接着同样通过位运算得到待添加元素在HashEntry[] table数组中的位置。接着再判断这个链表中是否已经存在这个key，如果存在并且onlyIfAbsent为false，就覆盖原value；如果链表不存在key，则将新的node通过UNSAFE放到table数组指定的位置。

2.3 get操作

get操作比较简单，不需要加锁。可见性由volatile来保证：HashEntry的value是volatile的，Segment中的HashEntry[] table数组也是volatile。这保证了其他线程对哈希表的修改能够及时地被读线程发现。

public V get(Object key) {
    Segment<K,V> s; // manually integrate access methods to reduce overhead
    HashEntry<K,V>[] tab;
    int h = hash(key);
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 计算key应该落在segments数组的哪个segment中
    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); // 计算key应该落在table的哪个位置
             e != null; e = e.next) {
            K k;
            if ((k = e.key) == key || (e.hash == h && key.equals(k))) // 如果key和当前节点的key指向同一块内存地址或者当前节点的hash
                return e.value;                                       // 等于key的hash并且key"equals"当前节点的key则说明当前节点就是目标节点
        }
    }
    return null; // key不在当前哈希表中，返回null
}

2.4 rehash

private void rehash(HashEntry<K,V> node) {
    /*
     * Reclassify nodes in each list to new table.  Because we
     * are using power-of-two expansion, the elements from
     * each bin must either stay at same index, or move with a
     * power of two offset. We eliminate unnecessary node
     * creation by catching cases where old nodes can be
     * reused because their next fields won't change.
     * Statistically, at the default threshold, only about
     * one-sixth of them need cloning when a table
     * doubles. The nodes they replace will be garbage
     * collectable as soon as they are no longer referenced by
     * any reader thread that may be in the midst of
     * concurrently traversing table. Entry accesses use plain
     * array indexing because they are followed by volatile
     * table write.
     */
    HashEntry<K,V>[] oldTable = table;
    int oldCapacity = oldTable.length;
    int newCapacity = oldCapacity << 1; // 每次扩容成原来capacity的2倍，这样元素在新的table中的索引要么不变要么是原来的索引加上2的一个倍数
    threshold = (int)(newCapacity * loadFactor); // 新的扩容阈值
    HashEntry<K,V>[] newTable =
        (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 新的segment table数组
    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;  // 在拷贝原来链表的元素到新的table中时有个优化：通过遍历找到原先链表中的lastRun节点，这个节点以及它的后续节点都不需要重新拷贝，直接放到新的table中就行
                     last != null;
                     last = last.next) {
                    int k = last.hash & sizeMask;
                    if (k != lastIdx) {
                        lastIdx = k;
                        lastRun = last;
                    }
                }
                newTable[lastIdx] = lastRun; // lastRun节点以及lastRun后续节点都不需要重新拷贝，直接赋值引用
                // Clone remaining nodes
                for (HashEntry<K,V> p = e; p != lastRun; p = p.next) { // 循环拷贝原先链表lastRun之前的节点到新的table链表中
                    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]); // rehash之后，执行添加新的节点
    newTable[nodeIndex] = node;
    table = newTable;
}

由于rehash过程中是加排它锁的，这样其他的写入请求将被阻塞等待。而对于读请求，需要分情况讨论：读请求在rehash之前，此时segment中的table数组指针还是指向原先旧的数组，所以读取是安全的；如果读请求在rehash之后，因为table数组和HashEntry的value都是volatile，所以读线程也能及时读取到更新的值，因此也是线程安全的。所以rehash不会影响到读。

2.5 remove操作

public V remove(Object key) {
    int hash = hash(key);
    Segment<K,V> s = segmentForHash(hash); // key落在哪个segment中
    return s == null ? null : s.remove(key, hash, null); // 如果segment为null，则说明哈希表中没有key，直接返回null，否则调用Segment的remove
}

final V remove(Object key, int hash, Object value) { // Segment的remove方法
    if (!tryLock()) // 获取Segment的锁，套路还是一样的首先进行若干次 `tryLock()`, 如果失败了则通过 `lock()` 方法阻塞等待直到获取锁
        scanAndLock(key, hash);
    V oldValue = null;
    try {
        HashEntry<K,V>[] tab = table;
        int index = (tab.length - 1) & hash;
        HashEntry<K,V> e = entryAt(tab, index); // 找到key具体在table的哪个链表中，e代表链表当前节点
        HashEntry<K,V> pred = null; // pred代表e节点的前置节点
        while (e != null) {
            K k;
            HashEntry<K,V> next = e.next;
            if ((k = e.key) == key ||
                (e.hash == hash && key.equals(k))) { // 找到了这个key对应的HashEntry
                V v = e.value;
                if (value == null || value == v || value.equals(v)) {
                    if (pred == null) // 如果当前节点的前置节点为空，说明要删除的节点是当前链表的头节点，直接将当前链表的头节点指向当前节点的next就可以了
                        setEntryAt(tab, index, next);
                    else
                        pred.setNext(next); // 否则修改前置节点的next指针，指向当前节点的next节点，这样当前节点将不再"可达"，可以被GC回收
                    ++modCount;
                    --count;
                    oldValue = v;
                }
                break;
            }
            pred = e;
            e = next;
        }
    } finally {
        unlock(); // 解锁
    }
    return oldValue;
}

remove时，首先会找到这个key落在哪个Segment中，如果key没有落在任何一个Segment中，说明key不存在，直接返回null。找到具体的Segment后，调用Segment的remove方法来进行删除：找到key落在Segment的table数组中的哪个链表中，遍历链表，如果要删除的节点是当前链表的头节点，则直接修改链表的头指针为当前节点的next节点；如果要删除的节点不是头节点，继续遍历找到目标节点，修改目标节点的前置节点的next指针指向目标节点的next节点完成操作。
安全性分析：remove时首先会加锁，其他mutable请求都是会被阻塞的，对于读请求也是安全的。如果读取的key不是当前要删除的key不会有任何问题。如果读取的key恰巧是当前需要删除key：读请求在remove之前，这时可以读取到；如果读请求在remove操作之后，由于HashEntry的next指针都是volatile的，所以读线程也是可以及时发现这个key已经被删除了的。也是安全的。

2.6 size操作

ConcurrentHashMap的size操作在Java7实现还是比较有意思的。其首先会进行若干次尝试，每次对各个Segment的count求和，如果任意前后两次求和结果相同，则说明在这段时间之内各个Segment的元素个数没有改变，直接返回当前的求和结果就行了。如果超过一定重试次数之后，会采取悲观策略，直接锁定各个Segment，然后依次求和。注意这里是锁定所有Segment，因此在采取悲观策略时整个哈希表都不能有写入操作。这里先乐观再悲观的策略和前面的put操作中的scanAndLockForPut有异曲同工之妙。

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; // 首先不加锁，每次对各个Segment的count累加求和，如果任意两次的累加结果相同，则直接返回这个结果；超过一定的次数之后悲观锁定所有的Segment，再求和。锁定之后整个哈希表不能有任何的写入操作。
    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 (;;) {
            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的count累加，不加锁
                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; // 如果溢出，返回最大整型作为结果，否则返回累加结果
}

2.7 jdk1.7实现总结

1. ConcurrentHashMap采取分段锁，是一种典型的"lock-stripping"策略，目的是为了降低高并发情况下的锁竞争。
2. rehash过程的扩容不是segment数组的扩容，而是Segment中HashEntry[] table数组的扩容，Segment[] segments数组是final的，在哈希表初始化完成后不再更改。
3. ConcurrentHashMap中很多地方用到了volatile，保证了可见性，例如，Segment中的HashEntry[] table数组，HashEntry中的value和next指针都是volatile的。
4. 加锁是低效的，线程的上下文切换需要消耗性能，因此ConcurrentHashMap很多地方都用到了乐观重试的策略，在超过一定次数之后再采取悲观策略。例如size操作。
5. 在使用ConcurrentHashMap之前请明确你的数据结构是否真的会有多线程并发操作，如果没有，仅仅是单线程操作，请使用HashMap，因为在不考虑并发安全性问题时，不论是HashTable还是ConcurrentHashMap他们的性能都没有HashMap好。

3. JDK1.8 ConcurrentHashMap原理

3.1 get操作源码

• 首先计算hash值，定位到该table索引位置，如果是首节点符合就返回

• 如果遇到扩容的时候，会调用标志正在扩容节点ForwardingNode的find方法，查找该节点，匹配就返回

• 以上都不符合的话，就往下遍历节点，匹配就返回，否则最后就返回null

//会发现源码中没有一处加了锁
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode()); //计算hash
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {//读取首节点的Node元素
        if ((eh = e.hash) == h) { //如果该节点就是首节点就返回
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        //hash值为负值表示正在扩容，这个时候查的是ForwardingNode的find方法来定位到nextTable来
        //eh=-1，说明该节点是一个ForwardingNode，正在迁移，此时调用ForwardingNode的find方法去nextTable里找。
        //eh=-2，说明该节点是一个TreeBin，此时调用TreeBin的find方法遍历红黑树，由于红黑树有可能正在旋转变色，所以find里会有读写锁。
        //eh>=0，说明该节点下挂的是一个链表，直接遍历该链表即可。
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {//既不是首节点也不是ForwardingNode，那就往下遍历
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

3.2 volatile原理

对于可见性，Java提供了volatile关键字来保证可见性、有序性。但不保证原子性。

普通的共享变量不能保证可见性，因为普通共享变量被修改之后，什么时候被写入主存是不确定的，当其他线程去读取时，此时内存中可能还是原来的旧值，因此无法保证可见性。

• volatile关键字对于基本类型的修改可以在随后对多个线程的读保持一致，但是对于引用类型如数组，实体bean，仅仅保证引用的可见性，但并不保证引用内容的可见性。。

• 禁止进行指令重排序。

背景：为了提高处理速度，处理器不直接和内存进行通信，而是先将系统内存的数据读到内部缓存（L1，L2或其他）后再进行操作，但操作完不知道何时会写到内存。

• 如果对声明了volatile的变量进行写操作，JVM就会向处理器发送一条指令，将这个变量所在缓存行的数据写回到系统内存。但是，就算写回到内存，如果其他处理器缓存的值还是旧的，再执行计算操作就会有问题。

• 在多处理器下，为了保证各个处理器的缓存是一致的，就会实现缓存一致性协议，当某个CPU在写数据时，如果发现操作的变量是共享变量，则会通知其他CPU告知该变量的缓存行是无效的，因此其他CPU在读取该变量时，发现其无效会重新从主存中加载数据。

image

总结下来：

第一：使用volatile关键字会强制将修改的值立即写入主存；

第二：使用volatile关键字的话，当线程2进行修改时，会导致线程1的工作内存中缓存变量的缓存行无效（反映到硬件层的话，就是CPU的L1或者L2缓存中对应的缓存行无效）；

第三：由于线程1的工作内存中缓存变量的缓存行无效，所以线程1再次读取变量的值时会去主存读取。

3.3 volatile修饰数组有什么意义?

我们知道volatile可以修饰数组的，只是意思和它表面上看起来的样子不同。举个栗子，volatile int array[10]是指array的地址是volatile的而不是数组元素的值是volatile的.

3.4 用volatile修饰的Node

get操作可以无锁是由于Node的元素val和指针next是用volatile修饰的，在多线程环境下线程A修改结点的val或者新增节点的时候是对线程B可见的。

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    //可以看到这些都用了volatile修饰
    volatile V val;
    volatile Node<K,V> next;
 
    Node(int hash, K key, V val, Node<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.val = val;
        this.next = next;
    }
 
    public final K getKey()       { return key; }
    public final V getValue()     { return val; }
    public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
    public final String toString(){ return key + "=" + val; }
    public final V setValue(V value) {
        throw new UnsupportedOperationException();
    }
 
    public final boolean equals(Object o) {
        Object k, v, u; Map.Entry<?,?> e;
        return ((o instanceof Map.Entry) &&
                (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
                (v = e.getValue()) != null &&
                (k == key || k.equals(key)) &&
                (v == (u = val) || v.equals(u)));
    }
 
    /**
     * Virtualized support for map.get(); overridden in subclasses.
     */
    Node<K,V> find(int h, Object k) {
        Node<K,V> e = this;
        if (k != null) {
            do {
                K ek;
                if (e.hash == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))
                    return e;
            } while ((e = e.next) != null);
        }
        return null;
    }
}

既然volatile修饰数组对get操作没有效果那加在数组上的volatile的目的是什么呢？

其实就是为了使得Node数组在扩容的时候对其他线程具有可见性而加的volatile

3.5 Jdk1.8 ConcurrentHashMap总结

• 在1.8中ConcurrentHashMap的get操作全程不需要加锁，这也是它比其他并发集合比如hashtable、用Collections.synchronizedMap()包装的hashmap;安全效率高的原因之一。

• get操作全程不需要加锁是因为Node的成员val是用volatile修饰的和数组用volatile修饰没有关系。

• 数组用volatile修饰主要是保证在数组扩容的时候保证可见性。