【死磕 Java 集合】— ConcurrentHashMap源码分析(三)

撸了今年阿里、腾讯和美团的面试,我有一个重要发现…….

作者:彤哥

出处:https://www.cnblogs.com/tong-yuan/


本章接着上两章,链接直达:

死磕 java集合之ConcurrentHashMap源码分析(一)

死磕 java集合之ConcurrentHashMap源码分析(二)


删除元素

删除元素跟添加元素一样,都是先找到元素所在的桶,然后采用分段锁的思想锁住整个桶,再进行操作。

public V remove(Object key) {
    // 调用替换节点方法
    return replaceNode(key, null, null);
}

final V replaceNode(Object key, V value, Object cv) {
    // 计算hash
    int hash = spread(key.hashCode());
    // 自旋
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0 ||
                (f = tabAt(tab, i = (n - 1) & hash)) == null)
            // 如果目标key所在的桶不存在,跳出循环返回null
            break;
        else if ((fh = f.hash) == MOVED)
            // 如果正在扩容中,协助扩容
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            // 标记是否处理过
            boolean validated = false;
            synchronized (f) {
                // 再次验证当前桶第一个元素是否被修改过
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) {
                        // fh>=0表示是链表节点
                        validated = true;
                        // 遍历链表寻找目标节点
                        for (Node<K,V> e = f, pred = null;;) {
                            K ek;
                            if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                            (ek != null && key.equals(ek)))) {
                                // 找到了目标节点
                                V ev = e.val;
                                // 检查目标节点旧value是否等于cv
                                if (cv == null || cv == ev ||
                                        (ev != null && cv.equals(ev))) {
                                    oldVal = ev;
                                    if (value != null)
                                        // 如果value不为空则替换旧值
                                        e.val = value;
                                    else if (pred != null)
                                        // 如果前置节点不为空
                                        // 删除当前节点
                                        pred.next = e.next;
                                    else
                                        // 如果前置节点为空
                                        // 说明是桶中第一个元素,删除之
                                        setTabAt(tab, i, e.next);
                                }
                                break;
                            }
                            pred = e;
                            // 遍历到链表尾部还没找到元素,跳出循环
                            if ((e = e.next) == null)
                                break;
                        }
                    }
                    else if (f instanceof TreeBin) {
                        // 如果是树节点
                        validated = true;
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        TreeNode<K,V> r, p;
                        // 遍历树找到了目标节点
                        if ((r = t.root) != null &&
                                (p = r.findTreeNode(hash, key, null)) != null) {
                            V pv = p.val;
                            // 检查目标节点旧value是否等于cv
                            if (cv == null || cv == pv ||
                                    (pv != null && cv.equals(pv))) {
                                oldVal = pv;
                                if (value != null)
                                    // 如果value不为空则替换旧值
                                    p.val = value;
                                else if (t.removeTreeNode(p))
                                    // 如果value为空则删除元素
                                    // 如果删除后树的元素个数较少则退化成链表
                                    // t.removeTreeNode(p)这个方法返回true表示删除节点后树的元素个数较少
                                    setTabAt(tab, i, untreeify(t.first));
                            }
                        }
                    }
                }
            }
            // 如果处理过,不管有没有找到元素都返回
            if (validated) {
                // 如果找到了元素,返回其旧值
                if (oldVal != null) {
                    // 如果要替换的值为空,元素个数减1
                    if (value == null)
                        addCount(-1L, -1);
                    return oldVal;
                }
                break;
            }
        }
    }
    // 没找到元素返回空
    return null;
}

(1)计算hash;

(2)如果所在的桶不存在,表示没有找到目标元素,返回;

(3)如果正在扩容,则协助扩容完成后再进行删除操作;

(4)如果是以链表形式存储的,则遍历整个链表查找元素,找到之后再删除;

(5)如果是以树形式存储的,则遍历树查找元素,找到之后再删除;

