java併發編程學習之ConcurrentHashMap(JDK1.7)

以前HashMap中提過，併發的時候，可能形成死循環，且線程不安全，那麼在多線程中能夠用ConcurrentHashMap來避免這一狀況。

Segment

ConcurrentHashMap是由多個Segment組成的，Segment繼承了ReentrantLock，每次加鎖都是對某個Segment，不會影響其餘Segment，達到了鎖分離（也叫分段鎖）的做用。
每一個Segment又包含了HashEntry數組，HashEntry是一個鏈表。以下圖所示：

初始化

initialCapacity：初始容量大小，默認16。
loadFactor：擴容因子，table擴容使用，Segments不擴容。默認0.75，當Segment容量大於initialCapacity*loadFactor時，開始擴容
concurrencyLevel：併發數，默認16，直接影響segmentShift和segmentMask的值，以及Segment的初始化數量。Segment初始化的數量，爲最接近且大於的辦等於2的N次方的值，好比concurrencyLevel=16，Segment數量爲16，concurrencyLevel=17，Segment數量爲32。segmentShift的值是這樣的，好比Segment是32，相對於2的5次方，那麼他的值就是32-5，爲27，後面無符號右移27位，也就是取高5位的時候，就是0到31的值，此時Segment的下標也是0到31，取模後對應着每一個Segment。segmentMask就是2的n次方-1，這邊n是5，用於取模。以前在hashmap的indexFor方法有提過。
初始化的時候，還要初始化第一個Segment，以及Segment中table數組的大小，這邊大小是大於等於initialCapacity除以Segment數組的個數，平均分配，最小是2，且是2的N次方。好比initialCapacity是32，concurrencyLevel是16的時候，那麼Segment的個數也是16，32除以16，等於2，若是initialCapacity是33，Segment是16，33除以16，取4。

public ConcurrentHashMap() {
    this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    if (concurrencyLevel > MAX_SEGMENTS)
        concurrencyLevel = MAX_SEGMENTS;
    // Find power-of-two sizes best matching arguments
    int sshift = 0;
    int ssize = 1;
    while (ssize < concurrencyLevel) {
        ++sshift;
        ssize <<= 1;
    }
    this.segmentShift = 32 - sshift;//用於高位，判斷落在哪一個Segment
    this.segmentMask = ssize - 1;//用於取模。以前在hashmap的indexFor方法有提過。2的n次方-1
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c)
        cap <<= 1;
    // create segments and segments[0]
    Segment<K,V> s0 =
        new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
                         (HashEntry<K,V>[])new HashEntry[cap]);//初始化第一個位置的Segment
    Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];//初始化Segments
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

詳細流程

put方法

public V put(K key, V value) {
    Segment<K,V> s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    //無符號右移後取模，落在哪一個Segment上面
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        s = ensureSegment(j);
    return s.put(key, hash, value, false);
}

ensureSegment方法
肯定落在哪一個Segment上，若是爲空，就初始化，由於以前就初始化第一個Segment

private Segment<K,V> ensureSegment(int k) {
    final Segment<K,V>[] ss = this.segments;
    long u = (k << SSHIFT) + SBASE; // raw offset
    Segment<K,V> seg;
    if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
        //使用segment[0]的table長度和loadFactor來初始化
        Segment<K,V> proto = ss[0]; // use segment 0 as prototype
        int cap = proto.table.length;
        float lf = proto.loadFactor;
        int threshold = (int)(cap * lf);
        HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
            == null) { // recheck
            Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
            while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))//cas操做，只能一個設值成功，若是其餘成功了，就賦值，並返回
                   == null) {
                if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                    break;
            }
        }
    }
    return seg;
}

put方法

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);//獲取Segment的鎖
    V oldValue;
    try {
        HashEntry<K,V>[] tab = table;
        int index = (tab.length - 1) & hash;//上面是獲取Segment取高位的hash，這邊是tabel的hash，
        HashEntry<K,V> first = entryAt(tab, index);//取到hash位置的數組的表頭
        for (HashEntry<K,V> e = first;;) {//從頭結點遍歷
            if (e != null) {
                K k;
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) {//key相同，或者hash值同樣
                    oldValue = e.value;
                    if (!onlyIfAbsent) {//是否替換
                        e.value = value;
                        ++modCount;
                    }
                    break;
                }
                e = e.next;
            }
            else {
                if (node != null)//不爲空，設置爲表頭
                    node.setNext(first);
                else
                    node = new HashEntry<K,V>(hash, key, value, first);/初始化後放表頭
                int c = count + 1;
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    rehash(node);// 擴容
                else
                    setEntryAt(tab, index, node);//把新的節點放在tab的index上面
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();//釋放鎖
    }
    return oldValue;
}

