JDK之ConcurrentHashMap

    注：分析JDK8的ConcurrentHashMap，JDK6/7上的實現和JDK8上的不同。

1.構造方法中的參數含義

    構造方法中，有三個參數，以下，第三個參數纔是一位數組的長度，第一個參數和第二個參數與Map的擴容有關

    List-1

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)

    構造方法以下List-2

    List-2 initialCapacity的值不能小於concurrencyLevel，經過initialCapacity和loadFactor計算出值並賦值給sizeCtl

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;
    }

    sizeCtl在哪用到了，以下圖1所示，若是sizeCtl的值大於0，說明咱們指定了capacity，因此不會使用默認的DEFAULT_CAPACITY。注意sizeCtl的值是容量，不是當前Map中元素的個數。

                                                              圖1 初始化表

2.併發度，一維數組

    一維數組存儲在這個中，長度默認是16，即默認的ConcurrenyLevel。

    List-2

transient volatile Node<K,V>[] table=(Node<K,V>[])new Node<?,?>[n];

    Node的類結構以下所示：

    List-3

static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        volatile V val;
        volatile Node<K,V> next;

        Node(int hash, K key, V val, Node<K,V> next) {
            this.hash = hash;
            this.key = key;
            this.val = val;
            this.next = next;
        }
...
}

    來看張ConcurrentHashMap的數據結構圖，以下圖1所示：

                                                      圖1 ConcurrentHashMap的數據結構

    如圖1所示，當A對應的鏈表長度達到8後，就會轉換爲紅黑樹。爲何是8：屬性TREEIFY_THRESHOLD上有註釋，不過沒怎麼看懂。我我的以爲，若是鏈表長度小於8就轉換爲紅黑樹，效率上可能不如鏈表，畢竟紅黑樹的調整是比鏈表複雜的，這時不如直接使用鏈表；若是鏈表長度超過8，那麼使用鏈表，查詢效率會變低(最壞狀況下遍歷整條鏈表)，此時用紅黑樹那麼就能將查詢的複雜度由O(n)下降到LogN。

    紅黑樹中的節點TreeNode：

    List-4

static final class TreeNode<K,V> extends Node<K,V> {
        TreeNode<K,V> parent;  // red-black tree links
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;

        TreeNode(int hash, K key, V val, Node<K,V> next,
                 TreeNode<K,V> parent) {
            super(hash, key, val, next);
            this.parent = parent;
        }
...
}

3.多線程進行put，如何保證線程安全

    來看一張圖，以下圖2：

                                                             圖2 putVal方法中

    假設put(x)時，插入的bucket位置是圖1中A對應的鏈表，那麼會對A進行synchronized，這樣就保證了同一時刻只有線程能訪問synchronized的代碼塊，這個代碼塊中，有插入Node的操做，這樣就保證了線程安全。

    ConcurrentHashMap的get()方法是不須要加上synchronize的，由於不少關鍵的屬性都有volatile修飾保證了可見性，因此get()方法上不加synchronize的狀況下，也能夠獲得結果。

4.如何擴容的，何時擴容

    問題1：每次進行put後，都會調用addCount方法，對Map中當前元素數量加1。以後會調用sumCount()方法獲取Map中當前有多少個元素，若是等於或者大於sizeCtl的值，就會觸發擴容。

    以下這個方法中就是觸發擴容的入口。

addCount(long x, int check)

    以下這個屬性與擴容有關。

private transient volatile Node<K,V>[] nextTable;

    實現擴容的代碼在以下方法中，實現起來很複雜，能夠參考這篇文章。

transfer(Node<K,V>[] tab, Node<K,V>[] nextTab)

    

5.怎麼實現統計Map中元素個數的功能

    與倆個屬性有關，以下:

/**
     * Base counter value, used mainly when there is no contention,
     * but also as a fallback during table initialization
     * races. Updated via CAS.
     */
    private transient volatile long baseCount;
    /**
     * Table of counter cells. When non-null, size is a power of 2.
     */
    private transient volatile CounterCell[] counterCells;

    注：counterCells的長度與CPU個數有關。

    CounterCell類以下:

@sun.misc.Contended static final class CounterCell {
        volatile long value;
        CounterCell(long x) { value = x; }
    }

     解說下這個baseCount和counterCells是如何工做的：

• 狀況1： 假設只有一個線程進行put，Map中元素個數加1，會對baseCount進行CAS進行值加1，因爲此時只有一個線程，因此CAS會成功，注意這裏的是Unsafe的CAS，而不是AtomicInteger之類的CAS，因此若是失敗會立刻返回false。
• 狀況2：假設有多個線程進行put，此時多個線程對baseCount進行CAS，此時如有些線程發生CAS失敗，說明對baseCount競爭失敗，注意這裏使用的是Unsafe的CAS，而不是AtomicInteger之類的CAS，因此若是失敗會立刻返回false。此時從counterCells中隨機選取一個出來，對CounterCell的value作CAS進行加1，若是此時發生CAS失敗，說明碰到了競爭，怎麼處理呢，看下圖，將counterCells的數組長度翻倍，以後拷貝元素組的值到新數組中對應的位置。

    調用size()時，會調用sumCount()，sumCount()方法實現以下：

final long sumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }

    首先得到baseCount的值，以後遍歷每一個counterCells中的value，就是Map中的元素的值，固然，獲得的這個值不必定準確、不是實時的。

    JDK8的ConcurrentHashMap中計算Map中元素個數的方法與LongAddr、DoubleAdder很相似。

    JDK8中獲取size的實現，比JDK6/7中的要好不少了，若是你看過JDK6/7中ConcurrentHashMap的實現，應該會有所感覺的