(6)如果是以树形式存储的,删除元素之后树较小,则退化成链表;

(7)如果确实删除了元素,则整个map元素个数减1,并返回旧值;

(8)如果没有删除元素,则返回null;

获取元素

获取元素,根据目标key所在桶的第一个元素的不同采用不同的方式获取元素,关键点在于find()方法的重写。

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 计算hash
    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)
            // hash小于0,说明是树或者正在扩容
            // 使用find寻找元素,find的寻找方式依据Node的不同子类有不同的实现方式
            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;
}

(1)hash到元素所在的桶;

(2)如果桶中第一个元素就是该找的元素,直接返回;

(3)如果是树或者正在迁移元素,则调用各自Node子类的find()方法寻找元素;

(4)如果是链表,遍历整个链表寻找元素;

(5)获取元素没有加锁;

获取元素个数

元素个数的存储也是采用分段的思想,获取元素个数时需要把所有段加起来。

public int size() {
    // 调用sumCount()计算元素个数
    long n = sumCount();
    return ((n < 0L) ? 0 :
            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                    (int)n);
}

final long sumCount() {
    // 计算CounterCell所有段及baseCount的数量之和
    CounterCell[] as = counterCells; CounterCell a;
    long sum = baseCount;
    if (as != null) {
        for (int i = 0; i < as.length; ++i) {
            if ((a = as[i]) != null)
                sum += a.value;
        }
    }
    return sum;
}

(1)元素的个数依据不同的线程存在在不同的段里;(见addCounter()分析)

(2)计算CounterCell所有段及baseCount的数量之和;

(3)获取元素个数没有加锁;

总结

(1)ConcurrentHashMap是HashMap的线程安全版本;

(2)ConcurrentHashMap采用(数组 + 链表 + 红黑树)的结构存储元素;

(3)ConcurrentHashMap相比于同样线程安全的HashTable,效率要高很多;

(4)ConcurrentHashMap采用的锁有 synchronized,CAS,自旋锁,分段锁,volatile等;

(5)ConcurrentHashMap中没有threshold和loadFactor这两个字段,而是采用sizeCtl来控制;

(6)sizeCtl = -1,表示正在进行初始化;

(7)sizeCtl = 0,默认值,表示后续在真正初始化的时候使用默认容量;

(8)sizeCtl > 0,在初始化之前存储的是传入的容量,在初始化或扩容后存储的是下一次的扩容门槛;

(9)sizeCtl = (resizeStamp << 16) + (1 + nThreads),表示正在进行扩容,高位存储扩容邮戳,低位存储扩容线程数加1;

(10)更新操作时如果正在进行扩容,当前线程协助扩容;

(11)更新操作会采用synchronized锁住当前桶的第一个元素,这是分段锁的思想;

(12)整个扩容过程都是通过CAS控制sizeCtl这个字段来进行的,这很关键;

(13)迁移完元素的桶会放置一个ForwardingNode节点,以标识该桶迁移完毕;

(14)元素个数的存储也是采用的分段思想,类似于LongAdder的实现;

(15)元素个数的更新会把不同的线程hash到不同的段上,减少资源争用;

(16)元素个数的更新如果还是出现多个线程同时更新一个段,则会扩容段(CounterCell);

(17)获取元素个数是把所有的段(包括baseCount和CounterCell)相加起来得到的;

(18)查询操作是不会加锁的,所以ConcurrentHashMap不是强一致性的;

(19)ConcurrentHashMap中不能存储key或value为null的元素;

彩蛋——值得学习的技术

ConcurrentHashMap中有哪些值得学习的技术呢?

我认为有以下几点:

(1)CAS + 自旋,乐观锁的思想,减少线程上下文切换的时间;

(2)分段锁的思想,减少同一把锁争用带来的低效问题;

(3)CounterCell,分段存储元素个数,减少多线程同时更新一个字段带来的低效;

