ConcurrentHashMap分析

2017-08-11

1. JDK 1.7版本

1.1 介绍

1.1.1 作用

ConcurrentHashMap是JUC下的一个实现类，解决高并发下HashMapd的线程安全问题，并且相对于Hashtable来说，效率更高，因为内部使用了segment分段锁，使多个线程能够同时操作，而不会引发线程安全的问题，实现方面用到了无锁的CAS（ReentrantLock）。

1.1.2 类结构分析

1.2 原理分析

1.3 源码分析

1.3.1 put

① 确定对应的segment
② 对segment加锁，获取segment中的table
③ 通过对key进行hash得到hash值，以此获取对应的table元素（HashEntry链表）
④ 如果key相同，则替换旧值，不同则新增一个HashEntry并确认是否需要扩容
⑤ 解锁

public V put(K key, V value) {
        Segment<K,V> s;
        if (value == null)//value不允许为null，HashMap则key和value都可以是null（bucket为0）
            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);
    }
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;
                //以此获取对应的table元素（HashEntry链表），hash是对key进行hash运算
                int index = (tab.length - 1) & hash;
                HashEntry<K,V> first = entryAt(tab, index);
                //遍历HashEntry链表
                for (HashEntry<K,V> e = first;;) {
                    //HashEntry不为null
                    if (e != null) {
                        K k;
                        //确认key是否相同，相同则替代旧值
                        if ((k = e.key) == key ||
                            (e.hash == hash && key.equals(k))) {
                            oldValue = e.value;
                            if (!onlyIfAbsent) {
                                e.value = value;
                                //修改modCount
                                ++modCount;
                            }
                            break;
                        }
                        e = e.next;
                    }
                    else {
                        if (node != null)
                            node.setNext(first);
                        else
                            //新增一个HashEntry对象
                            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
                        ++modCount;
                        count = c;
                        oldValue = null;
                        break;
                    }
                }
            } finally {
                unlock();
            }
            return oldValue;
        }

1.3.2 get

① 定位segment
② 定位HashEntry[]
③ 遍历HashEntry[]，并获取key相同的值

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;
        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);
                 e != null; e = e.next) {
                K k;
                if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                    return e.value;
            }
        }
        return null;
    }

1.3.3 remove

jdk1.6需要两个链表来实现remove，而jdk1.7则不需要

final V remove(Object key, int hash, Object value) {
            if (!tryLock())
                scanAndLock(key, hash);
            V oldValue = null;
            try {
                HashEntry<K,V>[] tab = table;
                int index = (tab.length - 1) & hash;
                //获取到HashEntry[]
                HashEntry<K,V> e = entryAt(tab, index);
                HashEntry<K,V> pred = null;
                while (e != null) {
                    K k;
                    //指向下一个元素：遍历作用
                    HashEntry<K,V> next = e.next;
                    if ((k = e.key) == key ||
                        (e.hash == hash && key.equals(k))) {
                        V v = e.value;
                        if (value == null || value == v || value.equals(v)) {
                            if (pred == null)
                                //当删除节点是链表头，更新数组项的值
                                setEntryAt(tab, index, next);
                            else
                                //当删除节点是非链表头，则通过设置指针来移除
                                pred.setNext(next);
                            ++modCount;
                            --count;
                            oldValue = v;
                        }
                        break;
                    }
                    pred = e;
                    e = next;
                }
            } finally {
                unlock();
            }
            return oldValue;
        }

1.3.4 reshah扩容

① 重新新建一个ConcurrentHashMap的数据结构，将原有的值全部负责到新的数据结构中
② 扩容后容量为原来容量的两倍

private void rehash(HashEntry<K,V> node) {
            HashEntry<K,V>[] oldTable = table;
            int oldCapacity = oldTable.length;
            int newCapacity = oldCapacity << 1;//扩容后的容量为原来容量的两倍
            threshold = (int)(newCapacity * loadFactor);
            HashEntry<K,V>[] newTable =
                (HashEntry<K,V>[]) new HashEntry[newCapacity];
            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;
                             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;
        }

1.3.5 size

size()方法首先会进行两次尝试(不加锁)，两次结果的modCount都一致的话，则表示此过程中的ConcurrentHashMap并未改变，返回各个segment的count之和，即是size的值；假若modCount不一致，则对每个segment依次加锁，获取其count值，最终才是真正的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;
        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：初始值是-1
        try {
            for (;;) {
                // 此时 判断 retries++（为0）是否等于 RETRIES_BEFORE_LOCK（2）
                // 由此可见，头两次是不会进入循环体，进行加锁操作的，第三次才会加锁
                if (retries++ == RETRIES_BEFORE_LOCK) {
                    //对每个segment进行加锁
                    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<K,V> seg = segmentAt(segments, j);
                    if (seg != null) {
                        //获取modCount值
                        sum += seg.modCount;
                        //获取这个segment中元素个数
                        int c = seg.count;
                        // size是多个segment的count（元素个数）累加
                        if (c < 0 || (size += c) < 0)
                            //溢出
                            overflow = true;
                    }
                }
                //判断两次得到的元素个数是否相同，相同则退出，返回size的值
                //不相同（last已经保存了第一次sum的值）则数据元素改变了，要对每个segment加锁
                if (sum == last)
                    break;
                last = sum;
            }
        } finally {
            if (retries > RETRIES_BEFORE_LOCK) {
                for (int j = 0; j < segments.length; ++j)
                    segmentAt(segments, j).unlock();
            }
        }
        //不溢出则返回size值
        return overflow ? Integer.MAX_VALUE : size;
    }

2. JDK 1.8 版本浅析

2.1 介绍

1.8版本的ConcurrentHashMap做了很大的改动：
① 结构就变成了“数组+链表+红黑树”的形式
② 操作时，不再锁整个segment，而是对节点node加锁
③ 加锁采用了synchronized，而不是1.7版本中的ReentrantLock
其中，当table[i]的所代表的链表长度大于8时，就会将结构由链表改变为红黑树，此时查询时的时间复杂度由原来的O(n)变成了O(logN)，提高了查询效率。

2.2 类结构分析

2.3 原理图分析

2.4 源码分析

2.4.1 put()

① 得到table[i]
② 对某节点加锁(如果key对应位置是空时，不用加锁，用CAS)
③ 首节点
3.1 首节点是链表：key存在的话，则设置value；否则新建node节点，加入链表
3.2 首节点是红黑树：则用红黑树的方式设置值
④ bitCount >= 8时，将链表转换为红黑树的结构形式

/** Implementation for put and putIfAbsent */
    final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            //如果这个位置没有值 ，直接放进去，不需要加锁
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
                V oldVal = null;
                //② 对某节点加锁
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                    //④ bitCount >= 8时，将链表转换为红黑树的结构形式
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }