春節(jié)馬上就要到了，估計(jì)有不少的人已經(jīng)做好了假期的游玩準(zhǔn)備，有些已經(jīng)準(zhǔn)備好回老家過年了吧，肯定也有很多人早早的最好了假期之后的求職準(zhǔn)備，畢竟年后都是各個(gè)公司的“離職潮”有離職，肯定也有求職的，為此，小編已經(jīng)準(zhǔn)備好了一些多線程面試題及答案，祝大家在新年一年，求職順利。

ConcurrentHashMap非阻塞Hash集合?

ConcurrentHashMap是Java并發(fā)包中提供的一個(gè)線程安全且高效的HashMap實(shí)現(xiàn)，ConcurrentHashMap在并發(fā)編程的場景中使用頻率非常之高，本文就來分析下ConcurrentHashMap的實(shí)現(xiàn)原理，并對其實(shí)現(xiàn)原理進(jìn)行分析。

● ConcurrentLinkedQuere 類圖

ConcurrentHashMap是由Segment數(shù)組結(jié)構(gòu)和HashEntry數(shù)組結(jié)構(gòu)組成。Segment是一種可重入鎖。

ReentrantLock在ConcurrentHashMap里扮演鎖的角色，HashEntry則用于存儲(chǔ)鍵值對數(shù)據(jù)。一個(gè)ConcurrentHashMap里包含一個(gè)Segment數(shù)組，Segment的結(jié)構(gòu)和HashMap類似，是一種數(shù)組和鏈表結(jié)構(gòu)，一個(gè)Segment里包含一個(gè)HashEntry數(shù)組，每個(gè)HashEntry是一個(gè)鏈表結(jié)構(gòu)的元素，每個(gè)Segment守護(hù)者一個(gè)HashEntry數(shù)組里的元素，當(dāng)對HashEntry數(shù)組的數(shù)據(jù)進(jìn)行修改時(shí)，必須首先獲得它對應(yīng)的Segment鎖。

● ConcurrentLinkedQuere實(shí)現(xiàn)原理

眾所周知，哈希表是非常高效的，復(fù)雜度為O(1)的數(shù)據(jù)結(jié)構(gòu)，在Java開發(fā)中，我們最常見到最頻繁使用的就是HashMap和HashTable，但是在線程競爭激烈的并發(fā)場景中使用都不夠合理。

● HashMap：先說HashMap，HashMap是線程不安全的，在并發(fā)環(huán)境下，可能會(huì)形成環(huán)狀鏈表(擴(kuò)容時(shí)可能造成，具體原因自行百度google或查看源碼分析)，導(dǎo)致get操作時(shí)，cpu 空轉(zhuǎn)，所以，在并發(fā)環(huán)境中使用HashMap是非常危險(xiǎn)的。

● HashTable：HashTable和HashMap的實(shí)現(xiàn)原理幾乎一樣，差別無非是：

1、HashTable不允許key和value為null;

2、HashTable是線程安全的。但是HashTable線程安全的策略實(shí)現(xiàn)代價(jià)卻太大了，簡單粗暴，get/put所有相關(guān)操作都是synchronized的，這相當(dāng)于給整個(gè)哈希表加了一把大鎖，多線程訪問時(shí)候，只要有一個(gè)線程訪問或操作該對象，那其他線程只能阻塞，相當(dāng)于將所有的操作串行化，在競爭激烈的并發(fā)場景中性能就會(huì)非常差。如下圖：

HashTable性能差主要是由于所有操作需要競爭同一把鎖，而如果容器中有多把鎖，每一把鎖鎖一段數(shù)據(jù)，這樣在多線程訪問時(shí)不同段的數(shù)據(jù)時(shí)，就不會(huì)存在鎖競爭了，這樣便可以有效地提高并發(fā)效率。這就是ConcurrentHashMap所采用的“分段鎖”思想。

● ConcurrentLinkedQuere源碼解析

ConcurrentHashMap采用了非常精妙的"分段鎖"策略，ConcurrentHashMap的主干是個(gè)Segment數(shù)組。

final Segment<K,V>[] segments;

