JAVA-JDK1.7-ConCurrentHashMap-源码并且debug说明

概述

 在一个程序员的成长过程就一定要阅读源码，并且了解其中的原理，只有这样才可以深入了解其中的功能，就像ConCurrentHashMap 是线程安全的，到底是如何安全的？以及如何正确使用它？rehash在什么情况？红黑树存储原理？不了解其中源码原理是不行的。所以今天就写一篇Java源码的，关于ConCurrentHashMap的源码，大部分来自于网上摘抄，有的需要自己测试验证，现作为随笔记录于此，便于记录整理 ，并且分享给大家共同进步。

ConCurrentHashMap的出生背景

哈希表是非常高效的数据结构，在JAVA开发中，最常见最频繁使用的就是HashMap和HashTable，但是在线程的使用范围中就显得乏力了，简单来说，HashMap是线程不安全的，在并发中环境中，可能会形成环状链表，导致get操作时候，cpu空转，死循环。HashTable虽然和HashMap原理几乎一样，差别只有2个；第一是，不允许key和value为null；第二是，他是线程安全的。但是HashTable线程安全的策略简单粗暴，所有相关操作都是synchronized的，相当于给整个hash表加了一把大锁，在多线程访问的时候，只要有一个线程访问或者操作这个对象，那其他线程只能阻塞，相当于将所有操作串行化，牺牲了很大的性能。所以就需要性能好又安全的哈希表结构，所以就有了本文的主角ConCurrentMap。现在我们逐步说明不同版本的原理。

ConCurrentHashMap简介

 ConCurrentHashMap采用了锁分段技术：加入容器中有多把锁，没一把锁用于锁容器中一部分数据，那么当线程访问容器中里不同的数据段的数据时，线程间就不会存在锁竞争，从而就可以提高并发访问效率，简而言之，就是将数据一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也可能被其他线程访问。

ConCurrentHashMap的数据结构是由一个Segment数组和多个HashEntry组成，并且在代码中HashEntry和Segment是两个静态内部类，数据结构如下图所示：

Segment数组的意义就是将一个大的table分割成多个小的table 来进行加锁，也就是上面提到的锁分离技术，而每一个Segment元素存储的是HashEntry数组+链表，这个和HashMap的数据存储结构一样。HashEntry用来封装映射表的键/值对；Segment充当锁的角色，每个Segment对象守护散列表的若干个的桶。每个桶是由若干个HashEntry对象连接起来的链表。

ConCurrentHashMap不允许key为null，ConCurrentHashMap不允许空值的主要原因是，在非并发映射中几乎不能容忍的模糊性是无法容纳的，主要一点是如果map.get(key)返回null，则无法检测key是否显示映射为null或未映射，在非并发中你可以使用map.contains(key)进行检查，但在并发映射中，映射可能在调用之间发生变化。

继承关系：ConcurrentHashMap继承了AbstractMap类，同时实现了ConcurrentMap接口。

HashEntry 类

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

/**
     * ConcurrentHashMap list entry. Note that this is never exported
     * out as a user-visible Map.Entry.
     */
    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;
        }

        39     }

在 ConcurrentHashMap 中，在散列时如果产生“碰撞”，将采用“链地址法”来处理“碰撞”：把“碰撞”的 HashEntry 对象链接成一个链表。由于 HashEntry 的 next 域为 final 型，所以新节点只能在链表的表头处插入。下图是在一个空桶中依次插入 A，B，C 三个 HashEntry 对象后的结构图：

Segment 类

Segment 类继承于 ReentrantLock 类，从而使得 Segment 对象能充当锁的角色。每个 Segment 对象用来守护其（成员对象 table 中）包含的若干个桶。

