Java技术探索「ConcurrentHashMap」深入浅出的源码分析（JDK1.8版本）

Posted 2021-08-16 java构架师

前提概要

ConcurrentHashMap是concurrent家族中的一个类，由于它可以高效地支持并发操作，以及被广泛使用，经典的开源框架Spring的底层数据结构就是使用ConcurrentHashMap实现的。

与同是线程安全的老大哥HashTable相比，它已经更胜一筹，因此它的锁更加细化，而不是像HashTable一样为几乎每个方法都添加了synchronized锁，这样的锁无疑会影响到性能。

原理简介

本文的分析的源码是JDK8的版本，与JDK7的版本有很大的差异。实现线程安全的思想也已经完全变了，它摒弃了Segment（锁段）的概念，而是启用了一种全新的方式实现，利用CAS算法。

它沿用了与它同时期(JDK1.8)的HashMap版本的思想，底层依然由“数组”+链表+红黑树的方式思想，但是为了做到并发，又增加了很多辅助的类，例如TreeBin，Traverser等对象内部类。

重要的属性

首先来看几个重要的属性，与HashMap相同的就不再介绍了，这里重点解释一下sizeCtl这个属性。可以说它是ConcurrentHashMap中出镜率很高的一个属性，因为它是一个控制标识符，在不同的地方有不同用途，而且它的取值不同，也代表不同的含义。

sizeCtl

• 负数代表正在进行初始化或扩容操作
• -1 代表正在初始化
• -N 表示有N-1个线程正在进行扩容操作
• 正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小，这一点类似于扩容阈值的概念。

还后面可以看到，它的值始终是当前ConcurrentHashMap容量的0.75倍，这与loadfactor是对应的。

/**
* 盛装Node元素的数组 它的大小是2的整数次幂
* Size is always a power of two. Accessed directly by iterators.
*/
transient volatile Node<K,V>[] table;
/**
* Table initialization and resizing control. When negative, the
* table is being initialized or resized: -1 for initialization,
* else -(1 + the number of active resizing threads). Otherwise,
* when table is null, holds the initial table size to use upon
* creation, or 0 for default. After initialization, holds the
* next element count value upon which to resize the table.
hash表初始化或扩容时的一个控制位标识量。
负数代表正在进行初始化或扩容操作
-1代表正在初始化
-N 表示有N-1个线程正在进行扩容操作
正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小
*/
private transient volatile int sizeCtl;

// 以下两个是用来控制扩容的时候 单线程进入的变量
/**
* The number of bits used for generation stamp in sizeCtl.
* Must be at least 6 for 32bit arrays.
*/
private static int RESIZE_STAMP_BITS = 16;
/**
* The bit shift for recording size stamp in sizeCtl.
*/
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
/*
* Encodings for Node hash fields. See above for explanation.
*/
static final int MOVED = -1; // hash值是-1，表示是一个forwardNode节点
static final int TREEBIN = -2; // hash值是-2 表示这时一个TreeBin节点

重要的内部类

Node（链表）

Node是最核心的内部类，它包装了key-value键值对，所有插入ConcurrentHashMap的数据都包装在这里面。

如下所示：

static class Node<K,V> implements Map.Entry<K,V> {
        final int hash; //hash值
        final K key; //键值key
        volatile V val;//带有sync锁的value
        volatile Node<K,V> next;//带有sync锁的next指针
        Node(int hash, K key, V val, Node<K,V> next) {
            this.hash = hash; // hash
            this.key = key; // key 
            this.val = val; // value
            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; }
        //不允许直接改变value的值
        public final V setValue(V value) {
            throw new UnsupportedOperationException();
        }
	   // 比较两个node是否相等
        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)) && // 再比 key
                    (v == (u = val) || v.equals(u))); // 再比 value值
        }
        /**
         * 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;
        }
    }

这个Node内部类与HashMap中定义的Node类很相似

• 但是有一些差别它对value和next属性设置了volatile的sync锁
• 它不允许调用setValue方法直接改变Node的value域
• 它增加了find方法辅助map.get()方法

TreeNode（树节点）

• 树节点类，另外一个核心的数据结构。
• 当链表长度过长的时候，会转换为TreeNode。

但是与HashMap不相同的是，它并不是直接转换为红黑树，而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。

而且TreeNode在ConcurrentHashMap集成自Node类，而并非HashMap中的集成自LinkedHashMap.Entry<K,V>类，也就是说TreeNode带有next指针，这样做的目的是方便基于TreeBin的访问。

TreeBin

这个类并不负责包装用户的key、value信息，而是包装的很多TreeNode节点。它代替了TreeNode的根节点，也就是说在实际的ConcurrentHashMap“数组”中，存放的是TreeBin对象，而不是TreeNode对象，这是与HashMap的区别。另外这个类还带有了读写锁。

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

/**
 * Creates bin with initial set of nodes headed by b.
 */
