ConcurrentHashMap的底层实现原理与源码剖析

最新推荐文章于 2024-02-20 17:48:11 发布

程序员楂楂

最新推荐文章于 2024-02-20 17:48:11 发布

阅读量2.9k

点赞数 6

分类专栏： HashMap源码剖析

本文链接：https://blog.csdn.net/ToBe_Coder/article/details/86996665

版权

HashMap源码剖析专栏收录该内容

3 篇文章 0 订阅

订阅专栏

简单介绍：在程序中需要线程安全而不需要同步的情况下，ConcurrentHashMap与HashTable是完全可以互操作的。

HashMap是根据散列值分段存储的，同步Map在同步的时候锁住了所有的段，而ConcurrentHashMap加锁的时候根据散列值锁住了散列值对应的那段，因此提高了并发性能。

ConcurrentHashMap内部结构：
ConcurrentHashMap为了提高本身的并发能力，在内部采用了一个叫做Segment的结构，一个Segment其实就是一个类HashTable的结构，Segment内部维护了一个链表数组，我们用下面这一幅图来看下ConcurrentHashMap的内部结构：

从上图可以了解到， ConcurrentHashMap定位一个元素的过程需要进行两次Hash操作。第一次Hash定位到Segment，第二次Hsah定位到元素所在的链表的头部，因此，这一种结构的副作用是Hash的过程要比普通的HashMap要长，但是带来的好处是写操作的时候可以只对元素所在的Segment进行加锁即可，不会影响到其他的Segment。这样，在最理想的情况下，ConcurrentHashMap可以最高同时支持Segment数量大小的写操作（刚好这些写操作非常平均的分布在所有的Segment上），所以，通过这一种结构，ConcurrentHashMap的并发能力可以大大提高。

以下对源码进行分析：

Segment：segment是ConcurrentHashMap的一个静态内部类，继承了ReentrantLock。

static class Segment<K,V> extends ReentrantLock implements Serializable {
        private static final long serialVersionUID = 2249069246763182397L;//序列版本号
        final float loadFactor;//加载因子
        Segment(float lf) { this.loadFactor = lf; } //构造函数
    }

ReentrantLock：重入锁（ReentrantLock）是一种递归无阻塞的同步机制。Java无法直接访问底层操作系统，而是通过本地（native）方法来访问。不过尽管如此，JVM还是开了一个后门，JDK中有一个类Unsafe,它提供了硬件级别的原子操作，也就是说Unsafe类提供了硬件级别的操作。

JDK1.8进行ConcurrentHashMap底层实现分析：

jdk8中，利用CAS+Synchronized来保证并发更新的安全，底层依然采用数组+链表+红黑树的存储结构。

ConcurrentHashMap类中的数据结构源码分析：

Node节点：

    static class Node<K,V> implements Map.Entry<K,V> {   //Node 节点
        final int hash; //哈希值
        final K key;  //键
        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; }
        //计算Hash值
        public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
        public final String toString(){ return key + "=" + val; }
        public final V setValue(V value) {
            throw new UnsupportedOperationException();
        }
        //比较两个entry是否相等
        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)));
        }

        /**
         * Virtualized support for map.get(); overridden in subclasses.
         */
        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;
        }
    }

存放节点的数组：

//默认为null，在第一次进行插入操作时进行初始化，大小总是2幂次方，可以由迭代器进行访问。
transient volatile Node<K,V>[] table;

扩容时新生节点数组：

    //默认为null，扩容时新生成的数组。
    private transient volatile Node<K,V>[] nextTable;

    //对表初始化和调整控制。sizeCtl默认为0，如果ConcurrentHashMap实例化时有传参数，sizeCtl会是一个2的幂次方的值。
    //-1表示Map正在初始化
    // —N表示有N-1个线程在执行扩容操作
    // 如果table是null的，那么这个值就是由构造方法执行时确定的
    private transient volatile int sizeCtl;   //本质就是一个标志控制变量

实例化ConcurrentHashMap时带参数时，会根据参数调整table的大小，大小总是2幂次方：

private static final int tableSizeFor(int c) {
        int n = c - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }

ConcurrentHashMap空构造器：

/**创建一个新的，空的map,并且默认的表的大小为16
* Creates a new, empty map with the default initial table size (16).
*/
public ConcurrentHashMap() {
}

ConcurrentHashMap带初始容量的构造器：

//创建一个新的空映射，带有初始表大小
public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
               MAXIMUM_CAPACITY :
               tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    this.sizeCtl = cap;
}

ConcurrentHashMap带初始容量，加载因子和并发数的构造器：

public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
		//如果初始容量小于并发数，那么初始容量就等于并发数，相当于一个位桶一个并发数
        if (initialCapacity < concurrencyLevel)   // Use at least as many bins
            initialCapacity = concurrencyLevel;   // as estimated threads
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
        this.sizeCtl = cap;
    }

注：concurrencyLevel：
concurrencyLevel，能够同时更新ConccurentHashMap且不产生锁竞争的最大线程数，在Java8之前实际上就是ConcurrentHashMap中的分段锁个数，即Segment[]的数组长度。正确地估计很重要，当低估，数据结构将根据额外的竞争，从而导致线程试图写入当前锁定的段时阻塞；相反，如果高估了并发级别，你遇到过大的膨胀，由于段的不必要的数量; 这种膨胀可能会导致性能下降，由于高数缓存未命中。
在Java8里，仅仅是为了兼容旧版本而保留。唯一的作用就是保证构造map时初始容量不小于concurrencyLevel。

table初始化操作：