Segment繼承了ReentrantLock，所以它就是一種可重入鎖(ReentrantLock)。在ConcurrentHashMap中，一個(gè)Segment就是一個(gè)子哈希表，Segment里維護(hù)了一個(gè)HashEntry數(shù)組，并發(fā)環(huán)境下，對于不同Segment的數(shù)據(jù)進(jìn)行的操作是不用考慮鎖競爭的。(就按默認(rèn)的ConcurrentLeve為16來講，理論上就允許16個(gè)線程并發(fā)執(zhí)行)所以，對于同一個(gè)Segment的操作才需考慮線程同步，不同的Segment則無需考慮。Segment類似于HashMap，一個(gè)Segment維護(hù)著一個(gè)HashEntry數(shù)組。

transient volatile HashEntry<K，V>[] table;  

HashEntry是目前我們提到的最小的邏輯處理單元了。一個(gè)ConcurrentHashMap維護(hù)一個(gè)Segment數(shù)組，一個(gè)Segment維護(hù)一個(gè)HashEntry數(shù)組。

static final class HashEntry<K, V> {
    final int hash;
    final K key;
    volatile V value;
    volatile HashEntry<K, V> next;
    //其他省略
}

我們說Segment類似哈希表，那么一些屬性就跟我們之前提到的HashMap差不多，比如負(fù)載因子loadFactor，比如閾值threshold等等，看下Segment的構(gòu)造方法。

Segment(float lf, int threshold, HashEntry<K, V>[] tab) {
    this.loadFactor = lf;//負(fù)載因子
    this.threshold = threshold;//閾值
    this.table = tab;//主干數(shù)組即 HashEntry 數(shù)組
}

我們來看下ConcurrentHashMap的構(gòu)造方法

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException();
    //MAX_SEGMENTS 為 1<<16=65536，也就是最大并發(fā)數(shù)為 65536
    if (concurrencyLevel > MAX_SEGMENTS) concurrencyLevel = MAX_SEGMENTS;
    //2 的 sshif 次方等于 ssize，例:ssize=16，sshift=4;ssize=32，sshif=5
    int sshift = 0;
    //ssize 為 segments 數(shù)組長度，根據(jù) concurrentLevel 計(jì)算得出
    int ssize = 1;
    while (ssize < concurrencyLevel) {
        ++sshift;
        ssize <<= 1;
    }
    //segmentShift 和 segmentMask 這兩個(gè)變量在定位 segment 時(shí)會(huì)用到，后面會(huì)詳細(xì)講
    this.segmentShift = 32 - sshift;
    this.segmentMask = ssize - 1;
    if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY;
    //計(jì)算 cap 的大小，即 Segment 中 HashEntry 的數(shù)組長度，cap 也一定為 2 的 n 次方.
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c) cap <<= 1;
    //創(chuàng)建 segments 數(shù)組并初始化第一個(gè) Segment，其余的 Segment 延遲初始化
    Segment<K, V> s0 = new Segment<K, V>(loadFactor, (int) (cap * loadFactor), (HashEntry<K, V>[]) new HashEntry[cap]);
    Segment<K, V>[] ss = (Segment<K, V>[]) new Segment[ssize];
    UNSAFE.putOrderedObject(ss, SBASE, s0);
    this.segments = ss;
}

初始化方法有三個(gè)參數(shù)，如果用戶不指定則會(huì)使用默認(rèn)值，initialCapacity為16，loadFactor為0.75(負(fù)載因子，擴(kuò)容時(shí)需要參考)，concurrentLevel為16。

從上面的代碼可以看出來，Segment數(shù)組的大小ssize是由concurrentLevel來決定的，但是卻不一定等于concurrentLevel，ssize一定是大于或等于concurrentLevel的最小的2的次冪。比如：默認(rèn)情況下concurrentLevel是16，則ssize為16;若concurrentLevel為14，ssize為16;若concurrentLevel為17，則ssize為32。為什么Segment的數(shù)組大小一定是2的次冪?其實(shí)主要是便于通過按位與的散列算法來定位Segment的index。其實(shí)，put方法對segment也會(huì)有所體現(xiàn)。

public V put(K key, V value) {
    Segment<K, V> s;
    //concurrentHashMap 不允許 key/value 為空
    if (value == null)
        throw new NullPointerException();
    //hash 函數(shù)對 key 的 hashCode 重新散列，避免差勁的不合理的 hashcode，保證散列均勻
    int hash = hash(key);
    //返回的 hash 值無符號(hào)右移 segmentShift 位與段掩碼進(jìn)行位運(yùn)算，定位 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);
}

