ConcurrentHashMap 详解二-白红宇

上一 part 说了如何扩容, 接下来讨论不同情况的插入

插入元素到索引位置

if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {    if (casTabAt(tab, i, null,                 new Node
    
     (hash, key, value, null)))        break;                   // no lock when adding to empty bin}/** * 读取 tab 对应位置 i 的值 * 这里使用原子读方式, 为什么要这样做暂时没搞明白, 这里给出自己一个假设 * tab 的元素不是 volatile, 虽然节点把 val 和 next 设置 volatile * 但 tab 索引的第一个元素不算 volatile, 如果 tab[i] 方式读取可能是读取工作内存的值 * 所以用原子读方式读取主内存的值才行 */static final 
     
       Node
      
        tabAt(Node
       
        [] tab, int i) {    return (Node
        
         )U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);}/** * CAS 算法, 如果 tab 的位置 i 的值与 c 相同, 就用 v 更新它 */static final 
         
           boolean casTabAt(Node
          
           [] tab, int i, Node
           
             c, Node
            
              v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);}

插入元素时正在扩容

有可能线程 A 进行插入时, 线程 B 正在对数组扩容, 这时候我们需要协助扩容

// 当哈希值是 MOVED, 表示其他线程在扩容, 参考上一 part 扩容部分if ((fh = f.hash) == MOVED)    // 这个方法放到下面 addCount 方法后面讲    tab = helpTransfer(tab, f);/** * ForwardingNode 类 */static final class ForwardingNode
    
      extends Node
     
       {    // 表示扩容后的新数组    final Node
      
       [] nextTable;    ForwardingNode(Node
       
        [] tab) {        super(MOVED, null, null, null);        this.nextTable = tab;    }}

插入元素到链表或者树

V oldVal = null;// f 是头结点, 先把它锁上, 防止其他线程同时操作这个链表或者树synchronized (f) {    // 再次判断这个头结点是不是原来的头结点    // 这是为了防止在锁上前一刻被其他线程修改了    if (tabAt(tab, i) == f) {        // hash 值大于等于 0, 说明这是链表, 参考上一 part        // 然后就是链表节点的插入了, 跟 HashMap 差不多, 可以参考 HashMap 详解        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;                }            }        }        // 注意对于树结构, ConcurrentHashMap 是用 TreeBin 进行封装        // 插入树节点也可以参考 HashMap, 差不多的过程        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) {    // 链表转树结构, 同样可以参考 HashMap 详解    if (binCount >= TREEIFY_THRESHOLD)        treeifyBin(tab, i);    if (oldVal != null)        return oldVal;    break;}

是否扩容

节点插入后会调用 addCount 方法来判断是否需要扩容

private final void addCount(long x, int check) {    CounterCell[] as; long b, s;    // 这里是统计容量 size, 执行加一操作, 不仔细讨论了    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;                // CAS 把 sizeCtl 成功加一, 本线程开始协助扩容                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))                    transfer(tab, nt);            }            // 没有其他线程扩容, 说明本线程是第一个扩容的            // 此时就把 sizeCtl 设置成一个非常大的负数            // 因为是第一个扩容, 所以新数组是 null            else if (U.compareAndSwapInt(this, SIZECTL, sc,                                         (rs << RESIZE_STAMP_SHIFT) + 2))                transfer(tab, null);            s = sumCount();        }    }}/** * 返回一个标识, 这个标识经过 RESIZE_STAMP_SHIFT 左移必定为负数 * Integer.numberOfLeadingZeros 返回 n 对应 32 位二进制数左侧 0 的个数 * 如 9（0000 0000 0000 0000 0000 0000 0000 1001）返回 28 * 1 << (RESIZE_STAMP_BITS - 1) = 2^15，其中 RESIZE_STAMP_BITS = 16 * RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS = 16 */static final int resizeStamp(int n) {    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));}/** * 协助扩容方法 * 经过 addCount 方法, 这个协助扩容就更容易理解了 */final Node
     
      [] helpTransfer(Node
      
       [] tab, Node
       
         f) {    Node
        
         [] nextTab; int sc;    if (tab != null && (f instanceof ForwardingNode) &&        (nextTab = ((ForwardingNode
         
          )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; // 同样是把 sizeCtl 加一, 然后协助扩容 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { transfer(tab, nextTab); break; } } return nextTab; } return table;}