2.8 ConcurrentHashMap

JDK1.5起，java.util.concurrent包

1 JDK1.7

1.1 总体介绍

ConcurrentHashMap采用了分段锁技术，效率较Hashtable高，不允许空值null。

ConcurrentHashMap是 Segment 数组。Segment 继承于ReentrantLock，可以充当锁。

1.2 方法剖析

1）put

虽然HashEntry中的value是用volatile关键词修饰的，但是并不能保证并发的原子性，所以 put 操作时仍然需要加锁处理。

1. 通过key定位到Segment
2. 再调用Segment的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;
    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);
}

3.1 尝试获取锁，如果失败则利用 scanAndLockForPut() 自旋获取锁
3.2 如自旋重试次数达到上限，则改为阻塞获取锁，保证能获取成功
3.3 在Segment中存值

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    //加锁，锁定的是Segment
    HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        ...//类似hashmap存储
    } finally {
        unlock();
    }
    return oldValue;
}

2）get

由于HashEntry中的value属性是用volatile关键词修饰的，保证了内存可见性，所以每次获取时都是最新值。

get 不需要加锁。

2 JDK1.8

2.1 总体介绍

类似HashMap，遍历链表效率太低，1.8 加了红黑树。

Segment抛弃了原有的基于ReentrantLock的分段锁，而采用了 CAS（空桶） + synchronized（非空） 来保证并发安全性。

2.2 方法剖析

1）put

1. 根据 key 计算出 hashcode
2. 判断初始化
3. 定位 Node f
4. 如果 f 为空，则利用 CAS 尝试写入，失败则自旋保证成功
5. 判断扩容
6. 如都不满足，则利用 synchronized 锁写入
7. 如链表大于阈值则要转换为红黑树

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());//根据 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) {//定位 Node f 
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))//利用 CAS 尝试写入，失败则自旋保证成功
                break;                   // no lock when adding to empty bin
        }
        else if ((fh = f.hash) == MOVED)//判断扩容
            tab = helpTransfer(tab, f);
        else {//如果都不满足
            V oldVal = null;
            synchronized (f) {//则利用 synchronized 锁
                ...//写入数据，尾插法
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)//如链表大于阈值则要转换为红黑树
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}

2）get

1. 根据计算出来的 hashcode 寻址，如果就在桶上那么直接返回值。
2. 如果是红黑树那就按照树的方式获取值。
3. 就不满足那就按照链表的方式遍历获取值。

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