ConcurrentHashMap 1.8原理解析

JDK 1.8中，Hash家族有这么一些存在，HashMap,HashTable,LinkedHashMap,ConcurrentHashMap。这里面支持线程安全的有HashTable以及ConcurrentHashMap。对Hash有一个基本了解能够参考本人的从Hash到一致性Hash原理(深度好文) 。

那既然说到ConcurrentHashMap，天然要讨论的就是它的线程安全性和效能。咱们先来看一下HashTable的线程安全性以及效能的低下。

public synchronized V put(K key, V value) {
    // Make sure the value is not null
    if (value == null) {
        throw new NullPointerException();
    }

    // Makes sure the key is not already in the hashtable.
    Entry<?,?> tab[] = table;
    int hash = key.hashCode();
    int index = (hash & 0x7FFFFFFF) % tab.length;
    @SuppressWarnings("unchecked")
    Entry<K,V> entry = (Entry<K,V>)tab[index];
    for(; entry != null ; entry = entry.next) {
        if ((entry.hash == hash) && entry.key.equals(key)) {
            V old = entry.value;
            entry.value = value;
            return old;
        }
    }

    addEntry(hash, key, value, index);
    return null;
}

public synchronized V get(Object key) {
    Entry<?,?> tab[] = table;
    int hash = key.hashCode();
    int index = (hash & 0x7FFFFFFF) % tab.length;
    for (Entry<?,?> e = tab[index] ; e != null ; e = e.next) {
        if ((e.hash == hash) && e.key.equals(key)) {
            return (V)e.value;
        }
    }
    return null;
}

咱们对比一下HashMap 1.7的这两个方法(由于1.8点HashMap会生成红黑树，咱们暂时先不考虑红黑树的问题)

public V put(K key, V value) {
        if (key == null)
            return putForNullKey(value);
        int hash = hash(key);
        int i = indexFor(hash, table.length);
        for (Entry<K,V> e = table[i]; e != null; e = e.next) {
            Object k;
            if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {
                V oldValue = e.value;
                e.value = value;
                e.recordAccess(this);
                return oldValue;
            }
        }

        modCount++;
        addEntry(hash, key, value, i);
        return null;
    }

public V get(Object key) {
        if (key == null)
            return getForNullKey();
        Entry<K,V> entry = getEntry(key);

        return null == entry ? null : entry.getValue();
    }

final int hash(Object k) {
        int h = 0;
        if (useAltHashing) {
            if (k instanceof String) {
                return sun.misc.Hashing.stringHash32((String) k);
            }
            h = hashSeed;
        }

        h ^= k.hashCode();

        // This function ensures that hashCodes that differ only by
        // constant multiples at each bit position have a bounded
        // number of collisions (approximately 8 at default load factor).
        //一种算法，进行4次位移，获得相对比较分散的链表
        h ^= (h >>> 20) ^ (h >>> 12);
        return h ^ (h >>> 7) ^ (h >>> 4);
    }

在这里咱们能够看到，他们除了key的hash算法不一样，HashTable的比较简单，只是用key的哈希值与0x7FFFFFFF(十进制2147483647，二进制1111111111111111111111111111111)进行一次与运算，即为只要相同二进制位上不论0，1所有都变成1，再对table数组的长度取模。而HashMap 1.7则为key的哈希值进行各位无符号位移加异或运算，取得最终哈希值。而后是HashTable不接收null的key,而HashMap接受。他们最大的区别就在于synchronized显示器锁了。

synchronized的本质是全部对象的字节码中有一个monitor的对象头，任何线程拿到了这个monitor的对象头就能够对这个对象进行操做，而拿不到monitor对象头的线程就只能等待，直到拿到了monitor的线程放弃，其余线程才能争夺这个对象头来对对象进行操做。那么问题来了，当大量线程高并发的时候，只要有一个线程拿到了这个对象头，其余线程对这个对象是既不能读也不能写。而对于HashMap来讲，若是多线程对其进行操做，那么任意线程均可以胡乱修改里面值的内容形成脏读，因此HashMap是线程不安全的。

那么咱们今天的主角登场了ConcurrentHashMap。咱们一样来看一下这两个方法。如下是1.8的源码，首先咱们要清楚的是1.8跟1.7已经彻底不一样，1.7是使用segements(16个segement)，每一个segement都有一个table(Map.Entry数组)，至关于16个HashMap，同步机制为分段锁，每一个segment继承ReentrantLock；而1.8只有1个table(Map.Entry数组)，同步机制为CAS + synchronized保证并发更新。若是不搞清楚这个问题，那么你看1.8的源码可能会很懵逼。

public V put(K key, V value) {
    return putVal(key, value, false);
}

final V putVal(K key, V value, boolean onlyIfAbsent) {
    //不管key仍是value,不容许空
    if (key == null || value == null) throw new NullPointerException();
    //此处获取hash值的方法与HashTable相似
    int hash = spread(key.hashCode());
    int binCount = 0;
    //无限循环
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //若是节点数组为null，或者长度为0，初始化节点数组
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        //若是节点数组的某个节点为null，则put的时候就会采用无锁竞争来获取该节点的头把交椅
        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 { //若是hash冲突的时候，即多线程操做时，你们都有同样的hash值
            V oldVal = null;
            synchronized (f) { //锁定节点数组的该节点
                if (tabAt(tab, i) == f) {
                    //若是当前该节点为链表形态
                    if (fh >= 0) {
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            //找链表中找到相同的key，把新value替代老value
                            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;
                            //若是找不到key，就添加到链表到末尾
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);
                                break;
                            }
                        }
                    }
                    //若是当前为红黑树形态，进行红黑树到查找和替代(存在相同的key)，或者放入红黑树到新叶节点上(key不存在)
                    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) {
                //若是链表长度超过了8，链表转红黑树
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //统计节点个数，检查是否须要扩容
    addCount(1L, binCount);
    return 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());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

读取的时候，咱们没有看见锁到存在，说明读不受多线程影响。

对比ConcurrentHashMap和HashTable,咱们能够明显的看到，ConcurrentHashMap在写的时候，并无锁住整个节点数组，在新节点上使用的是无锁竞争，在老节点上锁住的仅仅是一个节点，读的时候若是不是刚好读到写线程写入相同Hash值的位置，不受影响(能够认为咱们的操做通常是读多写少，这种概率也比较低)。而HashTable是对整个节点数组进行锁定，读到时候不能写，写的时候不能读，这么一对比就能够明显感受到性能差距是巨大的。

虽然ConcurrentHashMap的并发性能还算比较优异，但在亿级计算中，却依然会成为性能瓶颈，具体能够参考本人的Fork/Join框架原理和使用探秘

至于这里为何会慢，我认为在这种超高并发下，节点数组的单节点的的写写竞争是互斥的，其次，因为红黑树具备读快写慢的特性，它要不断保持树的平衡而不断返转，因此才会使得高并发写的性能急剧降低。