解析ConcurrentHashMap: 預(yù)熱(內(nèi)部一些小方法分析)

更新時(shí)間：2021年06月10日 15:43:48 作者：興趣使然の草帽路飛

ConcurrentHashMap是由Segment數(shù)組結(jié)構(gòu)和HashEntry數(shù)組結(jié)構(gòu)組成。Segment的結(jié)構(gòu)和HashMap類似，是一種數(shù)組和鏈表結(jié)構(gòu)，今天給大家普及java面試常見(jiàn)問(wèn)題---ConcurrentHashMap知識(shí)，一起看看吧

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;// 該屬性是通過(guò)spread(int h)方法計(jì)算得到~
    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;
    }
    ...
}

1、spread(int h)方法

/**
 * 計(jì)算Node節(jié)點(diǎn)hash值的算法：參數(shù)h為hash值
 * eg: 
 * h二進(jìn)制為 --> 	 		 		 1100 0011 1010 0101 0001 1100 0001 1110
 * (h >>> 16) -->  					0000 0000 0000 0000 1100 0011 1010 0101 
 * (h ^ (h >>> 16)) -->				1100 0011 1010 0101 1101 1111 1011 1011
 * 注：(h ^ (h >>> 16)) 目的是讓h的高16位也參與尋址計(jì)算，使得到的hash值更分散，減少hash沖突產(chǎn)生~
 * ------------------------------------------------------------------------------
 * HASH_BITS -->					0111 1111 1111 1111 1111 1111 1111 1111
 * (h ^ (h >>> 16)) -->				1100 0011 1010 0101 1101 1111 1011 1011
 * (h ^ (h >>> 16)) & HASH_BITS --> 0100 0011 1010 0101 1101 1111 1011 1011
 * 注： (h ^ (h >>> 16))得到的結(jié)果再& HASH_BITS，目的是為了讓得到的hash值結(jié)果始終是一個(gè)正數(shù)
 */
static final int spread(int h) {
    // 讓原來(lái)的hash值異或^原來(lái)hash值的右移16位，再與&上HASH_BITS(0x7fffffff: 二進(jìn)制為31個(gè)1)
    return (h ^ (h >>> 16)) & HASH_BITS;
}

下面介紹tabAt(Node<K,V>[] tab, int i)方法：獲取 tab(Node[]) 數(shù)組指定下標(biāo) i 的Node節(jié)點(diǎn)。

2、tabAt方法

/**
 * 獲取 tab(Node[]) 數(shù)組指定下標(biāo) i 的Node節(jié)點(diǎn)
 * Node<K,V>[] tab：表示Node[]數(shù)組
 * int i：表示數(shù)組下標(biāo)
 */
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
    // ((long)i << ASHIFT) + ABASE 的作用：請(qǐng)看下面的分析描述~
    return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}

在分析((long)i << ASHIFT) + ABASE時(shí)，先復(fù)習(xí)一下上一篇文章：ConcurrentHashMap源碼解析_01 成員屬性、內(nèi)部類、構(gòu)造方法分析中介紹到的一些屬性字段的含義：

// Node數(shù)組的class對(duì)象
Class<?> ak = Node[].class;
// U.arrayBaseOffset(ak)：根據(jù)as獲取Node[]數(shù)組第一個(gè)元素的偏移地址ABASE
ABASE = U.arrayBaseOffset(ak);
// scale：表示數(shù)組中每一個(gè)單元所占用的空間大小，即scale表示Node[]數(shù)組中每一個(gè)單元所占用的空間
int scale = U.arrayIndexScale(ak);// scale必須為2的次冪數(shù)
// numberOfLeadingZeros(scale) 根據(jù)scale，返回當(dāng)前數(shù)值轉(zhuǎn)換為二進(jìn)制后，從高位到地位開(kāi)始統(tǒng)計(jì)，統(tǒng)計(jì)有多少個(gè)0連續(xù)在一塊：eg, 8轉(zhuǎn)換二進(jìn)制=>1000 則 numberOfLeadingZeros(8)的結(jié)果就是28，為什么呢？因?yàn)镮nteger是32位，1000占4位，那么前面就有32-4個(gè)0，即連續(xù)最長(zhǎng)的0的個(gè)數(shù)為28個(gè)
// 4轉(zhuǎn)換二進(jìn)制=>100 則 numberOfLeadingZeros(8)的結(jié)果就是29
// ASHIFT = 31 - Integer.numberOfLeadingZeros(4) = 2 那么ASHIFT的作用是什么呢？其實(shí)它有數(shù)組尋址的一個(gè)作用：
// 拿到下標(biāo)為5的Node[]數(shù)組元素的偏移地址(存儲(chǔ)地址)：假設(shè)此時(shí) 根據(jù)scale計(jì)算得到的ASHIFT = 2
// ABASE + (5 << ASHIFT) == ABASE + (5 << 2) == ABASE + 5 * scale(乘法運(yùn)算效率低)，就得到了下標(biāo)為5的數(shù)組元素的偏移地址
ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);

