"最近组内技术分享,我分到讲解 ConcurrentHashMap。 结合网上看到的一些资料,整理了点东西,分享一下。 主要参考下面文章: https://www.javadoop.com/post/hashmap# 初始化 https://www.cnblogs.com/nullzx/p/8647220.html 有解 .."

Java8 ConcurrentHashMap 源码解析

最近组内技术分享,我分到讲解 ConcurrentHashMap。
结合网上看到的一些资料,整理了点东西,分享一下。
主要参考下面文章:

https://www.javadoop.com/post/hashmap# 初始化
https://www.cnblogs.com/nullzx/p/8647220.html

有解释错误的地方,欢迎指摘。


Unsafe 里的 CAS 操作相关

首先介绍的 cas,主要是因为 8 的 ConcurrentHashMap 主要用的就是 cas。

CAS(compare-and-swap 比较交换) 操作。CAS 是一种低级别的、细粒度的技术, 它允许多个线程更新一个内存位置, 同时能够检测其他线程的冲突并进行恢复。它是许多高性能并发算法的基础。 CAS 是一些 CPU 直接支持的指令,操作都封装在 java 不公开的类库中,sun.misc.Unsafe。此类包含了对原子操作的封装, 具体用本地代码实现。本地的 C 代码直接利用到了硬件上的原子操作,在 Java 中无锁操作 CAS 基于以下 3 个方法实现。

//第一个参数o为给定对象,offset为对象内存的偏移量,通过这个偏移量迅速定位字段并设置或获取该字段的值,
//expected表示期望值,x表示要设置的值,下面3个方法都通过CAS原子指令执行操作。
public final native boolean compareAndSwapObject(Object o, long offset,Object expected, Object x);                                                                                                  
 
public final native boolean compareAndSwapInt(Object o, long offset,int expected,int x);
 
public final native boolean compareAndSwapLong(Object o, long offset,long expected,long x);

挂起与恢复

将一个线程进行挂起是通过 park 方法实现的,调用 park 后,线程将一直阻塞直到超时或者中断等条件出现。unpark 可以终止一个挂起的线程,使其恢复正常。Java 对线程的挂起操作被封装在 LockSupport 类中,LockSupport 类中有各种版本 pack 方法,其底层实现最终还是使用 Unsafe.park()方法和 Unsafe.unpark() 方法

ConcurrentHashMap

imagepng

1. 重要参数及初始化

桶的树化阈值:即 链表转成红黑树的阈值,在存储数据时,当链表长度 > 该值时,则将链表转换成红黑树

static final int TREEIFY_THRESHOLD = 8;

桶的链表还原阈值:即 红黑树转为链表的阈值,当在扩容(resize())时(此时 HashMap 的数据存储位置会重新计算),在重新计算存储位置后,当原有的红黑树内数量 < 6 时,则将 红黑树转换成链表

static final int UNTREEIFY_THRESHOLD = 6;

最小树形化容量阈值:即 当哈希表中的容量 > 该值时,才允许树形化链表 (即 将链表 转换成红黑树)
否则,若桶内元素太多时,则直接扩容,而不是树形化
为了避免进行扩容、树形化选择的冲突,这个值不能小于 4 * TREEIFY_THRESHOLD

static final int MIN_TREEIFY_CAPACITY = 64;

默认加载因子

private static final float LOAD_FACTOR = 0.75f;
static final int MOVED     = -1; // hash值是-1,表示这是一个forwardNode节点
static final int TREEBIN   = -2; // hash值是-2  表示这时一个TreeBin节点

和 HashMap 中的语义一样,代表整个哈希表。

transient volatile Node<K,V>[] table;

这是一个连接表,用于哈希表扩容,扩容完成后会被重置为 null。

/**
* The next table to use; non-null only while resizing.
*/
private transient volatile Node<K,V>[] nextTable;

该属性保存着整个哈希表中存储的所有的结点的个数总和,有点类似于 HashMap 的 size 属性。

private transient volatile long baseCount;

这是一个重要的属性,无论是初始化哈希表,还是扩容 rehash 的过程,都是需要依赖这个关键属性的。该属性有以下几种取值:

private transient volatile int sizeCtl;

线程迁移 bin 的起始位置,CAS(transferIndex) 成功者可迁移 transferIndex 前置 stride 个 bin(见 transfer)

