ConcurrentHashMap源码讲解

• 前言
• 一、重要常量和变量
• 二、构造器（5个）
• put(...)方法
• • 1.putVal(...)方法
  • 2. initTable()方法--初始化数组
  • 3. 其他四个方法

由于 HashMap 是线程不安全的，无法运用到高并发的环境中，这时候线程安全的 ConcurrentHashMap 就有用武之地了。
ConcurrentHashMap 是基于分段锁（synchronized）和 CAS 来保证线程安全的。

// 数组最大长度 = 2^30
private static final int MAXIMUM_CAPACITY = 1 << 30;
// 数组默认长度 = 16
private static final int DEFAULT_CAPACITY = 16;
// 加载因子=0.75，当数组使用率达到0.75时，会进行扩容
private static final float LOAD_FACTOR = 0.75f;
// 树化条件1：链表长度>8
static final int TREEIFY_THRESHOLD = 8;
// 取消树化条件：链表长度<6
static final int UNTREEIFY_THRESHOLD = 6;
// 树化条件2：数组长度>=64
static final int MIN_TREEIFY_CAPACITY = 64;
// 扩容转移最少需要16个索引，因为扩容时其他线程可以帮忙，所以限定每个线程转移槽位数最小为16
private static final int MIN_TRANSFER_STRIDE = 16;
// 当数组索引位的节点的hash等于这个常量时，表示正在扩容
static final int MOVED     = -1; 
// 此元素后为红黑树
static final int TREEBIN   = -2; 
// 表示临时保留的hash
static final int RESERVED  = -3; 
// 是一个16进制的数，大小位2^31-1,作用是将key的hash转成正数
static final int HASH_BITS = 0x7fffffff;

// 数组
transient volatile Node[] table;
// 数据转移时的中间表
private transient volatile Node[] nextTable;
// 是一个非常重要的变量。用来表示数组初始化和扩容的，
// 当为0时，表示还未初始化；
// 当为-1时，表示正在初始化；
// 小于-1时，表示正在扩容；
// 大于0时，表示初始化完成，这时候等于0.75*数组长度
private transient volatile int sizeCtl;
//数据转移时，当前已经迁移的元素下标
private transient volatile int transferIndex;

// 构造器一
public 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;
}

// 构造器三
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}

// 构造器四
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;
}

// 构造器五
public ConcurrentHashMap(Map m) {
    this.sizeCtl = DEFAULT_CAPACITY;
    putAll(m);
}

总共有五个构造器，这里就讲解比较常用的。

构造器二：参数是一个初始容量，先进行非法从参数判断，接着判断参数是否超过最大允许容量的一半（ >>> 无符号位移，一半也就是 2^29），超过了就将 sizeCtl （数组未初始化时表示为数组长度）赋值为 2^30 ，否则就将 参数×1.5+1 ，并且调用 tableSizeFor(...) 方法赋值。 tableSizeFor(...) 这个方法的作用是保证数组的长度为 2 的幂次，它最终得到一个 2 的幂次值，并且这个值大于等于且最接近参数。比如构造器参数 initialCapacity=16 ，那么 16*1.5+1=25 ，最接近的是 32 ，所以 sizeCtl=32 。

注意：为什么要将 初始容量×1.5+1 ？

假设我现在要存 16 个键值对，然后 new ConcurrentHashMap(16) ，如果只创建16 长度的数组在散列均匀的情况下肯定不够的（阈值为 0.75，也就是在大于 12 的时候就开始扩容了），这时候就需要扩容了。所以为了解决这个问题，需要将 16÷0.75 ，也就是 16×1.333 ，为了提高效率，将 16×1.5 采用位运算，而 +1 是为了进 1 ，弥补数据。

public V put(K key, V value) {
    return putVal(key, value, false);
}

