ConcurrentHashMap源码理解——第一篇

• • 数据结构
  • put过程
  • 课外知识

ConcurrentHashMap东西太多了，这一篇先说说它的数据结构和放元素的过程！
源码依赖于jdk1.8版本

先给出结论：数组+链表+红黑树
首先我们先看看它有哪些属性，以方便后面的理解。

transient volatile Node[] table;


private transient volatile Node[] nextTable;


private transient volatile long baseCount;


private transient volatile int sizeCtl;


private transient volatile int transferIndex;


private transient volatile int cellsBusy;


private transient volatile CounterCell[] counterCells;

还有一些常量也会用到

    
    private static final int MAXIMUM_CAPACITY = 1 << 30;

    
    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;

    
    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;

    
    private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

    
    static final int MOVED     = -1; // hash for forwarding nodes 节点迁移的标记
    static final int TREEBIN   = -2; // hash for roots of trees 
    static final int RESERVED  = -3; // hash for transient reservations
    static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

    
    static final int NCPU = Runtime.getRuntime().availableProcessors();

     兼容jdk1.8以前的分段锁
    private static final ObjectStreamField[] serialPersistentFields = {
        new ObjectStreamField("segments", Segment[].class),
        new ObjectStreamField("segmentMask", Integer.TYPE),
        new ObjectStreamField("segmentShift", Integer.TYPE)
    };

从下面的一段代码开始说起

 	ConcurrentHashMap concurrentHashMap = new ConcurrentHashMap<>(10);
    for (int i = 0; i < 12; i++) {
        concurrentHashMap.put(i, i);
    }

new ConcurrentHashMap<>(10)，通过源码看到只要指定了了初始容量，最总的初始容量一定会被变成大于它的最小2的幂次方，除非指定的容量大于了最大容量(MAXIMUM_CAPACITY)的一半。例如：指定了10，但是最终的初始化容量会是2⁴=16。
以下是源码(篇幅太长，我把注释删了):

    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(Map m) {
        this.sizeCtl = DEFAULT_CAPACITY;
        putAll(m);
    }

    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;
    }
  
	
    private static final int tableSizeFor(int c) {
        int n = c - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }

从以上代码中也能看到sizeCtl的大小被设置成了初始容量。

接下来进入第一次put：
我们先大概了解一次put的大体过程，再说细节

