乐趣区

ConcurrentHashMap

jdk1.7 的 ConcurrentHashMap

jdk1.7 采用的是 Segment+HashEntry 数组 + 链表来实现的

ConcurrentHashMap 初始化时有 16 个 Segment,每个 Segment 中又包含了 HashEntry 数组,每个 HashEntry 又是一个链表,可以看出,Segment 的结构和 HashMap 类似,你可以这样理解:ConcurrentHashMap 由多个 HashMap 组成。

线程安全的实现原理

ConcurrentHashMap 中的 Segment 在实现上继承了 ReentrantLock,这样就实现了分段锁的功能。下面我们通过 put()方法来看看它是如何保证线程安全的。

public V put(K key, V value) {
    Segment<K,V> s;
    // value 不能为 null
    if (value == null)
        throw new NullPointerException();
    // 计算 key 的 hash 值
    int hash = hash(key);
    // 定位到哪个 Segment
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment<K,V>)UNSAFE.getObject          
         (segments, (j << SSHIFT) + SBASE)) == null) 
        // 如果此 Segment 为 null,则初始化 Segment
        s = ensureSegment(j);
    // 往 Segment 中 put 值
    return s.put(key, hash, value, false);
}
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    // 尝试获取锁,如果获取不到,则进入 scanAndLockForPut 方法
    // scanAndLockForPut 中会再次 tryLock,如果还是获取不到,则 lock()挂起线程
    HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        // table 通过 volatile 修饰,保证拿到的都是最新的 table
        HashEntry<K,V>[] tab = table;
        // 得到数组下标,& 取模作用,详细介绍可以看我写的 jdk1.8HashMap 文章
        int index = (tab.length - 1) & hash;
        // 得到此下标下的 HashEntry
        HashEntry<K,V> first = entryAt(tab, index);
        // 遍历链表
        for (HashEntry<K,V> e = first;;) {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;
                        // 修改次数 +1,用于限制遍历中删除
                        ++modCount;
                    }
                    break;
                }
                // 继续遍历下个值
                e = e.next;
            }
            else {if (node != null)
                    // 头插法(头插法的问题看我写的 jdk1.7HashMap 文章)node.setNext(first);
                else
                    // new 新的 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;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        // 释放锁
        unlock();}
    return oldValue;
}

通过上面源码可以看出 jdk1.7 的 ConcurrentHashMap 通过 Segment 的 ReentrantLock 锁来实现线程安全。要注意的是上锁操作是this.lock(),也就是每个 Segment 的锁是分开的,其中一个上锁并不影响其他 Segment 操作,说明可以同时 16 个线程同时进来对不同 Segment 操作,这样可以有效的提高并发度。

这种 Segment+ReentrantLock 看起来可以很好的解决并发问题了,那为什么 jdk1.8 又不继续使用此方法了呢?下面我们看看 jdk1.8 的 ConcurrentHashMap

jdk1.8 的 ConcurrentHashMap

jdk1.8 采用的是 Node 数组 + 链表 / 红黑树实现

jdk1.8 的 ConcurrentHashMap 去除了外层的 Segment,直接使用数组加链表或红黑树,数据结构和 1.8 的 HashMap 一样,不同之处在于使用了 CAS+synchronized 保证了线程安全。

线程安全的实现原理

我们还是通过 put()方法看看它是如何保证线程安全的

public V put(K key, V value) {return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
    // 同样 value 不能为 null
    if (key == null || value == null) throw new NullPointerException();
    // 新的 hash 算法,和 1.7 的不同
    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)
            // 如果 table 为空,初始化
            tab = initTable();
        // f= 此下标的头节点,i= 下标
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 如果此下标节点为空,new 新节点放在此节点下
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        // fh= 头节点的 hash 值
        else if ((fh = f.hash) == MOVED)
            // 协助扩容
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            // 重点来了,synchronized 锁当前 node 节点
            synchronized (f) {if (tabAt(tab, i) == f) {
                    // 头节点 hash 值大于 0,说明是链表
                    if (fh >= 0) {
                        // 链表长度
                        binCount = 1;
                        // 遍历链表
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            // key 相同,覆盖原值
                            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;
                            // 到链表末端,new 新 node 挂在末端
                            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) {
                // 如果链表长度超过阈值,转红黑树
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}

通过上面源码可以看出,1.8 的 ConcurrentHashMap 采用的是 synchronized 来实现线程安全。synchronized (f)可以看出锁的是每一个 Node 对象,这样可以将锁细化,除非有多个线程同时操作同一个 Node,否则不会出现争抢锁,这样锁也不会升级,就算出现争抢锁,也可以通过自旋来获取锁,达到一定次数才会升级为重量级锁,这种情况是很少的。那为什么 1.7 的 ReentrantLock 不细化到 Node 上呢?

如果 1.7 的 ReentrantLock 细化到 Node 上,它只会在线程没有抢到锁的时候,新建节点再尝试一次,如果还没抢到,不会自旋,而是直接挂起,这样就增加了线程上下文开销的代价。当然,ReentrantLock 的 tryLock()也是可以自旋的,但是确定不了 tryLock 的时间。

所以,在锁细化到如此程度的情况下,使用 synchronized 是最好的选择。


<center> 扫一扫,关注我 </center>

退出移动版