table 是一个由 HashEntry 对象组成的数组。table 数组的每一个数组成员就是散列映射表的一个桶。 count 变量是一个计数器，它表示每个 Segment 对象管理的 table 数组（若干个 HashEntry 组成的链表）包含的 HashEntry 对象的个数。每一个 Segment 对象都有一个 count 对象来表示本 Segment 中包含的 HashEntry 对象的总数。

    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;

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

        transient int count;

        transient int modCount;

        transient int threshold;

        final float loadFactor;

        Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
            this.loadFactor = lf;
            this.threshold = threshold;
            this.table = tab;
        }

        final V put(K key, int hash, V value, boolean onlyIfAbsent) {省略}

        /**
         * Doubles size of table and repacks entries, also adding the
         * given node to new table
         */
        @SuppressWarnings("unchecked")
        private void rehash(HashEntry<K,V> node) {省略}

  
        private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {...}
        private void scanAndLock(Object key, int hash) {...}

        /**
         * Remove; match on key only if value null, else match both.
         */
        final V remove(Object key, int hash, Object value) {...}

        final boolean replace(K key, int hash, V oldValue, V newValue) {...}

        final V replace(K key, int hash, V value) {...}

        final void clear() {...}
    }

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

ConCurrentHashMap的构造方法

构造方法如下：

/**
     * Creates a new, empty map with the specified initial
     * capacity, load factor and concurrency level.
     *
     * @param initialCapacity the initial capacity. The implementation
     * performs internal sizing to accommodate this many elements.
     * @param loadFactor  the load factor threshold, used to control resizing.
     * Resizing may be performed when the average number of elements per
     * bin exceeds this threshold.
     * @param concurrencyLevel the estimated number of concurrently
     * updating threads. The implementation performs internal sizing
     * to try to accommodate this many threads.
     * @throws IllegalArgumentException if the initial capacity is
     * negative or the load factor or concurrencyLevel are
     * nonpositive.
     */
    @SuppressWarnings("unchecked")
    public ConcurrentHashMap(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;
    }

 这个构造函数有三个入参，如果用户不指定则会使用默认值，initialCapacity为16，loadFactor为0.75（负载因子，扩容时需要参考），concurrentLevel为16。现在我对初始化进行debug测试见下图：

 从上图可以知道当走到第31行代码时  31 this.segmentShift = 32 - sshift;  参数变量的值为图上所示，初始化大小 initialCapacity为16，负载 loadFactor为0.75（负载因子，扩容需要参考），并行度concurrencyLevel为16，sshit为4：2的sshit次方等于ssize，ssize为16：这是segment数组的长度，根据concurrencyLevel算出，代码可知ssize一定是大于或者等于concurrencyLevel最小的2次幂，为什么Segment大小一定是2次幂，那是因为通过位与散列算法来定位Segment的index，通过上面的结果计算出segmentShift和segmentMask，即segmentShit是28，为什么32减去4呢，因为hash()方法的输出最大位数就是32位数，segmentMask是散列运算的掩码，这里值是15，

关于segmentShit和segmentMask，

这两个全局变量主要作用就是定位segment，int j =(hash >>> segmentShift) & segmentMask。

       segmentMask：段掩码，假如segments数组长度为16，则段掩码为16-1=15；segments长度为32，段掩码为32-1=31。这样得到的所有bit位都为1，可以更好地保证散列的均匀性

　　segmentShift：2的sshift次方等于ssize，segmentShift=32-sshift。若segments长度为16，segmentShift=32-4=28;若segments长度为32，segmentShift=32-5=27。而计算得出的hash值最大为32位，无符号右移segmentShift，则意味着只保留高几位（其余位是没用的），然后与段掩码segmentMask位运算来定位Segment。

第一步debug就是确定Segment的数组长度为16，并且初始化定位segment的变量。现在进行第二步debug

当运行到42行代码时候，由上图所示确认了经过计算得出HashEntry的数组初始化长度为2，进行第三部debug 

这里初始化了Segment，创建segments数组并初始化第一个Segment，其余的Segment延迟初始化，并且添加第0个元素

在源码中我们可以看出在构造ConCurrentHashMap中主要做3件事情。

1，确认并发度，也就是Segment的数组长度，并保证他是2的n次幂。

2，确认HashEntry数组的初始化长度，并保证他是2的n次幂。

3，将Segment初始化，并只是填充第0个元素。

put()方法

