专栏首页晏霖深入理解ConcurrentHashMap

深入理解ConcurrentHashMap

前言 本文的分析的源码是JDK8的版本,上一节我们介绍了HashMap,提到了多线程避免扩容时出现死循环,时要使用ConcurrentHashMap,下面我来讲解一下ConcurrentHashMap这个东西简单的基本原理是什么,我们为什么用。

正文 如何保证HashMap的线程安全呢?

方法有很多,比如使用HashTable或者Collections.synchronizedMap,但是这两位选手都有一个共同的问题:性能。因为不管是读还是写操作,他们都会给整个集合上锁,导致同一时间的其他操作被阻塞。并且HashTable和Collections.synchronizedMap解决了HashMap的线程不安全的问题(言外之意HashTable和Collections.synchronizedMap线程安全的),但是带来了运行效率不佳的问题。

基于以上所述,兼顾了线程安全和运行效率的ConcurrentHashMap就出现了,ConcurrentHashMap利用CAS算法。它沿用了与它同时期的HashMap版本的思想,底层依然由“数组”+链表+红黑树的方式思想。

下面我贴了很多源码,其实重要的目的是方便大家阅读,我只做了一个体力活,给大家翻译成了汉语。

ConcurrentHashMap默认配置参数

    /* ---------------- Constants -------------- */
    //最大容量
    private static final int MAXIMUM_CAPACITY = 1 << 30;
    //默认容量
    private static final int DEFAULT_CAPACITY = 16;
    //array的大小,只用于toarray方法
    static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
    //根据这个数来计算segment的个数,segment的个数是仅小于这个数且是2的几次方的一个数(ssize)
    private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    //加载因子
    private static final float LOAD_FACTOR = 0.75f;
    //变成红黑树的链表节点数,超过8就由链表转换成红黑树
    static final int TREEIFY_THRESHOLD = 8;
    //当树节点小于6自动转换成链表
    static final int UNTREEIFY_THRESHOLD = 6;
    // 在转变成树之前,还会有一次判断,只有键值对数量大于 64 才会发生转换。这是为了避免在哈希表建立初期,多个键值对恰好被放入了同一个链表中而导致不必要的转化。
    static final int MIN_TREEIFY_CAPACITY = 64;
    //每次进行转移的最小值这个值作为一个下限来避免Rsisize遇到过多的内存争用
    private static final int MIN_TRANSFER_STRIDE = 16;
    // 生成sizeCtl所使用的bit位数
    private static int RESIZE_STAMP_BITS = 16;
    // 进行扩容所允许的最大线程数
    private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
    // 记录sizeCtl中的大小所需要进行的偏移位数
    private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

    // 一系列的标识
    static final int MOVED     = -1; // hash for forwarding nodes
    static final int TREEBIN   = -2; // hash for roots of trees
    static final int RESERVED  = -3; // hash for transient reservations
    static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
    // 获取可用的CPU个数
    static final int NCPU = Runtime.getRuntime().availableProcessors();
    // 进行序列化的属性
    private static final ObjectStreamField[] serialPersistentFields = {
        new ObjectStreamField("segments", Segment[].class),
        new ObjectStreamField("segmentMask", Integer.TYPE),
        new ObjectStreamField("segmentShift", Integer.TYPE)
    };

重要的一个对象Node

