JDK8中ConcurrentHashMap源码阅读

怼烎@ 2022-10-22 12:53 265阅读 0赞

# 数据结构 #

常量：

private static final int MAXIMUM_CAPACITY = 1 << 30;
     // 元素的最大个数
    private static final int DEFAULT_CAPACITY = 16;
     // 数组的默认初始长度，必须是2的幂次方
    static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
     // 数组的最大长度
    private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
     // 默认支持的并发数
    private static final float LOAD_FACTOR = 0.75f;
     // 负载因子
    static final int TREEIFY_THRESHOLD = 8;
     // 链表转红黑树阀值,> 8 链表转换为红黑树
    static final int UNTREEIFY_THRESHOLD = 6;
     // 红黑树转链表阀值，小于等于6
    static final int MIN_TREEIFY_CAPACITY = 64;
     // 链表转红黑树数组长度阈值
    private static final int MIN_TRANSFER_STRIDE = 16;
    private static int RESIZE_STAMP_BITS = 16;
    private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
     // help resize的最大线程数
    private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
     // 32-16=16，sizeCtl中记录size大小的偏移量
    static final int MOVED     = -1;
     // forwarding nodes的hash值
    static final int TREEBIN   = -2;
     // 树根节点的hash值
    static final int RESERVED  = -3;
     // ReservationNode的hash值
    static final int NCPU = Runtime.getRuntime().availableProcessors(); // CPU核数

属性：

transient volatile Node<K,V>[] table;
     // 用来存放Node节点数据的，默认为null，默认大小为16的数组，每次扩容时大小总是2的幂次方；
    private transient volatile Node<K,V>[] nextTable;
     // 扩容时新生成的数据，数组为table的两倍；
    private transient volatile long baseCount;
     // 计数器，与counterCells之和就为size
    
    // sizeCtl为0 代表数组没有初始化
    // sizeCtl为正数 数组已经初始化，记录的是数组的扩容阈值，如果没有初始化，记录的是数组初始容量
    // sizeCtl 为-1 代表数组正在初始化
    // sizeCtl 为负数，并且不是-1，表示数组正在扩容
    private transient volatile int sizeCtl;
    private transient volatile CounterCell[] counterCells;

内部类：

// 链表节点
    static class Node<K,V> implements Map.Entry<K,V> { 
        final int hash;
        final K key;
        volatile V val;
        volatile Node<K,V> next;
    
    // 红黑树节点
    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;
    
    // 用来管理和构造红黑树节点
    static final class TreeBin<K,V> extends Node<K,V> { 
        TreeNode<K,V> root;
        volatile TreeNode<K,V> first;
        volatile Thread waiter;
        volatile int lockState;
    
    // 扩容时的辅助类，hash=-1，其他值为null
    static final class ForwardingNode<K,V> extends Node<K,V> { 
        final Node<K,V>[] nextTable;

# 构造方法 #

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));
     // 取最近的2的幂次方数，不论当前数是否是2的幂次方
        this.sizeCtl = cap;
    }

ConcurrentHashMap有两种构造方法：

1.  空构造方法，啥也不干，sizeCtl=0。
2.  有参数的构造方法会初始化sizeCtl为一个2的幂次方正数，记录的是数组初始容量  
    。

> 注意如果传入初始化容量参数为32，那么实际上数组的初始容量为64，取32最近的2的幂次方数，而不是32，这是ConcurrentHashMap特有的，如果是HashMap传入32那么数组的初始容量就是32。

# put() #

public V put(K key, V value) { 
        return putVal(key, value, false);
    }
    final V putVal(K key, V value, boolean onlyIfAbsent) { 
        if (key == null || value == null) throw new NullPointerException(); // key和value都不支持null
        // 得到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，这个需要到后面才能看明白，不过从名字上也能猜到，肯定是因为在扩容
            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)
                                        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;
                            }
                        }
                    }
                }
                // binCount != 0 说明上面在做链表操作
                if (binCount != 0) { 
                    // 判断是否要将链表转换为红黑树，临界值和HashMap一样，也是8
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // size+1 和 扩容
        addCount(1L, binCount);
        return null;
    }

## initTable() ##

初始化数组：

private final Node<K,V>[] initTable() { 
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) { 
            if ((sc = sizeCtl) < 0)
     // sizeCtl<0说明有一个线程在进行初始化
                Thread.yield(); // 让出CPU执行权，让其他或者自己的线程执行（也就是谁先抢到谁执行）
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { 
                try { 
                    if ((tab = table) == null || tab.length == 0) { 
     // 双重检验是为了不重复创建 因为一个线程创建完后，会更改sizeCtl值
                        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);
     // sc=n*0.75，位操作更快
                    }
                } finally { 
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

## treeifyBin() ##

链表转红黑树，与HashMap中要满足的条件是一致的，链表长度大于8+数组的长度>=64。

private final void treeifyBin(Node<K,V>[] tab, int index) { 
        Node<K,V> b; int n, sc;
        if (tab != null) { 
            if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
                tryPresize(n << 1);
     // 数组长度小于64进行扩容
            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));
                    }
                }
            }
        }
    }

## addCount() ##

private final void addCount(long x, int check) { 
        CounterCell[] as; long b, s;
    	// size+1
        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();
        }
    	// 扩容
        if (check >= 0) { 
            Node<K,V>[] tab, nt; int n, sc;
    		// s=元素的个数，sc为扩容阈值，n为数组的长度
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) { 
                int rs = resizeStamp(n);
                if (sc < 0) { 
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                        transfer(tab, nt);
                }
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    transfer(tab, null);
     // 扩容
                s = sumCount();
            }
        }
    }