private transient volatile int transferIndex;

1.1 初始化

// 这构造函数里,什么都不干
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);
    }

通过提供初始容量,计算了 sizeCtl,sizeCtl = 【 (1.5 * initialCapacity + 1),然后向上取最近的 2 的 n 次方】。如 initialCapacity 为 10,那么得到 sizeCtl 为 16,如果 initialCapacity 为 11,得到 sizeCtl 为 32。

1.2 重要的内部类

Node
static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        volatile V val; // Java8增加volatile,保证可见性
        volatile Node<K,V> next;
 
        Node(inthash, 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; }
        // HashMap调用Objects.hashCode(),最终也是调用Object.hashCode();效果一样
        public final int hashCode()   { returnkey.hashCode() ^ val.hashCode(); }
        public final String toString(){ returnkey + "=" + val; }
        public final V setValue(V value) { // 不允许修改value值,HashMap允许
            throw new UnsupportedOperationException();
        }
        // HashMap使用if (o == this),且嵌套if;concurrent使用&&
        public final boolean equals(Object o) {
            Object k, v, u; Map.Entry<?,?> e;
            return ((oinstanceof 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(inth, Object k) { // 增加find方法辅助get方法
            Node<K,V> e = this;
            if (k != null) {
                do {
                    K ek;
                    if (e.hash == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        returne;
                } while ((e = e.next) != null);
            }
            return null;
        }
    }
TreeNode
// Nodes for use in TreeBins,链表>8,才可能转为TreeNode.
// HashMap的TreeNode继承至LinkedHashMap.Entry;而这里继承至自己实现的Node,将带有next指针,便于treebin访问。
    static final class TreeNode<K,V> extends Node<K,V> { 
        TreeNode<K,V> parent;  // red-black tree links
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;
 
        TreeNode(inthash, K key, V val, Node<K,V> next,
                 TreeNode<K,V> parent) {
            super(hash, key, val, next);
            this.parent = parent;
        }
 
        Node<K,V> find(inth, Object k) {
            return findTreeNode(h, k, null);
        }
 
        /**
         * Returns the TreeNode (or null if not found) for the given key
         * starting at given root.
         */ // 查找hash为h,key为k的节点
        final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
            if (k != null) { // 比HMap增加判空
                TreeNode<K,V> p = this;
                do  {
                    intph, dir; K pk; TreeNode<K,V> q;
                    TreeNode<K,V> pl = p.left, pr = p.right;
                    if ((ph = p.hash) > h)
                        p = pl;
                    elseif (ph < h)
                        p = pr;
                    elseif ((pk = p.key) == k || (pk != null && k.equals(pk)))
                        returnp;
                    elseif (pl == null)
                        p = pr;
                    elseif (pr == null)
                        p = pl;
                    elseif ((kc != null ||
                              (kc = comparableClassFor(k)) != null) &&
                             (dir = compareComparables(kc, k, pk)) != 0)
                        p = (dir < 0) ? pl : pr;
                    elseif ((q = pr.findTreeNode(h, k, kc)) != null)
                        returnq;
                    else
                        p = pl;
                } while (p != null);
            }
            return null;
        }
    }

树节点类,另外一个核心的数据结构。当链表长度过长的时候,会转换为 TreeNode。但是与 HashMap 不相同的是,它并不是直接转换为红黑树,而是把这些结点包装成 TreeNode 放在 TreeBin 对象中,由 TreeBin 完成对红黑树的包装。而且 TreeNode 在 ConcurrentHashMap 集成自 Node 类,而并非 HashMap 中的集成自 LinkedHashMap.Entry<K,V> 类,也就是说 TreeNode 带有 next 指针,这样做的目的是方便基于 TreeBin 的访问

TreeBin
TreeBin(TreeNode<K,V> b) {
    super(TREEBIN, null, null, null);//hash值为常量TREEBIN=-2,表示roots of trees
    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;
            inth = x.hash;
            Class<?> kc = null;
            for (TreeNode<K,V> p = r;;) {
                intdir, ph;
                K pk = p.key;
                if ((ph = p.hash) > h)
                    dir = -1;
                elseif (ph < h)
                    dir = 1;
                elseif ((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
// A node inserted at head of bins during transfer operations.连接两个table
// 并不是我们传统的包含key-value的节点,只是一个标志节点,并且指向nextTable,提供find方法而已。生命周期:仅存活于扩容操作且bin不为null时,一定会出现在每个bin的首位。
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); // 此节点hash=-1,key、value、next均为null
        this.nextTable = tab;
    }
 
    //进行get操作的线程若发现槽中的节点为ForwordingNode类型
    //说明该桶中所有结点已迁移完成,会调用ForwordingNode的find方法在新表中进行查找
    Node<K,V> find(int h, Object k) {
        // 查nextTable节点,outer避免深度递归
        outer: for (Node<K,V>[] tab = nextTable;;) {
         // n表示新表的长度
            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 (;;) { // CAS算法多和死循环搭配!直到查到或null
                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;
            }
        }
    }
}

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

1.3 三个核心方法

 @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);
    }

2. put 过程分析

2.1 初始化数组:initTable

主要就是初始化一个合适大小的数组,然后会设置 sizeCtl。

初始化方法中的并发问题是通过对 sizeCtl 进行一个 CAS 操作来控制的。

该方法的核心思想就是,只允许一个线程对表进行初始化,如果不巧有其他线程进来了,那么会让其他线程交出 CPU 等待下次系统调度。这样,保证了表同时只会被一个线程初始化。

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        // sizeCtl 小于零说明已经有线程正在进行初始化操作
        // 当前线程应该放弃 CPU 的使用
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        // CAS 一下,将 sizeCtl 设置为 -1,代表抢到了锁
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    // sc 大于零说明容量已经初始化了,否则使用默认 DEFAULT_CAPACITY 默认初始容量是 16
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    // 初始化数组,长度为 16 或初始化时提供的长度
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    // 将这个数组赋值给 table,table 是 volatile 的
                    table = tab = nt;
                    // 如果 n 为 16 的话,那么这里 sc = 12
                    // 其实就是 0.75 * n
                    sc = n - (n >>> 2);
                }
            } finally {
                // 设置 sizeCtl 为 sc,我们就当是 12 吧
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

2.2 链表转红黑树: treeifyBin

treeifyBin 不一定就会进行红黑树转换,也可能是仅仅做数组扩容

private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    if (tab != null) {
        // MIN_TREEIFY_CAPACITY 为 64
        // 所以,如果数组长度小于 64 的时候,其实也就是 32 或者 16 或者更小的时候,会进行数组扩容
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
            // 后面我们再详细分析这个方法
            tryPresize(n << 1);
        // b 是头结点
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            // 加锁
            synchronized (b) {

                if (tabAt(tab, index) == b) {
                    // 下面就是遍历链表,建立一颗红黑树
                    TreeNode<K,V> hd = null, tl = null;
                    for (Node<K,V> e = b; e != null; e = e.next) {
                        TreeNode<K,V> p =
                            new TreeNode<K,V>(e.hash, e.key, e.val,
                                              null, null);
                        if ((p.prev = tl) == null)
                            hd = p;
                        else
                            tl.next = p;
                        tl = p;
                    }
                    // 将红黑树设置到数组相应位置中
                    setTabAt(tab, index, new TreeBin<K,V>(hd));
                }
            }
        }
    }
}