Node是最核心的内部类,它包装了key-value键值对,所有插入ConcurrentHashMap的数据都包装在这里面,类似于一个HashMap

    //Node节点定义,可以看出Node是一个键值对
    static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        //ConcurrentHashMap中的HashEntry相对于HashMap中的Entry有一定的差异性:HashEntry中的value以及next都被volatile修饰,这样在多线程读写过程中能够保持它们的可见性。
        volatile V val;
        volatile Node<K,V> next;

        Node(int hash, K key, V val, Node<K,V> next) {
            this.hash = hash;
            this.key = key;
            this.val = val;
            this.next = next;
        }

        public final K getKey()       { return key; }
        public final V getValue()     { return val; }
        public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
        public final String toString(){ return key + "=" + val; }
        //不允许直接setValue
        public final V setValue(V value) {
            throw new UnsupportedOperationException();
        }

        public final boolean equals(Object o) {
            Object k, v, u; Map.Entry<?,?> e;
            return ((o instanceof Map.Entry) &&
                    (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
                    (v = e.getValue()) != null &&
                    (k == key || k.equals(key)) &&
                    (v == (u = val) || v.equals(u)));
        }

        /**
         * 它增加了find方法辅助map.get()方法。
         */
        Node<K,V> find(int h, Object k) {
            Node<K,V> e = this;
            if (k != null) {
                do {
                    K ek;
                    if (e.hash == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                } while ((e = e.next) != null);
            }
            return null;
        }
    }
   这个Node内部类与HashMap中定义的Node类很相似,但是有一些差别  
   它对value和next属性设置了volatile同步锁  
   它不允许调用setValue方法直接改变Node的value域  
   它增加了find方法辅助map.get()方法  

这里贴了一个能说明ConcurrentHashMap使用CAS算法的部分源码了,读者自行阅读源码会发现在ConcurrentHashMap中,随处可以看到U, 大量使用了U.compareAndSwapXXX的方法,这个方法是利用一个CAS算法实现无锁化的修改值的操作,简单说CAS就是对比替换,原理跟使用SVN差不多,千万记住是无锁操作哦。这里不过多解释,我会单独再发博客单独说说CAS……

 if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
                return;
            s = sumCount();
        }

TreeNode 这是一个数据结构,当链表长度过长的时候,会转换为TreeNode,是把这些结点包装成TreeNode放在TreeBin对象中,由TreeBin完成对红黑树的包装。

TreeBin 他是一个对象,包装的很多TreeNode节点,它代替了TreeNode的根节点,也就是说在实际的ConcurrentHashMap“数组”中,存放的是TreeBin对象,而不是TreeNode对象。

这里是它的构造方法源码。可以看到在构造TreeBin节点时,仅仅指定了它的hash值为TREEBIN常量,这也就是个标识为。同时也看到我们熟悉的红黑树构造方法

        /**
         * 创建以B为初始节点集的bin.
         */
        TreeBin(TreeNode<K,V> b) {
            super(TREEBIN, null, null, null);
            this.first = b;
            TreeNode<K,V> r = null;
            for (TreeNode<K,V> x = b, next; x != null; x = next) {
                next = (TreeNode<K,V>)x.next;
                x.left = x.right = null;
                if (r == null) {
                    x.parent = null;
                    x.red = false;
                    r = x;
                }
                else {
                    K k = x.key;
                    int h = x.hash;
                    Class<?> kc = null;
                    for (TreeNode<K,V> p = r;;) {
                        int dir, ph;
                        K pk = p.key;
                        if ((ph = p.hash) > h)
                            dir = -1;
                        else if (ph < h)
                            dir = 1;
                        else if ((kc == null &&
                                  (kc = comparableClassFor(k)) == null) ||
                                 (dir = compareComparables(kc, k, pk)) == 0)
                            dir = tieBreakOrder(k, pk);
                            TreeNode<K,V> xp = p;
                        if ((p = (dir <= 0) ? p.left : p.right) == null) {
                            x.parent = xp;
                            if (dir <= 0)
                                xp.left = x;
                            else
                                xp.right = x;
                            r = balanceInsertion(r, x);
                            break;
                        }
                    }
                }
            }
            this.root = r;
            assert checkInvariants(root);
        }