● 從源碼看出，put的主要邏輯也就兩步：

定位segment并確保定位的Segment已初始化。

調(diào)用Segment的put方法。

Ps：關(guān)于segmentShift和segmentMask

segmentShift和segmentMask這兩個(gè)全局變量的主要作用是用來定位Segment。

int j = (hash >>> segmentShift) & segmentMask;

segmentMask：段掩碼，假如segments數(shù)組長度為16，則段掩碼為16-1=15;segments長度為32，段掩碼為32-1=31。這樣得到的所有bit位都為1，可以更好地保證散列的均勻性。

segmentShift：2的sshift次方等于ssize，segmentShift=32-sshift。若segments長度為16，segmentShift=32-4=28;若segments長度為32，segmentShift=32-5=27。而計(jì)算得出的hash值最大為32位，無符號(hào)右移segmentShift，則意味著只保留高幾位(其余位是沒用的)，然后與段掩碼segmentMask位運(yùn)算來定位Segment。

ConcurrentLinkedQuere 方法。

● Get 操作：

public V get(Object key) {
    Segment<K, V> s;
    HashEntry<K, V>[] tab;
    int h = hash(key);
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    //先定位 Segment,再定位 HashEntry
    if ((s = (Segment<K, V>) UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) {
        for (HashEntry<K, V> e = (HashEntry<K, V>) UNSAFE.getObjectVolatile(tab, ((long) (((tab.length - 1) & h)) << TSHIFT) + TBASE);
             e != null; e = e.next) {
            K k;
            if ((k = e.key) == key || (e.hash == h && key.equals(k))) return e.value;
        }
    }
    return null;
}

get方法無需加鎖，由于其中涉及到的共享變量都使用volatile修飾，volatile可以保證內(nèi)存可見性，所以不會(huì)讀取到過期數(shù)據(jù)。

來看下concurrentHashMap代理到Segment上的put方法，Segment中的put方法是要加鎖的。只不過是鎖粒度細(xì)了而已。

● Put 操作：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry<K, V> node = tryLock() ? null :
            scanAndLockForPut(key, hash, value);
    //tryLock 不成功時(shí)會(huì)遍歷定位到的 HashEnry 位置的鏈表（遍歷主要是為了使 CPU 緩存鏈表）,
    // 若找不到,則創(chuàng)建 HashEntry。tryLock 一定次數(shù)后（MAX_SCAN_RETRIES 變量決定）,則lock。
    // 若遍歷過程中,由于其他線程的操作導(dǎo)致鏈表頭結(jié)點(diǎn)變化,則需要重新遍歷。
    V oldValue;
    try {
        HashEntry<K, V>[] tab = table;
        int index = (tab.length - 1) & hash;//定位 HashEntry,可以看到,這個(gè) hash 值在定位 Segment
        //時(shí)和在 Segment 中定位 HashEntry 都會(huì)用到,只不過定位 Segment 時(shí)只用到高幾位。
        HashEntry<K, V> first = entryAt(tab, index);
        for (HashEntry<K, V> e = first; ; ) {
            if (e != null) {
                K k;
                if ((k = e.key) == key ||
                        (e.hash == hash && key.equals(k))) {
                    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;
                //若 c 超出閾值 threshold,需要擴(kuò)容并 rehash。擴(kuò)容后的容量是當(dāng)前容量的 2 倍。這樣可以最大程
                //度避免之前散列好的 entry 重新散列,具體在另一篇文章中有詳細(xì)分析,不贅述。擴(kuò)容并 rehash 的這個(gè)過程是比較消耗資源的。
                if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node);
                else
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

ConcurrentLinkedQuere 總結(jié)

ConcurrentHashMap作為一種線程安全且高效的哈希表的解決方案，尤其是其中的“分段鎖”的方案，相比HashTable的全表鎖在性能上的提升非常之大。

以上就是“新年求職大禮，多線程面試題及答案”，你能回答上來嗎?如果想要了解更多的Java面試題相關(guān)內(nèi)容，可以關(guān)注動(dòng)力節(jié)點(diǎn)Java官網(wǎng)。