(4)@sun.misc.Contended(CounterCell上的注解),避免伪共享;(p.s.伪共享我们后面也会讲的^^)

(5)多线程协同进行扩容;

(6)你又学到了哪些呢?

彩蛋——不能解决的问题

ConcurrentHashMap不能解决什么问题呢?

请看下面的例子:

private static final Map<Integer, Integer> map = new ConcurrentHashMap<>();

public void unsafeUpdate(Integer key, Integer value) {
    Integer oldValue = map.get(key);
    if (oldValue == null) {
        map.put(key, value);
    }
}

这里如果有多个线程同时调用unsafeUpdate()这个方法,ConcurrentHashMap还能保证线程安全吗?

答案是不能。因为get()之后if之前可能有其它线程已经put()了这个元素,这时候再put()就把那个线程put()的元素覆盖了。

那怎么修改呢?

答案也很简单,使用putIfAbsent()方法,它会保证元素不存在时才插入元素,如下:

public void safeUpdate(Integer key, Integer value) {
    map.putIfAbsent(key, value);
}

那么,如果上面oldValue不是跟null比较,而是跟一个特定的值比如1进行比较怎么办?也就是下面这样:

public void unsafeUpdate(Integer key, Integer value) {
    Integer oldValue = map.get(key);
    if (oldValue == 1) {
        map.put(key, value);
    }
}

这样的话就没办法使用putIfAbsent()方法了。

其实,ConcurrentHashMap还提供了另一个方法叫replace(K key, V oldValue, V newValue)可以解决这个问题。

replace(K key, V oldValue, V newValue)这个方法可不能乱用,如果传入的newValue是null,则会删除元素。

public void safeUpdate(Integer key, Integer value) {
    map.replace(key, 1, value);
}

那么,如果if之后不是简单的put()操作,而是还有其它业务操作,之后才是put(),比如下面这样,这该怎么办呢?

public void unsafeUpdate(Integer key, Integer value) {
    Integer oldValue = map.get(key);
    if (oldValue == 1) {
        System.out.println(System.currentTimeMillis());
        /**
         * 其它业务操作
         */
        System.out.println(System.currentTimeMillis());

        map.put(key, value);
    }
}

这时候就没办法使用ConcurrentHashMap提供的方法了,只能业务自己来保证线程安全了,比如下面这样:

public void safeUpdate(Integer key, Integer value) {
    synchronized (map) {
        Integer oldValue = map.get(key);
        if (oldValue == null) {
            System.out.println(System.currentTimeMillis());
            /**
             * 其它业务操作
             */
            System.out.println(System.currentTimeMillis());

            map.put(key, value);
        }
    }
}

这样虽然不太友好,但是最起码能保证业务逻辑是正确的。

当然,这里使用ConcurrentHashMap的意义也就不大了,可以换成普通的HashMap了。

上面只是举一个简单的例子,我们不能听说ConcurrentHashMap是线程安全的,就认为它无论什么情况下都是线程安全的,还是那句话尽信书不如无书。

这也正是我们读源码的目的之一,了解其本质,才能在我们的实际工作中少挖坑,不论是挖给别人还是挖给自己^^。


好了,整个ConcurrentHashMap就讲完了。

文章暂不支持留言功能,如果您有任何建议或意见请在公众号后台给我留言,留言必回复。

喜欢这篇关于ConcurrentHashMap讲解的,赏个鸡腿呗~~

赞(7) 打赏

如未加特殊说明,此网站文章均为原创,转载必须注明出处。Java 技术驿站 » 【死磕 Java 集合】— ConcurrentHashMap源码分析(三)
分享到: 更多 (0)

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

关注【Java 技术驿站】公众号,每天早上 8:10 为你推送一篇技术文章

扫描二维码关注我!


关注【Java 技术驿站】公众号 回复 “VIP”,获取 VIP 地址永久关闭弹出窗口

免费获取资源

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