由上面的幾個(gè)屬性字段的復(fù)習(xí)介紹，不難得出：

((long)i << ASHIFT) + ABASE 就是得到當(dāng)前Node[] 數(shù)組下標(biāo)為i的節(jié)點(diǎn)對(duì)象的偏移地址。

然后再通過(guò)(Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE) 方法，根據(jù)Node[]和目標(biāo)節(jié)點(diǎn)Node的偏移地址兩個(gè)參數(shù)，得到下標(biāo)為i的Node節(jié)點(diǎn)對(duì)象。

雖然這樣很繞，不如，但是直接根據(jù)偏移地址去尋找數(shù)組元素效率較高~

3、casTabAt方法

/**
 * 通過(guò)CAS的方式去向Node數(shù)組指定位置i設(shè)置節(jié)點(diǎn)值，設(shè)置成功返回true，否則返回false
 * Node<K,V>[] tab：表示Node[]數(shù)組
 * int i：表示數(shù)組下標(biāo)
 * Node<K,V> c：期望節(jié)點(diǎn)值
 * Node<K,V> v：要設(shè)置的節(jié)點(diǎn)值
 */
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                    Node<K,V> c, Node<K,V> v) {
    // 調(diào)用Unsafe的比較并交換去設(shè)置Node[]數(shù)組指定位置的節(jié)點(diǎn)值，參數(shù)如下：
    // tab：Node[]數(shù)組
    // ((long)i << ASHIFT) + ABASE：下標(biāo)為i數(shù)組桶的偏移地址
    // c：期望節(jié)點(diǎn)值
    // v：要設(shè)置的節(jié)點(diǎn)值
    return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}

4、setTabAt方法

/**
 * 根據(jù)數(shù)組下標(biāo)i，設(shè)置Node[]數(shù)組指定下標(biāo)位置的節(jié)點(diǎn)值：
 * Node<K,V>[] tab：表示Node[]數(shù)組
 * int i：表示數(shù)組下標(biāo)
 * Node<K,V> v：要設(shè)置的節(jié)點(diǎn)值
 */
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    // ((long)i << ASHIFT) + ABASE：下標(biāo)為i數(shù)組桶的偏移地址
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}

5、resizeStamp方法

/**
 * table數(shù)組擴(kuò)容時(shí)，計(jì)算出一個(gè)擴(kuò)容標(biāo)識(shí)戳，當(dāng)需要并發(fā)擴(kuò)容時(shí)，當(dāng)前線程必須拿到擴(kuò)容標(biāo)識(shí)戳才能參與擴(kuò)容：
 */
static final int resizeStamp(int n) {
    // RESIZE_STAMP_BITS：固定值16，與擴(kuò)容相關(guān)，計(jì)算擴(kuò)容時(shí)會(huì)根據(jù)該屬性值生成一個(gè)擴(kuò)容標(biāo)識(shí)戳
    return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}
=

舉例子分析一下：

當(dāng)我們需要table容量從16 庫(kù)容到32時(shí)：Integer.numberOfLeadingZeros(16) 會(huì)得到，怎么得來(lái)的呢？