scanAndLockForPut方法
嘗試獲取鎖，沒獲取到先初始化node

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
    HashEntry<K,V> first = entryForHash(this, hash);//獲取hash後的頭結點，有存在null的狀況
    HashEntry<K,V> e = first;
    HashEntry<K,V> node = null;
    int retries = -1; // negative while locating node
    while (!tryLock()) {//這個put方法先嚐試獲取，獲取不到，這邊while循環嘗試獲取
        HashEntry<K,V> f; // to recheck first below
        if (retries < 0) {
            if (e == null) {//結點爲空的時候
                if (node == null) // speculatively create node
                    node = new HashEntry<K,V>(hash, key, value, null);//初始化node
                retries = 0;
            }
            else if (key.equals(e.key))//頭結點不爲空的時候
                retries = 0;
            else
                e = e.next;
        }
        else if (++retries > MAX_SCAN_RETRIES) {//超太重試次數，直接進入阻塞隊列等待鎖
            lock();
            break;
        }
        else if ((retries & 1) == 0 &&
                 (f = entryForHash(this, hash)) != first) {//不等於first，就是已經有其餘節點進入
            e = first = f; // re-traverse if entry changed
            retries = -1;
        }
    }
    return node;
}

rehash方法，擴容，對table擴容

private void rehash(HashEntry<K,V> node) {
    HashEntry<K,V>[] oldTable = table;
    int oldCapacity = oldTable.length;
    int newCapacity = oldCapacity << 1;//左移，以前的2倍
    threshold = (int)(newCapacity * loadFactor);
    HashEntry<K,V>[] newTable =
        (HashEntry<K,V>[]) new HashEntry[newCapacity];
    int sizeMask = newCapacity - 1;
    for (int i = 0; i < oldCapacity ; i++) {
        HashEntry<K,V> e = oldTable[i];
        if (e != null) {
            HashEntry<K,V> next = e.next;
            int idx = e.hash & sizeMask;
            if (next == null)   //  爲空，沒有後面的節點，直接給新數組
                newTable[idx] = e;
            else { // Reuse consecutive sequence at same slot
                HashEntry<K,V> lastRun = e;
                int lastIdx = idx;
                //由於數組是2倍的擴容，因此從新hash後，要麼落在跟以前索引同樣的位置，要麼就是加上oldCapacity 的值，
                //好比容量是2，擴容4，如今hash是2，4，6，10，14那麼後面3個都是除4餘2，能夠直接複製
                for (HashEntry<K,V> last = next;
                     last != null;
                     last = last.next) {
                    int k = last.hash & sizeMask;
                    if (k != lastIdx) {//hash不同，從新
                        lastIdx = k;
                        lastRun = last;
                    }
                }
                //執行上面，就是lastRun是6，10，14
                newTable[lastIdx] = lastRun;//上面
                // Clone remaining nodes克隆的時候，碰到lastrun，直接根據因此給值，可是前面有可能的索引跟lastrun同樣，好比2
                for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
                    V v = p.value;
                    int h = p.hash;
                    int k = h & sizeMask;
                    HashEntry<K,V> n = newTable[k];
                    newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
                }
            }
        }
    }
    int nodeIndex = node.hash & sizeMask; // add the new node
    node.setNext(newTable[nodeIndex]);//加入到頭結點
    newTable[nodeIndex] = node;
    table = newTable;
}

get方法

public V get(Object key) {
    Segment<K,V> s; // manually integrate access methods to reduce overhead
    HashEntry<K,V>[] tab;
    int h = hash(key);
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&//找到Segment，邏輯同put
        (tab = s.table) != null) {
        for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);//找到table，邏輯同put
             e != null; e = e.next) {//遍歷table
            K k;
            if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                return e.value;
        }
    }
    return null;
}