ConcurrentHashMap 詳解二

上一 part 說了如何擴容, 接下來讨論不同情況的插入

插入元素到索引位置

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
}

/**
 * 讀取 tab 對應位置 i 的值
 * 這裡使用原子讀方式, 為什麼要這樣做暫時沒搞明白, 這裡給出自己一個假設
 * tab 的元素不是 volatile, 雖然節點把 val 和 next 設定 volatile
 * 但 tab 索引的第一個元素不算 volatile, 如果 tab[i] 方式讀取可能是讀取工作記憶體的值
 * 是以用原子讀方式讀取主記憶體的值才行
 */
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 算法, 如果 tab 的位置 i 的值與 c 相同, 就用 v 更新它
 */
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);
}           

插入元素時正在擴容

有可能線程 A 進行插入時, 線程 B 正在對數組擴容, 這時候我們需要協助擴容

// 當哈希值是 MOVED, 表示其他線程在擴容, 參考上一 part 擴容部分
if ((fh = f.hash) == MOVED)
    // 這個方法放到下面 addCount 方法後面講
    tab = helpTransfer(tab, f);

/**
 * ForwardingNode 類
 */
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);
        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<K,V> 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<K,V> pred = e;
                if ((e = e.next) == null) {
                    pred.next = new Node<K,V>(hash, key,
                                              value, null);
                    break;
                }
            }
        }

        // 注意對于樹結構, ConcurrentHashMap 是用 TreeBin 進行封裝
        // 插入樹節點也可以參考 HashMap, 差不多的過程
        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;
            }
        }
    }
}
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<K,V>[] 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<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;
            // 同樣是把 sizeCtl 加一, 然後協助擴容
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}