2.3 扩容:tryPresize

扩容也是做翻倍扩容的,扩容后数组容量为原来的 2 倍

// 首先要说明的是,方法参数 size 传进来的时候就已经翻了倍了
private final void tryPresize(int size) {
    // c:size 的 1.5 倍,再加 1,再往上取最近的 2 的 n 次方。
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
        tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    while ((sc = sizeCtl) >= 0) {
        Node<K,V>[] tab = table; int n;

        // 这个 if 分支和之前说的初始化数组的代码基本上是一样的,在这里,我们可以不用管这块代码
        if (tab == null || (n = tab.length) == 0) {
            n = (sc > c) ? sc : c;
            if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                    if (table == tab) {
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = nt;
                        sc = n - (n >>> 2); // 0.75 * n
                    }
                } finally {
                    sizeCtl = sc;
                }
            }
        }
        else if (c <= sc || n >= MAXIMUM_CAPACITY)
            break;
        else if (tab == table) {
            int rs = resizeStamp(n);
            if (sc < 0) {
                Node<K,V>[] nt;
                //RESIZE_STAMP_SHIFT=16,MAX_RESIZERS=2^15-1
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                // 2. 用 CAS 将 sizeCtl 加 1,然后执行 transfer 方法
                //    此时 nextTab 不为 null
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2)
            //   计算出来结果是一个比较大的负数
            //  调用 transfer 方法,此时 nextTab 参数为 null
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
        }
    }
}