第一步主要是验证传参以及定位segment的index，主要流程是先获取key的hash值，定位segment，调用segment的put方法。这里就需要上文初始化得到的segmentShit和segmentMask的值就是下图的22行代码。

/**
     * Maps the specified key to the specified value in this table.
     * Neither the key nor the value can be null.
     *
     * <p> The value can be retrieved by calling the <tt>get</tt> method
     * with a key that is equal to the original key.
     *
     * @param key key with which the specified value is to be associated
     * @param value value to be associated with the specified key
     * @return the previous value associated with <tt>key</tt>, or
     *         <tt>null</tt> if there was no mapping for <tt>key</tt>
     * @throws NullPointerException if the specified key or value is null
     */
    @SuppressWarnings("unchecked")
    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);
    }

hash方法源码为：也就是上图的19行代码：总的来说就是先拿到key的hashCode，然后对这个值进行再散列。

private int hash(Object k) {
        int h = hashSeed;

        if ((0 != h) && (k instanceof String)) {
            return sun.misc.Hashing.stringHash32((String) k);
        }

        h ^= k.hashCode();

        // Spread bits to regularize both segment and index locations,
        // using variant of single-word Wang/Jenkins hash.
        h += (h <<  15) ^ 0xffffcd7d;
        h ^= (h >>> 10);
        h += (h <<   3);
        h ^= (h >>>  6);
        h += (h <<   2) + (h << 14);
        return h ^ (h >>> 16);
    }

返回第一张图put方法的23行的代码方法源码为：这里主要调用segment的put方法，

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大致过程是：

1. 加锁
2. 定位key在tabl数组上的索引位置index，获取到头结点
3. 判断是否有hash冲突
4. 如果没有冲突直接将新节点node添加到数组index索引位
5. 如果有冲突，先判断是否有相同key
6. 有相同key直接替换对应node的value值
7. 没有添加新元素到链表尾部
8. 解锁

这里需要注意的是scanAndLockForPut方法，他在没有获取到锁的时候不仅会通过自旋获取锁，还会做一些其他的查找或新增节点的工，以此来提升put性能。

在上图代码有一段是调用scanAndLockForPut方法，对应行数第三行，该方法的源码是：

/**
         * Scans for a node containing given key while trying to
         * acquire lock, creating and returning one if not found. Upon
         * return, guarantees that lock is held. UNlike in most
         * methods, calls to method equals are not screened: Since
         * traversal speed doesn't matter, we might as well help warm
         * up the associated code and accesses as well.
         *
         * @return a new node if key not found, else null
         */
        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方法在当前线程获取不到segment锁的情况下，完成查找或新建节点的工作。当获取到锁后直接将该节点加入链表即可，提升了put操作的性能。大致过程：

1. 定位key在HashEntry数组的索引位，并获取第一个节点
2. 尝试获取锁，如果成功直接返回，否则进入自旋
3. 判断是否有hash冲突，没有就直接完成新节点的初始化
4. 有hash冲突，开始遍历链表查找是否有相同key
5. 如果没找到相同key，那么就完成新节点的初始化
6. 如果找到相同key，判断循环次数是否大于最大扫描次数
7. 如果循环次数是否大于最大扫描次数，就直接CAS拿锁（阻塞式）
8. 如果循环次数不大于最大扫描次数，判断头结点是否有变化
9. 进入下次循环。

上面的Segment.rehash() 扩容方法就是segment的put方法第30行代码，Segment.rehash() 的源码是：

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

在这里我们可以发现每次扩容是针对一个单独的Segment的，在扩容完成之前中不会对扩容前的数组进行修改，这样就可以保证get()不被扩容影响。大致过程是：

1. 新建扩容后的数组，容量是原来的两倍
2. 遍历扩容前的数组
3. 通过e.hash & sizeMask;计算key新的索引位
4. 转移数据
5. 将扩容后的数组指向成员变量table

 现在进行put方法的debug测试：

第一步，传值put(2,"ww")，计算hash值，定位要保存的segment数组，