private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
			//如果一个线程发现sizeCtl<0，意味着另外的线程执行CAS操作成功，当前线程只需要让出cpu时间片
            if ((sc = sizeCtl) < 0)
                Thread.yield(); // lost initialization race; just spin
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//U.compareAndSwapInt(this, SIZECTL, sc, -1)操作成功返回true
                try {//当一个线程将SIZECTL处的值修改为-1时，其他线程就不能初始化table
                    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);
                    }
                } finally {
					//初始化完毕之后将内存中SIZECTL处即sizeCtl改回为原来的值。
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

注：
sizeCtl默认为0，如果ConcurrentHashMap实例化时有传参数，sizeCtl会是一个2的幂次方的值。所以执行第一次put操作的线程会执行Unsafe.compareAndSwapInt方法修改sizeCtl为-1，有且只有一个线程能够修改成功，其它线程通过Thread.yield()让出CPU时间片等待table初始化完成。

U.compareAndSwapInt(this, SIZECTL, sc, -1)这个方法有四个参数，其中第一个参数为需要改变的对象，第二个为偏移量(即之前求出来的valueOffset的值)，第三个参数为期待的值，第四个为更新后的值。整个方法的作用即为若调用该方法时，value的值与expect这个值相等，那么则将value修改为update这个值，并返回一个true，如果调用该方法时，value的值与expect这个值不相等，那么不做任何操作，并返回一个false。

table初始化操作会延迟到第一次put操作，分析put操作：

//put（） and putIfAbsent（）的实现方法
	final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
		//计算key的hash值
        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)//如果tab为空的情况
                tab = initTable();//对哈希表进行初始化，
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {f为指定元素在hash数组中的位置，如果指定位置为空，那么将元素放入其中
                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 { //待插入元素应放在链表中的情况
                V oldVal = null;
                synchronized (f) {
                    if (tabAt(tab, i) == f) { //找到f所在的数组
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                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;
                                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) {//binCount刚好为8，则将链表转换为红黑树
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

代码解释：
    1、hash算法：
static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;}
    2、table中定位索引位置，n是table的大小
int index = (n - 1) & hash
    3、获取table中对应索引的元素f。
    查看源码：
    static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
        return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    }
    Doug Lea采用Unsafe.getObjectVolatile来获取，也许有人质疑，直接table[index]不可以么，为什么要这么复杂？
    在java内存模型中，我们已经知道每个线程都有一个工作内存，里面存储着table的副本，虽然table是volatile修饰的，但不能保证线程每次都拿到table中的最新元素，Unsafe.getObjectVolatile可以直接获取指定内存的数据，保证了每次拿到数据都是最新的。
    4、如果f为null，说明table中这个位置第一次插入元素，利用Unsafe.compareAndSwapObject方法插入Node节点。
    如果CAS成功，说明Node节点已经插入，随后addCount(1L, binCount)方法会检查当前容量是否需要进行扩容。
    如果CAS失败，说明有其它线程提前插入了节点，自旋重新尝试在这个位置插入节点。
    5、如果f的hash值为-1，说明当前f是ForwardingNode节点，意味有其它线程正在扩容，则一起进行扩容操作。
    6、其余情况把新的Node节点按链表或红黑树的方式插入到合适的位置，这个过程采用同步内置锁实现并发。
    7、在节点f上进行同步，节点插入之前，再次利用tabAt(tab, i) == f判断，防止被其它线程修改。
    如果f.hash >= 0，说明f是链表结构的头结点，遍历链表，如果找到对应的node节点，则修改value，否则在链表尾部加入节点。
    如果f是TreeBin类型节点，说明f是红黑树根节点，则在树结构上遍历元素，更新或增加节点。
    如果链表中节点数binCount >= TREEIFY_THRESHOLD(默认是8)，则把链表转化为红黑树结构。

TreeNode源码分析：

//在树结构中的节点
	static final class TreeNode<K,V> extends Node<K,V> {
        TreeNode<K,V> parent;  // 红黑树链接
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;

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

        Node<K,V> find(int h, Object k) {
            return findTreeNode(h, k, null);
        }

        /**根据所给的key从根节点开始查找,找到返回该树节点，没找到则返回null
         * Returns the TreeNode (or null if not found) for the given key
         * starting at given root.
         */
        final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
            if (k != null) {
                TreeNode<K,V> p = this;
                do  {
                    int ph, dir; K pk; TreeNode<K,V> q;
                    TreeNode<K,V> pl = p.left, pr = p.right;
                    if ((ph = p.hash) > h)
                        p = pl;
                    else if (ph < h)
                        p = pr;
                    else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
                        return p;
                    else if (pl == null)
                        p = pr;
                    else if (pr == null)
                        p = pl;
                    else if ((kc != null ||
                              (kc = comparableClassFor(k)) != null) &&
                             (dir = compareComparables(kc, k, pk)) != 0)
                        p = (dir < 0) ? pl : pr;
                    else if ((q = pr.findTreeNode(h, k, kc)) != null)
                        return q;
                    else
                        p = pl;
                } while (p != null);
            }
            return null;
        }
    }

程序员楂楂

关注

6
点赞
踩
17

收藏

觉得还不错? 一键收藏
0
评论
ConcurrentHashMap的底层实现原理与源码剖析

简单介绍：在程序中需要线程安全而不需要同步的情况下，ConcurrentHashMap与HashTable是完全可以互操作的。HashMap是根据散列值分段存储的，同步Map在同步的时候锁住了所有的段，而ConcurrentHashMap加锁的时候根据散列值锁住了散列值对应的那段，因此提高了并发性能。ConcurrentHashMap内部结构：ConcurrentHashMap为了提高本...
复制链接

扫一扫