2.4 数据迁移:transfer

我们在 putVal 方法中遍历整个 hash 表的桶结点,如果遇到 hash 值等于 MOVED,说明已经有线程正在扩容 rehash 操作,整体上还未完成,不过我们要插入的桶的位置已经完成了所有节点的迁移。

由于检测到当前哈希表正在扩容,于是让当前线程去协助扩容。

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
        //返回一个 16 位长度的扩容校验标识
        int rs = resizeStamp(tab.length);
        while (nextTab == nextTable && table == tab &&
               (sc = sizeCtl) < 0) {
            //sizeCtl 如果处于扩容状态的话
            //前 16 位是数据校验标识,后 16 位是当前正在扩容的线程总数
            //这里判断校验标识是否相等,如果校验符不等或者扩容操作已经完成了,直接退出循环,不用协助它们扩容了
            // 如果 sizeCtl 无符号右移  16 不等于 rs ( sc前 16 位如果不等于标识符,则标识符变化了)
            // 或者 sizeCtl == rs + 1  (扩容结束了,不再有线程进行扩容)(默认第一个线程设置 sc ==rs 左移 16 位 + 2,当第一个线程结束扩容了,就会将 sc 减一。这个时候,sc 就等于 rs + 1)
            // 或者 sizeCtl == rs + 65535  (如果达到最大帮助线程的数量,即 65535)
            // 或者转移下标正在调整 (扩容结束)
            // 结束循环,返回 table

            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                sc == rs + MAX_RESIZERS || transferIndex <= 0)
                break;
            //否则调用 transfer 帮助它们进行扩容
            //sc + 1 标识增加了一个线程进行扩容
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}

阅读源码之前,先要理解并发操作的机制。原数组长度为 n,所以我们有 n 个迁移任务,让每个线程每次负责一个小任务是最简单的,每做完一个任务再检测是否有其他没做完的任务,帮助迁移就可以了,而 Doug Lea 使用了一个 stride,简单理解就是步长,每个线程每次负责迁移其中的一部分,如每次迁移 16 个小任务。所以,我们就需要一个全局的调度者来安排哪个线程执行哪几个任务,这个就是属性 transferIndex 的作用。

第一个发起数据迁移的线程会将 transferIndex 指向原数组最后的位置,然后从后往前的 stride 个任务属于第一个线程,然后将 transferIndex 指向新的位置,再往前的 stride 个任务属于第二个线程,依此类推。当然,这里说的第二个线程不是真的一定指代了第二个线程,也可以是同一个线程,这个读者应该能理解吧。其实就是将一个大的迁移任务分为了一个个任务包。

ConcurrentHashMap 无锁多线程扩容,减少扩容时的时间消耗。
transfer 扩容操作:单线程构建两倍容量的 nextTable;允许多线程复制原 table 元素到 nextTable。

  1. 为每个内核均分任务,并保证其不小于 16;
  2. 若 nextTab 为 null,则初始化其为原 table 的 2 倍;
  3. 死循环遍历,直到 finishing。

这个方法的核心在于 sizeCtl 值的操作,首先将其设置为一个负数,然后执行 transfer(tab, null),再下一个循环将 sizeCtl 加 1,并执行 transfer(tab, nt),之后可能是继续 sizeCtl 加 1,并执行 transfer(tab, nt)。

所以,可能的操作就是执行 1 次 transfer(tab, null) + 多次 transfer(tab, nt),这里怎么结束循环的需要看完 transfer 源码才清楚。

以上说的都是单线程,多线程又是如何实现的呢?