以下是源码：

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

    final V putVal(K key, V value, boolean onlyIfAbsent) {
    	// ConcurrentHashMap不能存储null的键值对
        if (key == null || value == null) throw new NullPointerException();
        // hash扰动函数，再算一次hash值，尽可能避免hash冲突
        int hash = spread(key.hashCode());
        //链表大小
        int binCount = 0;
        // 无线循环的处理放入逻辑，知道放入成功，每次循环可能只处理了一部分逻辑。
        for (Node[] tab = table;;) {
            Node f; int n, i, fh;
            // 如果没初始化，就初始化数组
            if (tab == null || (n = tab.length) == 0)
            	// 见下方initTable()的注释
                tab = initTable();
             // (n - 1)&hash计算放置的位置，如果i位置没数据，就CAS放进去
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node(hash, key, value, null)))
                    break;                   
            }
            // 如果当前节点的hash是-1(MOVED就是-1),就帮助迁移，针对的是多线程
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
                V oldVal = null;
                // 锁定当前节点，此时已经出现hash冲突。
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                    	// 如果是链表，就按照链表的形式增加节点
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node e = f;; ++binCount) {
                                K ek;
                                // 如果放的节点已经存在就跳过或是覆盖
                                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;
                            // 如果创建了一个新的节点就返回null，否则返回key对应的节点
                            if ((p = ((TreeBin)f).putTreeval(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                	// 如果链表大小达到了阈值就树化
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                     // 如果有旧值就直接返回，不用扩容
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        // 元素容量计数器增加
        addCount(1L, binCount);
        return null;
    }

	private final Node[] initTable() {
        Node[] tab; int sc;
        // 一个线程初始化就行了，其它线程自选等待
        while ((tab = table) == null || tab.length == 0) {
        	// 如果sc小于0证明有线程正在初始化，让出CPU
            if ((sc = sizeCtl) < 0)
                Thread.yield();
            // CAS锁定sizeCtl。
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
                try {
                	// 双重检查
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node[] nt = (Node[])new Node[n];
                        table = tab = nt;
                        // 计算下次扩容的阈值  n - (n >>> 2) <=> n-n*(1-0.75)
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

	private final void addCount(long x, int check) {
        CounterCell[] as; long b, s;
        // 这一部分应该是多线程在协助迁移，我也没看懂
        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[] tab, nt; int n, sc;
             //判断是否符合扩容条件
            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);
                }
                //如果锁定了SIZECTL表明此时有线程正在扩容
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))
                    // 扩容函数
                    transfer(tab, null);
                //计算所有线程的总元素数量？？
                s = sumCount();
            }
        }
    }
    
    private final void transfer(Node[] tab, Node[] nextTab) {
        int n = tab.length, stride;
        // 每个线程会从某个索引处迁移数据，这是要迁移的步长
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE;
        // 如果临时数组为空，创建
        if (nextTab == null) {           
            try {
                @SuppressWarnings("unchecked")
                //扩容两倍
                Node[] nt = (Node[])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;
        //目标转移节点，如果从这个节点进入，就会进入新的数组中
        ForwardingNode fwd = new ForwardingNode(nextTab);
        boolean advance = true;
        boolean finishing = false; 
        //遍历每一个数组元素，i起始索引，bound迁移元素的终点索引
        for (int i = 0, bound = 0;;) {
            Node f; int fh;
            //寻找到下一个要迁移的元素
            while (advance) {
                int nextIndex, nextBound;
                // 倒序迁移，
                if (--i >= bound || finishing)
                    advance = false;
                else if ((nextIndex = transferIndex) <= 0) {
                    i = -1;
                    advance = false;
                }
                //CAS确定下一个迁移元素
                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
                }
            }
            //在原数组的i位置放上转移节点，后续的操作就会在新数组上操作
            else if ((f = tabAt(tab, i)) == null)
                advance = casTabAt(tab, i, null, fwd);
            //fwd是转移节点，hash是-1放在原数组上，新来的线程如果发现操作的节点的hash是-1
            //就会帮忙迁移，这里是-1表示已经处理了
            else if ((fh = f.hash) == MOVED)
                advance = true; 
            else {
            	//准备迁移的节点
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        Node ln, hn;
                        //链表的迁移方式
                        if (fh >= 0) {
                            int runBit = fh & n;
                            Node lastRun = f;
                            for (Node p = f.next; p != null; p = p.next) {
                            	// 计算迁移数据的hash值高位是1还是0
                                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 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(ph, pk, pv, ln);
                                else
                                    hn = new Node(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 t = (TreeBin)f;
                            TreeNode lo = null, loTail = null;
                            TreeNode hi = null, hiTail = null;
                            int lc = 0, hc = 0;
                            for (Node e = t.first; e != null; e = e.next) {
                                int h = e.hash;
                                TreeNode p = new TreeNode
                                    (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(lo) : t;
                            hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                                (lc != 0) ? new TreeBin(hi) : t;
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                    }
                }
            }
        }
    }

也不知道写的好不好容易理解吗？有意见尽管提，谢谢！

1. HashMap的树化为什么是红黑树？

   红黑树是二叉平衡树，它保证二叉树的最大分支高度不超过最小分支高度的两倍（高度概念是二叉树中的一个概念，没获取一次子节点，高度+1），这样树的旋转频率不会特别高就会剩下来插入时间提高性能。或者说它适合HashMap这个场景。

2. ConcurrentHashMap什么key和value不能为null，而HashMap可以？

   二义性问题，假定都可以放null，那么get(key)得到的value会有以下两个问题：

   • value可以是被一开始这是进入的null
   • 可能map中从来就没有key-value这个映射关系，返回的也是null
     但是HashMap可以用contains(key)来避免二义性问题，而ConcurrentHashMap作为多线程访问容器，很有可能在A线程调用containsKey(key)方法之前B线程 可以中间再插入一步操作put或是remove，这样A线程用得到的结果去处理一个不再是预想中的map就会有问题了。

3. HashMap的容量为什么必须是2的幂次方

4. ConcurrentHashMap的链表数据迁移解释