Java JUC包源码分析 - ConcurrentHashMap

ConcurrentHashMap之较于HashMap是保证了线程安全,其实现方式之精妙有很多值得学习的地方。同时,这篇文章也将持续更新,毕竟目前只研究了常用的几个操作,还有其他操作,等待我去深挖。

ConcurrentHashMap主要是依靠了CAS无锁算法和对操作的数组索引处的头节点加锁。

扩容操作是可以多线程协助进行,线程只要检测到在扩容就放下当前工作去协助扩容,扩容的优先级还挺高!毕竟耗时。

ps:昨晚写完,csdn在维护,居然无法发布。今天刚好补充了两个关键的方法进去。

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable {
    // 常量定义
    // 最大容量:2^30
    private static final int MAXIMUM_CAPACITY = 1 << 30;

    // 默认容量16
    private static final int DEFAULT_CAPACITY = 16;

    // 最大的数组大小
    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;

    // 从树转链表的节点树阀值
    static final int UNTREEIFY_THRESHOLD = 6;

    // 在链表转树时table的最小容量
    static final int MIN_TREEIFY_CAPACITY = 64;

    private static final int MIN_TRANSFER_STRIDE = 16;
    private static int RESIZE_STAMP_BITS = 16;
    // 2^16-1,help resize的最大线程数
    private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
    // 32-16=16,sizeCtl中记录size大小的偏移量
    private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

    // Node节点hash值的几个取值
    static final int MOVED     = -1; // 移动节点的hash值
    static final int TREEBIN   = -2; // 树根节点的hash值
    static final int RESERVED  = -3; // 短暂保留的节点的hash值
    static final int HASH_BITS = 0x7fffffff; // 普通节点hash值可用的位

    // CPU的核数
    static final int NCPU = Runtime.getRuntime().availableProcessors();

    // 成员变量
    // 存放键值对的数组
    transient volatile Node<K,V>[] table;

    // 下一个要使用的table,只有在扩容时不为null
    private transient volatile Node<K,V>[] nextTable;

    private transient volatile long baseCount;

    // 负数代表正在进行初始化或扩容操作
    // -1代表正在初始化
    // -N表示有N-1个线程正在进行扩容操作
    // 正数或0代表hash表还没有被初始化,这个数值表示初始化或下一次进行扩容的大小,类似于扩容 
    // 阈值。它的值始终是当前ConcurrentHashMap容量的0.75倍,这与loadfactor是对应的。
    // 实际容量>=sizeCtl,则扩容。
    private transient volatile int sizeCtl;

    /**
     * The next table index (plus one) to split while resizing.
     */
    private transient volatile int transferIndex;

    /**
     * Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
     */
    private transient volatile int cellsBusy;

    /**
     * Table of counter cells. When non-null, size is a power of 2.
     */
    private transient volatile CounterCell[] counterCells;

    // views
    private transient KeySetView<K,V> keySet;
    private transient ValuesView<K,V> values;
    private transient EntrySetView<K,V> entrySet;

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

    // 插入键值对,onlyIfAbsent参数是如果键值对已存在就不更新了。返回旧的value
    final V putVal(K key, V value, boolean onlyIfAbsent) {
        // 不允许传入的键值对为null
        if (key == null || value == null) throw new NullPointerException();
        // 计算hash值
        int hash = spread(key.hashCode());
        // 判断是否链表转树的变量
        int binCount = 0;
        // 遍历数组
        for (Node<K,V>[] tab = table;;) {
            // f是通过key找到的已存在的节点,n是数组长度,i是数组下标,fh是f的hash值
            Node<K,V> f; int n, i, fh;
            // 如果table还是null或者没有元素,则初始化一个table(懒加载),然后回到上面for循环
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            // 如果计算出来的索引对应的节点f在table中还是null就以cas算法在i处新建一个node插入
            // 因为此时还没有加锁,所以有可能cas失败,失败就重试(返回for循环);cas成功就退出
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node<K,V>(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }
            // 如果索引i处的节点正处于移动状态,也就是在扩容中,并且这个节点已经被处理过了。
            // 当前线程就先帮助扩容,等会在回到
            // for循环继续插入值
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            // 否则就是可以在数组索引i处遍历链表或树来替换value或新建一个链表中的节点
            else {
                // 旧的value
                V oldVal = null;
                // 对数组i处的节点加锁,避免多个线程同时操作数组i处的链表或树。换句话说只要多个 
                // 线程操作的不是同一个数组下标的位置就可以并发执行
                synchronized (f) {
                    // 先判断一下当前的头节点还是不是上面计算出来的节点,毕竟在加锁之前可能会有 
                    // 其他线程已经改了这个节点的值
                    if (tabAt(tab, i) == f) {
                    // 头节点没有遭到变动
                        // 如果头节点的hash值>=0,就去遍历链表
                        // 因为树的根节点的hash值是-2,在class TreeBin的初始化构造函数设置的
                        if (fh >= 0) {
                            // 链表节点个数置1,头节点占一个
                            binCount = 1;
                            // 遍历链表
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                // 如果发现key相同,则替换旧value
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    // 如果是onlyIfAbsent为true,就不替换,只返回旧value
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                // 遍历链表的下一个节点,如果下一个节点为null了,说明到达链表 
                                // 尾部还没有找到key相同的node,此时新建一个节点链接到链表末尾
                                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;
                            // 在红黑树里操作,替换旧值或者新建node
                            // 如果不为null,说明在树里找到相同的key了,返回了相应的节点
                            // 如果为null,说明没找到,在树里新插入这个节点进去
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                // 替换旧value
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                // 当进入链表或树遍历时,才binCount != 0
                if (binCount != 0) {
                    // 进入树里操作binCount=2,不会>=8。
                    // 所以只有进入了链表操作,并且链表节点>=8了才需要链表转树操作
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    // 返回旧值
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 计数器+1,因为能执行到说明是新建了一个节点,并不是替换value
        addCount(1L, binCount);
        return null;
    }

}

接下来看下帮助扩容和扩容的实现:扩容代码真的有点复杂,需要下更多的功夫来看懂,所以来日再更新

    final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
        Node<K,V>[] nextTab; int sc;
        // ForwardingNode是一个标志性的类,它只会出现在扩容操作期间并且bin不为null的时候,它是 
        // 连接两个table的标志,ForwardingNode节点会出现在每个bin的头节点处,指向nextTable
        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;
    }
    
    // 把bin放到新的table里面去,通过forwarding节点来控制多线程情况下是否跳过当前节点去扩容
    private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        int n = tab.length, stride;
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        // 如果是开始扩容的第一个线程到达,发现还没开始,就新建一个两倍容量的nextTable
        if (nextTab == null) {            // initiating
            try {
                @SuppressWarnings("unchecked")
                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 = nextTab;
            transferIndex = n;
        }
        int nextn = nextTab.length;
        // 新建标志类头节点,它的hash值为-1,其他值为null,fwd.nextTable=nextTab
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        // 并发扩容的关键属性,等于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;
            while (advance) {
                int nextIndex, nextBound;
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                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;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                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
                }
            }
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node<K,V> ln, hn;
                        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);
                            }
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            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;
                                }
                            }
                            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;
                        }
                    }
                }
            }
        }
    }

 

    原文作者:JUC
    原文地址: https://blog.csdn.net/qq_24219459/article/details/82708384
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