遍历到 ForwardingNode 节点 ((fh = f.hash) == MOVED),说明此节点被处理过了,直接跳过。这是控制并发扩容的核心。由于给节点上了锁,只允许当前线程完成此节点的操作,处理完毕后,将对应值设为 ForwardingNode(fwd),其他线程看到 forward,直接向后遍历。如此便完成了多线程的复制工作,也解决了线程安全问题。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;

    // stride 在单核下直接等于 n,多核模式下为 (n>>>3)/NCPU,最小值是 16
    // stride 可以理解为”步长“,有 n 个位置是需要进行迁移的,
    //   将这 n 个任务分为多个任务包,每个任务包有 stride 个任务
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
        stride = MIN_TRANSFER_STRIDE; // subdivide range

    // 如果 nextTab 为 null,先进行一次初始化
    //    前面我们说了,外围会保证第一个发起迁移的线程调用此方法时,参数 nextTab 为 null
    //       之后参与迁移的线程调用此方法时,nextTab 不会为 null
    if (nextTab == null) {
        try {
            // 容量翻倍
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        // nextTable 是 ConcurrentHashMap 中的属性
        nextTable = nextTab;
        // transferIndex 也是 ConcurrentHashMap 的属性,用于控制迁移的位置
        transferIndex = n;
    }

    int nextn = nextTab.length;

    // ForwardingNode 翻译过来就是正在被迁移的 Node
    // 这个构造方法会生成一个Node,key、value 和 next 都为 null,关键是 hash 为 MOVED
    // 后面我们会看到,原数组中位置 i 处的节点完成迁移工作后,
    //    就会将位置 i 处设置为这个 ForwardingNode,用来告诉其他线程该位置已经处理过了
    //    所以它其实相当于是一个标志。
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);


    // advance 指的是做完了一个位置的迁移工作,可以准备做下一个位置的了
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab

    /*
     * 下面这个 for 循环,最难理解的在前面,而要看懂它们,应该先看懂后面的,然后再倒回来看
     * 
     */

    // i 是位置索引,bound 是边界,注意是从后往前
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;

        // 下面这个 while 真的是不好理解
        // advance 为 true 表示可以进行下一个位置的迁移了
        //   简单理解结局:i 指向了 transferIndex,bound 指向了 transferIndex-stride
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing)
                advance = false;

            // 将 transferIndex 值赋给 nextIndex
            // 这里 transferIndex 一旦小于等于 0,说明原数组的所有位置都有相应的线程去处理了
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                // 看括号中的代码,nextBound 是这次迁移任务的边界,注意,是从后往前
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) {
                // 所有的迁移操作已经完成
                nextTable = null;
                // 将新的 nextTab 赋值给 table 属性,完成迁移
                table = nextTab;
                // 重新计算 sizeCtl:n 是原数组长度,所以 sizeCtl 得出的值将是新数组长度的 0.75 倍
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }

            // 之前我们说过,sizeCtl 在迁移前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2
            // 然后,每有一个线程参与迁移就会将 sizeCtl 加 1,
            // 这里使用 CAS 操作对 sizeCtl 进行减 1,代表做完了属于自己的任务
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                // 任务结束,方法退出
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                    return;

                // 到这里,说明 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT,
                // 也就是说,所有的迁移任务都做完了,也就会进入到上面的 if(finishing){} 分支了
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        // 如果位置 i 处是空的,没有任何节点,那么放入刚刚初始化的 ForwardingNode ”空节点“
        else if ((f = tabAt(tab, i)) == null)
            advance = casTabAt(tab, i, null, fwd);
        // 该位置处是一个 ForwardingNode,代表该位置已经迁移过了
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            // 对数组该位置处的结点加锁,开始处理数组该位置处的迁移工作
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    // 头结点的 hash 大于 0,说明是链表的 Node 节点
                    if (fh >= 0) {
                        // 下面这一块和 Java7 中的 ConcurrentHashMap 迁移是差不多的,
                        // 需要将链表一分为二,
                        //   找到原链表中的 lastRun,然后 lastRun 及其之后的节点是一起进行迁移的
                        //   lastRun 之前的节点需要进行克隆,然后分到两个链表中
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            int b = p.hash & n;
                            if (b != runBit) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun;
                            ln = null;
                        }
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            if ((ph & n) == 0)
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else
                                hn = new Node<K,V>(ph, pk, pv, hn);
                        }
                        // 其中的一个链表放在新数组的位置 i
                        setTabAt(nextTab, i, ln);
                        // 另一个链表放在新数组的位置 i+n
                        setTabAt(nextTab, i + n, hn);
                        // 将原数组该位置处设置为 fwd,代表该位置已经处理完毕,
                        //    其他线程一旦看到该位置的 hash 值为 MOVED,就不会进行迁移了
                        setTabAt(tab, i, fwd);
                        // advance 设置为 true,代表该位置已经迁移完毕
                        advance = true;
                    }
                    else if (f instanceof TreeBin) {
                        // 红黑树的迁移
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        TreeNode<K,V> lo = null, loTail = null;
                        TreeNode<K,V> hi = null, hiTail = null;
                        int lc = 0, hc = 0;
                        for (Node<K,V> e = t.first; e != null; e = e.next) {
                            int h = e.hash;
                            TreeNode<K,V> p = new TreeNode<K,V>
                                (h, e.key, e.val, null, null);
                            if ((h & n) == 0) {
                                if ((p.prev = loTail) == null)
                                    lo = p;
                                else
                                    loTail.next = p;
                                loTail = p;
                                ++lc;
                            }
                            else {
                                if ((p.prev = hiTail) == null)
                                    hi = p;
                                else
                                    hiTail.next = p;
                                hiTail = p;
                                ++hc;
                            }
                        }
                        // 如果一分为二后,节点数<= 6,那么将红黑树转换回链表
                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                            (hc != 0) ? new TreeBin<K,V>(lo) : t;
                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                            (lc != 0) ? new TreeBin<K,V>(hi) : t;

                        // 将 ln 放置在新数组的位置 i
                        setTabAt(nextTab, i, ln);
                        // 将 hn 放置在新数组的位置 i+n
                        setTabAt(nextTab, i + n, hn);
                        // 将原数组该位置处设置为 fwd,代表该位置已经处理完毕,
                        //    其他线程一旦看到该位置的 hash 值为 MOVED,就不会进行迁移了
                        setTabAt(tab, i, fwd);
                        // advance 设置为 true,代表该位置已经迁移完毕
                        advance = true;
                    }
                }
            }
        }
    }
}