TreeBin(TreeNode<K,V> b) {
    super(TREEBIN, null, null, null);
    this.first = b; // 包装进去作为first节点
    TreeNode<K,V> r = null;
	//初始化根节点
    for (TreeNode<K,V> x = b, next; x != null; x = next) {
        next = (TreeNode<K,V>)x.next; //传入的节点的next赋值为当前treebin节点的next节点，作为数据同步使用
        x.left = x.right = null; //将左子树和右子树都是null
        if (r == null) {
            x.parent = null;//父节点为空
            x.red = false; // 不属于红子树节点
            r = x; //将红节点，临时存储起来，用于下次循环使用
        }
        else {
            K k = x.key; //如果不是根节点则就是下一个子树节点（属于next值）
            int h = x.hash; // 如果不是根节点就取出hash值
            Class<?> kc = null;
            for (TreeNode<K,V> p = r;;) {
                int dir, ph; //
                K pk = p.key;//获取根节点的key值 
                if ((ph = p.hash) > h) // 获取hash值判断是否属于小于根节点hash值
                    dir = -1; //属于左子树，
                else if (ph < h) // 否则属于右子树
                    dir = 1;
                else if ((kc == null &&
                          (kc = comparableClassFor(k)) == null) ||
                         (dir = compareComparables(kc, k, pk)) == 0)
					// 当hash值相等的时候，比较class类以及比较key值
                    dir = tieBreakOrder(k, pk);
                    TreeNode<K,V> xp = p;
                if ((p = (dir <= 0) ? p.left : p.right) == null) {
					// 判断比较结果是否小于零，且左右子树都为null的情况下
                    x.parent = xp;// 将父节点进行赋值
                    if (dir <= 0)
                        xp.left = x; // 进行判断是否属于左子树（<=0）
                    else
                        xp.right = x;// 进行判断是否属于右子树（>0）
					// 进行平衡插入机制（其实就是建立平衡关系）
                    r = balanceInsertion(r, x);
                    break;
                }
            }
        }
    }
    this.root = r; // 根节点
    assert checkInvariants(root);
}

ForwardingNode

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

/**
 * A node inserted at head of bins during transfer operations.
 */
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;
    }
	// 循环遍历查找相关的对应hash和key值的
    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;
			// 如果无法定位或者定位到null则直接返回
            if (k == null || tab == null || (n = tab.length) == 0 ||
                (e = tabAt(tab, (n - 1) & h)) == null)
                return null;
            for (;;) {
                int eh; K ek;
				// 遍历查找对比hash值和key值，从而进行分析是否相等。
                if ((eh = e.hash) == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))
					// 查找到后相等直接返回
                    return e;
				//如果hash值为-1 或者 -2 
                if (eh < 0) {
					// 如果是forwardingNode 则就是代表是-1
                    if (e instanceof ForwardingNode) {
						//直接获取到下一个表。
                        tab = ((ForwardingNode<K,V>)e).nextTable;
                        continue outer;
                    }
                    else
                        return e.find(h, k);
                }
				//直到为null，直接返回
                if ((e = e.next) == null)
                    return null;
            }
        }
    }
}

Unsafe与CAS

在ConcurrentHashMap中，随处可以看到Unsafe, 大量使用了Unsafe.compareAndSwapXXX的方法，这个方法是利用一个CAS算法实现无锁化的修改值的操作，他可以大大降低锁代理的性能消耗。

这个算法的基本思想就是不断地去比较当前内存中的变量值与你指定的一个变量值是否相等，如果相等，则接受你指定的修改的值，否则拒绝你的操作。