ForwardingNode 一个用于连接两个table的节点类。它包含一个nextTable指针,用于指向下一张表。而且这个节点的key value next指针全部为null,它的hash值为-1. 这里面定义的find的方法是从nextTable里进行查询下一个节点,而不是以自身为头节点进行查找下一个节点,可以比喻成两个车厢之间的链子。

    /**
     * 在传输过程中插入箱头的节点.
     */
    static final class ForwardingNode<K,V> extends Node<K,V> {
        final Node<K,V>[] nextTable;
        ForwardingNode(Node<K,V>[] tab) {
            super(MOVED, null, null, null);
            this.nextTable = tab;
        }

        Node<K,V> find(int h, Object k) {
            // loop to avoid arbitrarily deep recursion on forwarding nodes
            outer: for (Node<K,V>[] tab = nextTable;;) {
                Node<K,V> e; int n;
                if (k == null || tab == null || (n = tab.length) == 0 ||
                    (e = tabAt(tab, (n - 1) & h)) == null)
                    return null;
                for (;;) {
                    int eh; K ek;
                    if ((eh = e.hash) == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                    if (eh < 0) {
                        if (e instanceof ForwardingNode) {
                            tab = ((ForwardingNode<K,V>)e).nextTable;
                            continue outer;
                        }
                        else
                            return e.find(h, k);
                    }
                    if ((e = e.next) == null)
                        return null;
                }
            }
        }
    }

initTable 这是一个初始化方法,调用它的构造方法仅仅是设置了一些参数而已。而整个table的初始化是在向ConcurrentHashMap中插入元素的时候发生的。如调用put、computeIfAbsent、compute、merge等方法的时候都有调用initTable方法的代码,调用时机是检查table==null。

主要应用了关键属性sizeCtl ,如果这个值小于0,表示其他线程正在进行初始化,就放弃这个操作。ConcurrentHashMap的初始化只能由一个线程完成。如果获得了初始化权限,就用CAS方法将sizeCtl置为-1,防止其他线程进入。初始化数组后,将sizeCtl的值改为0.75*n

     /**
     * 使用大小记录的大小初始化表.
     */
    private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
//sizeCtl表示有其他线程正在进行初始化操作,把线程挂起。对于table的初始化工作,只能有一个线程在进行。
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//利用CAS方法把sizectl的值置为-1 表示本线程正在进行初始化 
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);//相当于0.75*n 设置一个扩容的阈值
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

扩容transfer 无论是HashMap还是ConcurrentHashMap使用的时候都需要扩容的,下面我只简单的说,让大家读最少的文字,理解最重要的东西。

上面我提到了ConcurrentHashMap是一个大的table,所以扩容也是针对这个大table操作的,简单的说就是把原来的table复制到一个他两倍大小的nextTable中。这个操作需要两部完成:

  • 第一步:构建一个nextTable,它的容量是原来的两倍,这个操作是单线程完成的。这个单线程的保证是通过RESIZE_STAMP_SHIFT这个常量经过一次运算来保证的。
  • 第二步:就是将原来table中的元素复制到nextTable中,这里允许多线程进行操作。

多线程扩容是如何完成。大体思想就是遍历、复制的过程,如果遍历到的节点是forward节点,就向后继续遍历,再加上给节点上锁的机制,就完成了多线程的控制。多线程遍历节点,处理了一个节点,就把对应点的值set为forward,另一个线程看到forward,就向后遍历。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。我说的forwad就是ForwardingNode,就是上面说的指向下一张表的指针。

PUT ConcurrentHashMap最常用的就是put和get两个方法。现在来介绍put方法,这个put方法依然沿用HashMap的put方法的思想,根据hash值计算这个新插入的点在table中的位置i,如果i位置是空的,直接放进去,否则进行判断,如果i位置是树节点,按照树的方式插入新的节点,否则把i插入到链表的末尾。有一个最重要的不同点就是ConcurrentHashMap不允许key或value为null值。另外由于涉及到多线程,put方法就要复杂一点。在多线程中可能有以下两个情况

如果一个或多个线程正在对ConcurrentHashMap进行扩容操作,当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到,是因为transfer方法中在空结点上插入forward节点,如果检测到需要插入的位置被forward节点占有,就帮助进行扩容;

如果检测到要插入的节点是非空且不是forward节点,就对这个节点加锁,这样就保证了线程安全。尽管这个有一些影响效率,但是还是会比hashTable的synchronized要好得多。