2.5 put 方法

在多线程中可能有以下两个情况

  1. 如果一个或多个线程正在对 ConcurrentHashMap 进行扩容操作,当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到,是因为 transfer 方法中在空结点上插入 forward 节点,如果检测到需要插入的位置被 forward 节点占有,就帮助进行扩容;
  2. 如果检测到要插入的节点是非空且不是 forward 节点,就对这个节点加锁,这样就保证了线程安全。尽管这个有一些影响效率,但是还是会比 hashTable 的 synchronized 要好得多。

整体流程

  1. 校验 key value 值,都不能是 null。这点和 HashMap 不同。
  2. 得到 key 的 hash 值。
  3. 死循环并更新 tab 变量的值。
  4. 如果容器没有初始化,则初始化。调用 initTable 方法。该方法通过一个变量 + CAS 来控制并发。稍后我们分析源码。
  5. 根据 hash 值找到数组下标,如果对应的位置为空,就创建一个 Node 对象用 CAS 方式添加到容器。并跳出循环。
  6. 如果 hash 冲突,也就是对应的位置不为 null,则判断该槽是否被扩容了(-1 表示被扩容了),如果被扩容了,返回新的数组。
  7. 如果 hash 冲突 且 hash 值不是 -1,表示没有被扩容。则进行链表操作或者红黑树操作,注意,这里的 f 头节点被锁住了,保证了同时只有一个线程修改链表。防止出现链表成环。
  8. 和 HashMap 一样,如果链表树超过 8,则修改链表为红黑树。
  9. 将数组加 1(CAS 方式),如果需要扩容,则调用 transfer 方法进行移动和重新散列,该方法中,如果是槽中只有单个节点,则使用 CAS 直接插入,如果不是,则使用 synchronized 进行同步,防止并发成环。
public V put(K key, V value) {
    return putVal(key, value, false);
}