numberOfLeadingZeros(n) 根據(jù)傳入的n，返回當(dāng)前數(shù)值轉(zhuǎn)換為二進(jìn)制后，從高位到地位開(kāi)始統(tǒng)計(jì)，統(tǒng)計(jì)有多少個(gè)0連續(xù)在一塊：
eg：16 轉(zhuǎn)換二進(jìn)制 => 1 0000 則 numberOfLeadingZeros(16)的結(jié)果就是 27，因?yàn)?code>Integer是32位，1 0000占5位，那么前面就有(32 - 5)個(gè)0，即連續(xù)最長(zhǎng)的0的個(gè)數(shù)為27個(gè)。
(1 << (RESIZE_STAMP_BITS - 1))：其中RESIZE_STAMP_BITS是一個(gè)固定值16，與擴(kuò)容相關(guān)，計(jì)算擴(kuò)容時(shí)會(huì)根據(jù)該屬性值生成一個(gè)擴(kuò)容標(biāo)識(shí)戳。

下面就來(lái)計(jì)算一下：

// 從16擴(kuò)容到32
16 -> 32 
// 用A表示：
numberOfLeadingZeros(16) => 1 0000 => 27 => 0000 0000 0001 1011
// 用B表示：
(1 << (RESIZE_STAMP_BITS - 1)) => (1 << (16 - 1) => 1000 0000 0000 0000 => 32768
// A | B 
Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1)) 
-----------------------------------------------------------------
0000 0000 0001 1011  ---> A
1000 0000 0000 0000  ---> B
-------------------  ---> | 按位或                             
1000 0000 0001 1011  ---> 計(jì)算得到擴(kuò)容標(biāo)識(shí)戳

6、tableSizeFor方法

/**
 * 根據(jù)c,計(jì)算得到大于等于c的，最小2的次冪數(shù)，該方法在HashMap源碼中分析過(guò)~
 * eg：c = 28 ，則計(jì)算得到的返回結(jié)果為 32
 * c = 28 ==> n=27 => 0b 11011
 *
 * 11011 | 01101 => 11111 ---- n |= n >>> 1
 * 11111 | 00111 => 11111 ---- n |= n >>> 2
 * ....
 * => 11111 + 1 = 100000 = 32
 */
private static final int tableSizeFor(int c) {
    int n = c - 1;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}

7、構(gòu)造方法(復(fù)習(xí))

復(fù)習(xí)一下上一篇文章中的并發(fā)HashMap的構(gòu)造方法：

// 無(wú)慘構(gòu)造
public ConcurrentHashMap() {
}
// 指定初始化容量
public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)
        throw new IllegalArgumentException();
    int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
            MAXIMUM_CAPACITY :
            tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
    /**
     * sizeCtl > 0
     * 當(dāng)目前table未初始化時(shí)，sizeCtl表示初始化容量
     */
    this.sizeCtl = cap;
}
// 根據(jù)一個(gè)Map集合來(lái)初始化
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
    // sizeCtl設(shè)置為默認(rèn)容量值
    this.sizeCtl = DEFAULT_CAPACITY;
    putAll(m);
}
// 指定初始化容量，和負(fù)載因子
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}
// 指定初始化容量，和負(fù)載因子，并發(fā)級(jí)別
public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
        throw new IllegalArgumentException();
    // 當(dāng)指定的初始化容量initialCapacity小于并發(fā)級(jí)別concurrencyLevel時(shí)
    if (initialCapacity < concurrencyLevel) 
        // 初始化容量值設(shè)置為并發(fā)級(jí)別的值。
        // 即，JDK1.8以后并發(fā)級(jí)別由散列表長(zhǎng)度決定
        initialCapacity = concurrencyLevel; 
    // 根據(jù)初始化容量和負(fù)載因子，去計(jì)算size
    long size = (long)(1.0 + (long)initialCapacity / loadFactor);
    // 根據(jù)size重新計(jì)算數(shù)組初始化容量
    int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
    /**
     * sizeCtl > 0
     * 當(dāng)目前table未初始化時(shí)，sizeCtl表示初始化容量
     */
    this.sizeCtl = cap;
}