addCount()方法的逻辑包括两部分：

1.  size++，这部分看下下面的理论逻辑，就不展开了。
2.  扩容，看后面的扩容方法transfer()。

size++里面使用的是LongAdder，基于base+数组实现，这个方法就不展开了，有兴趣的可以去看看LongAdder的源码。

这里阐述下为什么不使用AtomicInteger而使用LongAdder？

先来看下AtomicInteger的缺点：

1.  在高并发下，AtomicInteger内部使用了自旋+CAS，当竞争CAS线程过多时，会导致大量的线程自旋，可能会导致CPU使用率上升。
2.  AtomicInteger内部使用了一个volatile修饰的int类型的变量，在多核CPU下会导致这个int变量在各个高速缓存中存有副本，多线程下这个缓存行就会不停的失效，产生伪共享。

LongAdder中引入了分段累加的概念，内部一共有两个参数参与计数：第一个叫作 base，它是一个变量，第二个是Cell\[\]，是一个数组，数组的长度可以扩容到CPU的核数。

其中的base是用在竞争不激烈的情况下的，可以直接把累加结果改到base变量上。那么，当竞争激烈的时候，就要用到我们的 Cell\[\] 数组了。一旦竞争激烈，各个线程会分散累加到自己所对应的那个Cell\[\]数组的某一个对象中，而不会大家共用同一个。

这样一来，LongAdder会把不同线程对应到不同的Cell上进行修改，降低了冲突的概率，这是一种分段的理念，提高了并发性，这就和Java 7的 ConcurrentHashMap的16个Segment的思想类似。

最后怎么算size呢？base+数组中Cell.value之和。

分段锁的概念只是降低了线程之间的竞争，伪共享问题怎么解决呢？从源码可以发现Cell这个类使用了@sun.misc.Contended注解修饰，这个注解就能保证缓存行对齐，解决伪共享问题，而不再需要像Disruptor  
那样前面填充7个long类型的变量，后面填充7个long类型的变量了。

## transfer() ##

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { 
        int n = tab.length, stride;
        // 每核处理的量小于16，则强制赋值16
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        if (nextTab == null) {             // initiating
            try { 
                @SuppressWarnings("unchecked")
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];        //构建一个nextTable对象，其容量为原来容量的两倍
                nextTab = nt;
            } catch (Throwable ex) {       // try to cope with OOME
                sizeCtl = Integer.MAX_VALUE;
                return;
            }
            nextTable = nextTab;
            transferIndex = n;
        }
        int nextn = nextTab.length;
        // 连接点指针，用于标志位（fwd的hash值为-1，fwd.nextTable=nextTab）
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        // 当advance == true时，表明该节点已经处理过了
        boolean advance = true;
        boolean finishing = false; // to ensure sweep before committing nextTab
        for (int i = 0, bound = 0;;) { 
            Node<K,V> f; int fh;
            // 控制 --i ,遍历原hash表中的节点
            while (advance) { 
                int nextIndex, nextBound;
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) { 
                    i = -1;
                    advance = false;
                }
                // 用CAS计算得到的transferIndex
                else if (U.compareAndSwapInt
                        (this, TRANSFERINDEX, nextIndex,
                                nextBound = (nextIndex > stride ?
                                        nextIndex - stride : 0))) { 
                    bound = nextBound;
                    i = nextIndex - 1;
                    advance = false;
                }
            }
            if (i < 0 || i >= n || i + n >= nextn) { 
                int sc;
                // 已经完成所有节点复制了
                if (finishing) { 
                    nextTable = null;
                    table = nextTab;        // table 指向nextTable
                    sizeCtl = (n << 1) - (n >>> 1);     // sizeCtl阈值为原来的1.5倍
                    return;     // 跳出死循环，
                }
                // CAS 更扩容阈值，在这里面sizectl值减一，说明新加入一个线程参与到扩容操作
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { 
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    finishing = advance = true;
                    i = n; // recheck before commit
                }
            }
            // 遍历的节点为null，则放入到ForwardingNode 指针节点
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            // f.hash == -1 表示遍历到了ForwardingNode节点，意味着该节点已经处理过了
            // 这里是控制并发扩容的核心
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else { 
                // 节点加锁
                synchronized (f) { 
                    // 节点复制工作
                    if (tabAt(tab, i) == f) { 
                        Node<K,V> ln, hn;
                        // fh >= 0 ,表示为链表节点
                        if (fh >= 0) { 
                            // 构造两个链表 一个是原链表 另一个是原链表的反序排列
                            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);
                            }
                            // 在nextTable i 位置处插上链表
                            setTabAt(nextTab, i, ln);
                            // 在nextTable i + n 位置处插上链表
                            setTabAt(nextTab, i + n, hn);
                            // 在table i 位置处插上ForwardingNode 表示该节点已经处理过了
                            setTabAt(tab, i, fwd);
                            // advance = true 可以执行--i动作，遍历节点
                            advance = true;
                        }
                        // 如果是TreeBin，则按照红黑树进行处理，处理逻辑与上面一致
                        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;
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

## helpTransfer() ##

helpTransfer()方法为协助扩容方法，当调用该方法的时候，nextTable一定已经创建了，所以该方法主要则是进行复制工作。如下：

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) { 
            int rs = resizeStamp(tab.length);
            while (nextTab == nextTable && table == tab &&
                    (sc = sizeCtl) < 0) { 
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || transferIndex <= 0)
                    break;
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { 
                    transfer(tab, nextTab);
                    break;
                }
            }
            return nextTab;
        }
        return table;
    }

# get() #

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