因为当前线程中的值已经不是最新的值，你的修改很可能会覆盖掉其他线程修改的结果。这一点与乐观锁，SVN的思想是比较类似的。

unsafe静态块

unsafe代码块控制了一些属性的修改工作，比如最常用的SIZECTL 。 在这一版本的concurrentHashMap中，大量应用来的CAS方法进行变量、属性的修改工作。 利用CAS进行无锁操作，可以大大提高性能。

// 各个字段属性的偏移量，可以通过偏移量+首地址获取到对应的数据对象
private static final sun.misc.Unsafe U;
private static final long SIZECTL;
private static final long TRANSFERINDEX;
private static final long BASECOUNT;
private static final long CELLSBUSY;
private static final long CELLVALUE;
private static final long ABASE;
private static final int ASHIFT;
 
static {
    try {
        U = sun.misc.Unsafe.getUnsafe();
        Class<?> k = ConcurrentHashMap.class;
        SIZECTL = U.objectFieldOffset
            (k.getDeclaredField("sizeCtl"));
        TRANSFERINDEX = U.objectFieldOffset
            (k.getDeclaredField("transferIndex"));
        BASECOUNT = U.objectFieldOffset
            (k.getDeclaredField("baseCount"));
        CELLSBUSY = U.objectFieldOffset
            (k.getDeclaredField("cellsBusy"));
        Class<?> ck = CounterCell.class;
        CELLVALUE = U.objectFieldOffset
            (ck.getDeclaredField("value"));
        Class<?> ak = Node[].class;
        ABASE = U.arrayBaseOffset(ak);
        int scale = U.arrayIndexScale(ak);
        if ((scale & (scale - 1)) != 0)
            throw new Error("data type scale not a power of two");
        ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
    } catch (Exception e) {
        throw new Error(e);
    }
}

三个核心方法

ConcurrentHashMap定义了三个原子操作，用于对指定位置的节点进行操作。正是这些原子操作保证了ConcurrentHashMap的线程安全。

@SuppressWarnings("unchecked")
//获得在i位置上的Node节点
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);
}

• 利用CAS算法设置i位置上的Node节点。之所以能实现并发是因为他指定了原来这个节点的值是多少
• 在CAS算法中，会比较内存中的值与你指定的这个值是否相等，如果相等才接受你的修改，否则拒绝你的修改
• 因此当前线程中的值并不是最新的值，这种修改可能会覆盖掉其他线程的修改结果 有点类似于SVN

static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                    Node<K,V> c, Node<K,V> v) {
    return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
//利用volatile方法设置节点位置的值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}

初始化方法initTable

• 对于ConcurrentHashMap来说，调用它的构造方法仅仅是设置一些参数。
• 整个table的初始化是在向ConcurrentHashMap中插入元素的时候发生的。
  • 如调用put、computeIfAbsent、compute、merge等方法的时候，调用时机是检查table==null。

1. 初始化方法主要应用了关键属性sizeCtl 如果这个值 <0，表示其他线程正在进行初始化，就放弃这个操作。
2. 在这也可以看出ConcurrentHashMap的初始化只能由一个线程完成。
3. 如果获得了初始化权限，就用CAS方法将sizeCtl置为-1，防止其他线程进入。

初始化数组后，将sizeCtl的值改为0.75 * n，源码如下：

/**
 * Initializes table, using the size recorded in sizeCtl.
 */
private final Node<K,V>[] initTable() {
    Node<K以上是关于Java技术探索「ConcurrentHashMap」深入浅出的源码分析（JDK1.8版本）的主要内容，如果未能解决你的问题，请参考以下文章
 
 Java技术专题「性能优化系列」针对Java对象压缩及序列化技术的探索之路
 第四期技术分享：JAVA之深入浅出设计模式 + 移动技术前沿之探索Flutter框架
 ☕Java深层系列「并发编程系列」让我们一起探索一下CyclicBarrier的技术原理和源码分析
 Java技术探索「开发实战专题」Lombok插件开发实践必知必会操作！
 Java深层系列「并发编程系列」让我们一起探索一下CompletionService的技术原理和使用指南
 JVM技术专题「原理专题」让我们一起探索一下Netty(Java)底层的“零拷贝Zero-Copy”技术（上）