ConcurrentHashMap怎么用

小編給大家分享一下ConcurrentHashMap怎么用，希望大家閱讀完這篇文章之后都有所收獲，下面讓我們一起去探討吧！

    首先看一下putVal方法，

if (tab == null || (n = tab.length) == 0)
    tab = initTable();

    如果還沒有table的話，就要先初始化table

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        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<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    // size 控制在  n的0.75
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

    這一段代碼相對(duì)簡單，這里的sizeCtl是整個(gè)過程中的一個(gè)非常重要的屬性，在擴(kuò)容，初始化等過程過程中會(huì)多次遇到。在這里也是充當(dāng)了一個(gè)排他鎖的作用，當(dāng)它為-1的時(shí)候，其它線程等待。

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
}

     如果要插入的槽是空的，那么直接插入就可以了。

else if ((fh = f.hash) == MOVED)
    tab = helpTransfer(tab, f);

    那么如果要插入的hash值為moved狀態(tài)即-1的時(shí)候，那么就要執(zhí)行helpTransfer方法了，對(duì)，就是先讓幫助擴(kuò)容。這里就要扯出來比較多的東西了，我們一點(diǎn)點(diǎn)來進(jìn)行分析。

    首先看看什么時(shí)候一個(gè)node的hash值變成了-1，一路看下去，只有

static final class ForwardingNode<K,V> extends Node<K,V>

  這個(gè)類使用到了，它里面有一個(gè)屬性

final Node<K,V>[] nextTable;

從這里也大概就能看出來，這個(gè)時(shí)候ConcurrentHashmap處于擴(kuò)容狀態(tài)，通過ForwardingNode就可以找到擴(kuò)容后的table。

 接著來看helpTransfer

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    // 這里還要再次檢查 當(dāng)前節(jié)點(diǎn)是不是 ForwardingNode 因?yàn)槿绻皇堑脑?，沒有辦法找到nextTable，也就沒有辦法幫助擴(kuò)容了
    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;
            // 進(jìn)來一個(gè)線程，則對(duì)sizeCtl+1,用以標(biāo)記參與擴(kuò)容的線程數(shù)
            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                // 進(jìn)行擴(kuò)容操作
                transfer(tab, nextTab);
                break;
            }
        }
        return nextTab;
    }
    return table;
}

    下來看看transfer擴(kuò)容操作是如何執(zhí)行的，這里感覺是ConcurrentHashmap的一個(gè)精華點(diǎn)，嘆為觀止。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    // 首先進(jìn)行分段，既每個(gè)線程每次處理的node數(shù)量，最小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];
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        transferIndex = n;
    }
    int nextn = nextTab.length;
    // 在這里創(chuàng)建了ForwardingNode
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    boolean advance = true;
    boolean finishing = false; // to ensure sweep before committing nextTab
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        // 判斷是否要繼續(xù)
        while (advance) {
            int nextIndex, nextBound;
            // 如果已經(jīng)結(jié)束了或者當(dāng)前已經(jīng)到了邊界
            if (--i >= bound || finishing)
                advance = false;
            // 擴(kuò)容時(shí)用的指針已經(jīng)小于0，則結(jié)束
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            // 擴(kuò)容的指針，從大向小移動(dòng)，從大向小移動(dòng)，每次減小stride
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        // i 小于 0 ，已經(jīng)結(jié)束了
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            // 如果已經(jīng)結(jié)束了，那么把table設(shè)置為nextTable
            if (finishing) {
                nextTable = null;
                table = nextTab;
                sizeCtl = (n << 1) - (n >>> 1);
                return;
            }
            // 說明當(dāng)前的線程已經(jīng)工作結(jié)束，sizeCtl - 1
            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
            }
        }
        //如果節(jié)點(diǎn)為空，設(shè)置該節(jié)點(diǎn)為fwd
        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) {
                        // 這里為什么是 fh & n 做 & 運(yùn)算 因?yàn)?nbsp;15 的二進(jìn)制是 1111  16是10000 31是 11111
                        // 所以，擴(kuò)容前和擴(kuò)容后只有第一位 & 運(yùn)算后會(huì)變，其它位都不變，所以與 table.length & 就可以了
                        int runBit = fh & n;
                        Node<K,V> lastRun = f;
                        // 先遍歷一遍，確定 ni -> n rehash相等的一段，這樣下一次重新分配槽的時(shí)候這一段就不再遍歷
                        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);
                        // 把已完成的節(jié)點(diǎn)標(biāo)記為fwd
                        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;
                    }
                }
            }
        }
    }
}

    再來看一下對(duì)于元素總數(shù)的統(tǒng)計(jì)實(shí)現(xiàn)。