final V putVal(K key, V value, boolean onlyIfAbsent) {
    // 不允许null值null键
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());  // 通过hash算法获得hash值，下面展示
    int binCount = 0;
    // 并发情况下插入可能会失败，而这自旋是保证插入一定成功
    for (Node[] tab = table;;) {  
        Node f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)  // 如果数组还未初始化，那么就去初始化
            tab = initTable();
        // 如果数组当前索引位置为空
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 使用CAS执行插入操作
            if (casTabAt(tab, i, null,new Node(hash, key, value, null)))
                break;                  
        }
        // 当数组索引位置的节点为-1时，说明有线程正在扩容
        else if ((fh = f.hash) == MOVED) 
            tab = helpTransfer(tab, f); // 帮助扩容
        else {   // 出现hash冲突
            V oldVal = null;
            synchronized (f) {  // 锁的是链表头节点
                if (tabAt(tab, i) == f) { // 再次判断
                    if (fh >= 0) {  // 头节点hash大于0，表示是链表
                        binCount = 1;
                        for (Node e = f;; ++binCount) { // 开始遍历链表
                            K ek;
                            // key已经存在，则value覆盖
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value;
                                break;
                            }
                            Node pred = e;
                            if ((e = e.next) == null) { // 如果是尾节点，就执行插入
                                pred.next = new Node(hash, key, value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) { // 如果是红黑树
                        Node p;
                        binCount = 2;
                        if ((p = ((TreeBin)f).putTreeval(hash, key, value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD) // 当链表长度大于8，也就是9时，调用树化方法
                    treeifyBin(tab, i);
                if (oldVal != null) // 如果key已经存在，就将原来的value值返回
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount); // 扩容方法就在里面
    return null;
}

// 求hash的算法，h ^ (h >>> 16)与前面讲的HashMap相同，& HASH_BITS为了保证得到的hash一定是个非负数
static final int spread(int h) {
    return (h ^ (h >>> 16)) & HASH_BITS;
}

总体来说和 HashMap 是非常接近的，就是多了 synchronized 和 CAS 。

第一次 put(...) 时，数组还未进行初始化，调用 initTable() 方法进行初始化。初始化完后，进行第二次循环，根据 hash 算出索引位置（(n - 1) & hash)），这时索引位肯定是空的，所以直接使用 CAS 将新节点赋在索引位上，

• 赋值成功就跳出循环，调用 addCount(...) 方法；

• 赋值失败（其他线程赋值成功）就进行第三次循环，这个时候索引位置已经不为空了，所以开始获取锁，获取锁成功后，开始对索引位置的节点进行遍历，判断是否有 key 相同的:

  • 有相同的 value 就将覆盖。并且将原来的 value 返回；
  • 没有就创建新节点，同时将尾节点的 next 指针指向新节点，调用 addCount(...) 方法;

private final Node[] initTable() {
    Node[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        if ((sc = sizeCtl) < 0) // 有其他线程正在初始化
            Thread.yield();  
        // CAS将sizeCtl设为-1,保证只有一个线程正在执行扩容
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // 这里的sc就是sizeCtl (还未经过前面CAS设置的值)
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY; 
                    @SuppressWarnings("unchecked")
                    // 创建数组
                    Node[] nt = (Node[])new Node[n];
                    table = tab = nt;
                    sc = n - (n >>> 2); //n-0.25n=0.75n
                }
            } finally {
                sizeCtl = sc; //将sizeCtl设置为数组阈值,也就是0.75n
            }
            break;
        }
    }
    return tab;
}

线程进入 initTable() 方法时，先进入 while 循环，这个循环为了让那些失去初始化竞争的线程进行自旋，并且初始化成功后跳出循环。

• 当成员变量的 sizeCtl 小于 0 时，说明有线程正在初始化，那么就暂停当前线程，执行其他线程

• 当成员变量的 sizeCtl 不小于 0 ，并且 CAS 设置成功，就创建一个节点数组，根据 sizeCtl 确定数组长度：

  • sizeCtl 大于 0 ，数组长度为 sizeCtl ；

  • sizeCtl 不大于 0 ，数组查高度为默认值 16 ；

最后将 sizeCtl 设置为数组阈值（ 数组长度×0.75 ）

helpTransfer(...) ：帮助扩容，每个线程最少需要转移 16 个槽位

treeifyBin(...) ：树化方法，树化条件与 HashMap 一致

addCount(...) ：计数，将元素个数 +1 ，并且判断是否符合扩容条件，符合就进行扩容（也可能是帮助其他线程扩容）

transfer(...) ：扩容方法

在这里这四个方法就不详细讲解了，主要原因是因为在多线程情况下，有无数种可能，单靠文字描述非常费力，而且 ConcurrentHashMap 非常复杂，了解上面那部分其实也就差不多了，没必要每个方法都要一行一行代码去扣，毕竟人的精力是有限的。