get方法 get方法比较简单,给定一个key来确定value的时候,必须满足两个条件 key相同 hash值相同,对于节点可能在链表或树上的情况,需要分别去查找。

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());  
        //根据hash值确定节点位置  
        if ((tab = table) != null && (n = tab.length) > 0 &&  
            (e = tabAt(tab, (n - 1) & h)) != null) {  
            //如果搜索到的节点key与传入的key相同且不为null,直接返回这个节点    
            if ((eh = e.hash) == h) {  
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))  
                    return e.val;  
            }  
            //如果eh<0 说明这个节点在树上 直接寻找  
            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;  
    } 

mappingCount与Size方法 对于ConcurrentHashMap来说,这个table里到底装了多少东西其实是个不确定的数量,因为不可能在调用size()方法的时候像GC的“stop the world”一样让其他线程都停下来让你去统计,因此只能说这个数量是个估计值。对于这个估计值,ConcurrentHashMap也是大费周章才计算出来的。

mappingCount与size方法的类似 ,是JDK1.8的新方法,从给出的注释来看,应该使用mappingCount代替size方法,mappingCount其实也是一个大概的数值,因此可能在统计的时候有其他线程正在执行插入或删除操作。

 /**
     * Returns the number of mappings. This method should be used
     * instead of {@link #size} because a ConcurrentHashMap may
     * contain more mappings than can be represented as an int. The
     * value returned is an estimate; the actual count may differ if
     * there are concurrent insertions or removals.
     *
     * @return the number of mappings
     * @since 1.8
     */
    public long mappingCount() {
        long n = sumCount();
        return (n < 0L) ? 0L : n; // ignore transient negative values
    }


final long sumCount() {
        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;
    }

注:对本文有异议或不明白的地方微信探讨,wx:15524579896

本文分享自微信公众号 - 晏霖(yanlin199507),作者:晏霖

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2019-01-31

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • HashMap JDK8的原理讲解

    本文讲解 HashMap JDK 8 的原理,结合源码,只分析 put ,get ,resize 方法的流程。

    胖虎
  • java并发之重入锁-ReentrantLock

    目前主流的锁有两种,一种是synchronized,另一种就是ReentrantLock,JDK优化到现在目前为止synchronized的性能已经和重入锁不分...

    胖虎
  • Java8-对List转换Map、分组、求和、过滤

    在java8之后我们list转map再也不用循环put到map了,我们用lambda表达式,使用stream可以一行代码解决,下面我来简单介绍list转map的...

    胖虎
  • Android--ItemTouchHelper源码分析

    aruba
  • 跳表(skiplist)的原理及concurrentskiplistmap的源码学习

    本文分为两个部分,第一个是对跳表(SKipList)这种数据结构的介绍,第二部分则是对Java中ConcurrentSkilListMap的源码解读.

    呼延十
  • dubbo路由机制代码分析1

    这回说说,dubbo路由特性,dubbo的路由干的事,就是一个请求过来, dubbo依据配置的路由规则,计算出哪些提供者可以提供这次的请求服务。 所以,它的...

    技术蓝海
  • P1328 生活大爆炸版石头剪刀布

    题目描述 石头剪刀布是常见的猜拳游戏:石头胜剪刀,剪刀胜布,布胜石头。如果两个人出拳一样,则不分胜负。在《生活大爆炸》第二季第8 集中出现了一种石头剪刀布的升级...

    attack
  • 【java基础之ConcurrentHashMap源码分析】

    ConcurrentHashMap这个类在java.lang.current包中,这个包中的类都是线程安全的。Concurrent...

    用户5640963
  • iOS UIActivityIndicatorView(指示控制器)用法总结

    对于UIActivityIndicatorView的使用,我们一般会创建一个背景View,设置一定的透明度,然后将UIActivityIndicatorView...

    珲少
  • Spring循环依赖问题修复

    拆分的时候,把错误都处理完后,准备把工程起起来,发现弹簧的循环依赖问题。具体问题如下

    端游岚

扫码关注云+社区

领取腾讯云代金券