private final void addCount(long x, int check)

    首先我們遇到了CounterCell這個(gè)類，結(jié)構(gòu)很簡單，只有一個(gè)long value，它是存儲(chǔ)數(shù)量的最小單元。

    先看第一次的判斷條件，如果conterCells已經(jīng)不為空，說明之前已經(jīng)出現(xiàn)了并發(fā)增加baseCount，否則counterCell不會(huì)被初始化。

if ((as = counterCells) != null ||
    !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x))

    或者在改變baseCount的時(shí)候出現(xiàn)了沖突，執(zhí)行下面代碼。

CounterCell a; long v; int m;
boolean uncontended = true;
// 如果 counterCell未初始化，或者長度為0  亦或者沒有這個(gè)對(duì)應(yīng)的槽   再或者更新對(duì)應(yīng)槽的時(shí)候出現(xiàn)沖突
// 這個(gè)時(shí)候說明要么 counterCell未初始化，要么說明又出現(xiàn)了對(duì)于同一個(gè)槽沖突，所以需要 fullAddCount來解決沖突
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();

    再來看看fullAddCount做了什么

private final void fullAddCount(long x, boolean wasUncontended) {
    int h;
    // 如果ThreadLocalRandom還沒有被初始化過，說明還沒有發(fā)生過碰撞
    if ((h = ThreadLocalRandom.getProbe()) == 0) {
        ThreadLocalRandom.localInit();      // force initialization
        h = ThreadLocalRandom.getProbe();
        wasUncontended = true;
    }
    boolean collide = false;                // True if last slot nonempty
    for (;;) {
        CounterCell[] as; CounterCell a; int n; long v;
        // 如果數(shù)組已經(jīng)被初始化
        if ((as = counterCells) != null && (n = as.length) > 0) {
            // 隨機(jī)選取的槽還未被初始化
            if ((a = as[(n - 1) & h]) == null) {
                // 獲取鎖
                if (cellsBusy == 0) {            // Try to attach new Cell
                    CounterCell r = new CounterCell(x); // Optimistic create
                    // U.compareAndSwapInt(this, CELLSBUSY, 0, 1)通過cas操作來獲取鎖
                    if (cellsBusy == 0 &&
                        U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                        boolean created = false;
                        try {               // Recheck under lock
                            CounterCell[] rs; int m, j;
                            if ((rs = counterCells) != null &&
                                (m = rs.length) > 0 &&
                                rs[j = (m - 1) & h] == null) {
                                rs[j] = r;
                                created = true;
                            }
                        } finally {
                            cellsBusy = 0;
                        }
                        if (created)
                            break;
                        continue;           // Slot is now non-empty
                    }
                }
                collide = false;
            }
            // 有競(jìng)爭(zhēng)的
            else if (!wasUncontended)       // CAS already known to fail
                wasUncontended = true;      // Continue after rehash
            else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
                break;
            // 如果槽的數(shù)量已經(jīng)超過了cpu個(gè)數(shù)，就不會(huì)碰撞了
            else if (counterCells != as || n >= NCPU)
                collide = false;            // At max size or stale
            else if (!collide)
                collide = true;
            // 獲取鎖，并對(duì)CounterCell進(jìn)行擴(kuò)容操作
            else if (cellsBusy == 0 &&
                     U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
                try {
                    if (counterCells == as) {// Expand table unless stale
                        CounterCell[] rs = new CounterCell[n << 1];
                        for (int i = 0; i < n; ++i)
                            rs[i] = as[i];
                        counterCells = rs;
                    }
                } finally {
                    cellsBusy = 0;
                }
                collide = false;
                continue;                   // Retry with expanded table
            }
            h = ThreadLocalRandom.advanceProbe(h);
        }
        // counter cell 沒有初始化的情況
        else if (cellsBusy == 0 && counterCells == as &&
                 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
            boolean init = false;
            try {                           // Initialize table
                if (counterCells == as) {
                    // 進(jìn)行初始化
                    CounterCell[] rs = new CounterCell[2];
                    rs[h & 1] = new CounterCell(x);
                    counterCells = rs;
                    init = true;
                }
            } finally {
                cellsBusy = 0;
            }
            if (init)
                break;
        }
        else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
            break;                          // Fall back on using base
    }
}

    我們前面講了擴(kuò)容的機(jī)制，那么擴(kuò)容的發(fā)起者肯定就是在addCount中了

while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
       (n = tab.length) < MAXIMUM_CAPACITY)

    這里有判斷  s 為 sumCount  即 baseCount加上各個(gè)節(jié)點(diǎn)的和為總數(shù)。如果s大于sizeCtl或者table不為空而且沒有到達(dá)最大值，則進(jìn)行擴(kuò)容操作。

看完了這篇文章，相信你對(duì)“ConcurrentHashMap怎么用”有了一定的了解，如果想了解更多相關(guān)知識(shí)，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！