第二步，进入segment的put方法，值写入第7个segment数组中。在这一步需要执行的东西较多，但是比较重要的是，在写入前创建HashEentry节点初始化然后加锁，添加然后解锁。

 get()方法

我们可以看到get方法是没有加锁的，因为HashEntry的value和next属性是volatile的，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);
    // 计算出Segment的索引位
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    // 以原子的方式获取Segment
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
        (tab = s.table) != null) {
        // 原子方式获取HashEntry
        for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
             e != null; e = e.next) {
            K k;
            // key相同
            if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                // value是volatile所以可以不加锁直接取值返回
                return e.value;
        }
    }
    return null;
}

debug测试get(2)，原来就存在key==2的value，看下图，先计算hash值，然后计算segment索引位置，然后判断是否存在segment和hashTable然后循环对比key值，返回value，

size()方法

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;
    int size;
    // true表示size溢出32位（大于Integer.MAX_VALUE）
    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 (;;) {
            // retries 如果retries等于2则对所有Segment加锁
            if (retries++ == RETRIES_BEFORE_LOCK) {
                for (int j = 0; j < segments.length; ++j)
                    ensureSegment(j).lock(); // force creation
            }
            sum = 0L;
            size = 0;
            overflow = false;
            // 统计每个Segment元素个数
            for (int j = 0; j < segments.length; ++j) {
                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;
}

size的核心思想是先进性两次不加锁统计，如果两次的值一样则直接返回，否则第三个统计的时候会将所有segment全部锁定，再进行size统计，所以size()尽量少用。因为这是在并发情况下，size其他线程也会改变size大小，所以size()的返回值只能表示当前线程、当时的一个状态，可以算其实是一个预估值。

debug进行测试：for体现了除非对比第二次执行break才可以跳出循环，

isEmpty() 方法

public boolean isEmpty() {
    long sum = 0L;
    final Segment<K,V>[] segments = this.segments;
    for (int j = 0; j < segments.length; ++j) {
        Segment<K,V> seg = segmentAt(segments, j);
        if (seg != null) {
            // 只要有一个Segment的元素个数不为0则表示不为null
            if (seg.count != 0)
                return false;
            // 统计操作总数
            sum += seg.modCount;
        }
    }
    if (sum != 0L) { // recheck unless no modifications
        for (int j = 0; j < segments.length; ++j) {
            Segment<K,V> seg = segmentAt(segments, j);
            if (seg != null) {
                if (seg.count != 0)
                    return false;
                sum -= seg.modCount;
            }
        }
        // 说明在统计过程中ConcurrentHashMap又被操作过，
        // 因为上面判断了ConcurrentHashMap不可能会有元素，所以这里如果有操作一定是新增节点
        if (sum != 0L)
            return false;
    }
    return true;
}

1. 先判断Segment里面是否有元素，如果有直接返回，如果没有则统计操作总数；
2. 为了保证在统计过程中ConcurrentHashMap里面的元素没有发生变化，再对所有的Segment的操作数做了统计；
3. 最后 sum==0 表示ConcurrentHashMap里面确实没有元素返回true，否则一定进行过新增元素返回false。

和size方法一样这个方法也是一个若一致方法，最后的结果也是一个预估值。

总结

ConcurrentHashMap 拥有更高的并发性。在 HashTable 和由同步包装器包装的 HashMap 中，使用一个全局的锁来同步不同线程间的并发访问。同一时间点，只能有一个线程持有锁，也就是说在同一时间点，只能有一个线程能访问容器。这虽然保证多线程间的安全并发访问，但同时也导致对容器的访问变成串行化的了。

ConcurrentHashMap 的高并发性主要来自于三个方面：

用分离锁(Segment锁)实现多个线程间的更深层次的共享访问；

用 HashEntery 对象的不变性来降低执行读操作的线程在遍历链表期间对加锁的需求;

通过对同一个 Volatile 变量的写 / 读访问，协调不同线程间读 / 写操作的内存可见性；

参考链接：

https://my.oschina.net/xiaolyuh/blog/3080609

http://xiaobaoqiu.github.io/blog/2014/12/25/concurrenthashmap/