final V putVal(K key, V value, boolean onlyIfAbsent) {
    //对传入的参数进行合法性判断 不允许 key或value为null
    if (key == null || value == null) throw new NullPointerException();
    // 计算键所对应的 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 = initTable();

        // 找该 hash 值对应的数组下标,得到第一个节点 f
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 如果数组该位置为空,
            //    用一次 CAS 操作将这个新值放入其中即可,这个 put 操作差不多就结束了,可以拉到最后面了
            //          如果 CAS 失败,那就是有并发操作,进到下一个循环就好了
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        // hash 居然可以等于 MOVED(MOVED = -1;  hash for forwarding nodes),这个需要到后面才能看明白,不过从名字上也能猜到,肯定是因为在扩容
        else if ((fh = f.hash) == MOVED)
            // 帮助数据迁移,这个等到看完数据迁移部分的介绍后,再理解这个就很简单了
            tab = helpTransfer(tab, f);

        else { // 到这里就是说,f 是该位置的头结点,而且不为空

            V oldVal = null;
            // 获取数组该位置的头结点的监视器锁
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) { // 头结点的 hash 值大于 0,说明是链表
                        // 用于累加,记录链表的长度
                        binCount = 1;
                        // 遍历链表
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            // 如果发现了"相等"的 key,判断是否要进行值覆盖,然后也就可以 break 了
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent) //仅putIfAbsent()方法中onlyIfAbsent为true
                                    e.val = value; //putIfAbsent()包含key则返回get,否则put并返回  
                                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;
                        }
                    }
                }
            }

            //binCount != 0 说明向链表或者红黑树中添加或修改一个节点成功
            //binCount  == 0 说明 put 操作将一个新节点添加成为某个桶的首节点
            if (binCount != 0) {
                // 判断是否要将链表转换为红黑树,临界值和 HashMap 一样,也是 8
                if (binCount >= TREEIFY_THRESHOLD)
                    // 这个方法和 HashMap 中稍微有一点点不同,那就是它不是一定会进行红黑树转换,
                    // 如果当前数组的长度小于 64,那么会选择进行数组扩容,而不是转换为红黑树
                    //    具体源码我们就不看了,扩容部分后面说
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    // CAS 式更新baseCount,并判断是否需要扩容
    addCount(1L, binCount);
    //程序走到这一步说明此次 put 操作是一个添加操作,否则早就 return 返回了
    return null;
}

3.get 过程分析

  1. 计算 hash 值
  2. 根据 hash 值找到数组对应位置: (n - 1) & h
  3. 根据该位置处结点性质进行相应查找
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    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;
        }
        // 如果头结点的 hash 小于 0,说明 正在扩容,或者该位置是红黑树
        else if (eh < 0)
            // 参考 ForwardingNode.find(int h, Object k) 和 TreeBin.find(int h, Object k)
            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;
}

看完源码,通过一张图直观展示一下扩容操作

看懂上面了,下面就不用看了:

imagepng

我们这里假设旧表的长度是 8(实际上代码中表的最小长度也是 16,这样假设是为了画图方便),图中的数字表示结点的 hash 值。

从图中我们可以看出,扩容后表的长度变成了 16。我们现在要对比观察扩容前后每个结点的位置,显然可以得到一个有意思的结论:每个结点在扩容后要么留在了新表原来的位置上,要么去了新表 “原位置 +8”的位置上,而 8 就是旧表的长度。比如扩容前 3 号槽有 [3,11,19] 结点,扩容后 [3,19] 结点依然留在了原 3 号位置,而节点 [11] 去了“原位置 3 + 8 = 11”的位置。计算新表中槽的位置有很巧妙的方法,有兴趣的同学可以参照 transfer 函数的源代码。

扩容长度翻倍,并且扩容后长度仍然是 2 的整数次幂的特性在多线程扩容有很大的优势。原表中不同桶上的结点,在新表上一定不会分配到相同位置的槽上。我们可以让不同线程负责原表不同位置的桶中所有结点的迁移,这样两个线程的迁移操作是不会相互干扰的。

比如我们可以让一个线程负责原表中 3 号桶中所有结点的迁移,另一个线程负责原表中 4 号桶所有结点的迁移。原表中 3 号位置上的结点只能迁移到新表 3 号位置或 11 号位置上,绝对不会映射到其它位置上。而 4 号位置上的结点只能迁移到新表 4 号位置或 12 号位置上,所以在迁移结点的过程中,两个线程就不必在新表的对应槽上加锁了。

  • Java

    Java 是一种可以撰写跨平台应用软件的面向对象的程序设计语言,是由 Sun Microsystems 公司于 1995 年 5 月推出的。Java 技术具有卓越的通用性、高效性、平台移植性和安全性。

    2211 引用 • 7514 回帖 • 973 关注
回帖   
请输入回帖内容...