ConcurrentHashMap的實現原理與使用

ConcurrentHashMap是線程安全且高效的hashmap。本節讓我們一起研究一下該容器是如何在保證線程安全的同時又能保證高效的操作。

為什麽要使用ConcurrentHashMap

在並發編程中使用HashMap可能導致程序死循環。而使用線程安全的HashTable效率又非常低下，基於以上兩個原因，便有了ConcurrentHashMap的登場機會。

（1）線程不安全的HashMap

在多線程環境下，使用HashMap進行put操作會引起死循環，導致CPU利用率接近100%，所以在並發情況下不能使用HashMap。例如一下代碼

    final
 
 HashMap<String, String> map = new HashMap<>(2);
    Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
            for (int i = 0; i < 10000; i++) {
                new Thread(new Runnable() {
                    @Override
                    public void
 
 run() {
                        map.put(UUID.randomUUID().toString(), "");
                    }
                }, "ftf" + i).start();
            }
        }
    }, "ftf");
    thread.start;
    thread.join;

HashMap在並發執行put操作時會引起死循環，是因為多線程會導致HashMap的Entry鏈表形成環形數據結構，一旦形成環形數據結構，Entry的next節點永遠不會為空，就會產生死循環獲取Entry.

（2）效率低下的HashTable

HashTable容器使用synchronized來保證線程安全，但在線程競爭激烈的情況下HashTable的效率非常低下。因為當一個線程訪問HashTable的同步方法，其他線程也訪問HashTable的同步方法時，會進入阻塞或輪詢狀態。

（3）ConcurrentHashMap的鎖分段技術可有效提升並發訪問率

HashTable容器在競爭激烈的並發環境下表現出效率低下的原因是所有訪問HashTable的線程都必須競爭同一把鎖，假如容器裏有多把鎖，每一把鎖用於鎖容器的一部分數據，那麽當多線程訪問容器裏不同數據段的數據時，線程間就不會存在鎖競爭，從而可以有效提高並發訪問率，這就是ConcurrentHashMap所使用的鎖分段技術。首先將數據分成一段一段地存儲，然後給每一段數據配一把鎖，當一個線程占用鎖訪問其中一個段數據的時候，其他數據段也能被其他線程訪問。

ConcurrentHashMap的結構

ConcurrentHashMap是由Segment數組結構和HashEntry數組結構組成。Segment是一種可重入鎖（ReentrantLock），在ConcurrentHashMap裏扮演鎖的角色；HashEntry則用於儲存鍵值對數據。一個ConcurrentHashMap裏包含一個Segment數組。Segment的結構和HashEntry類似，是一種數組和鏈表結構。一個Segment裏包含一個HashEntry數組，每個HashEntry是一個鏈表結構的元素，每個Segment守護著一個HashEntry數組裏的元素，當對HashEntry數組的數據進行修改時，必須首先獲得與它對於的Segment鎖。

ConcurrentHashMap的初始化

ConcurrentHashMap初始化方法是通過initialCapacity，loadFactor和concurrencyLevel等幾個參數來初始化Segment數組、段偏移量segmentShift、段掩碼segmentMask和每個segment裏的HashEntry數組來實現的。

初始化segments數組：讓我們來看一下初始化segments數組的源代碼

if(concurrencyLevel > MAX_SEGMENTS)  concurrencyLevel = MAX_SEGMENTS 
 int sshift = 0;
        int ssize = 1;
        while (ssize < DEFAULT_CONCURRENCY_LEVEL) {
            ++sshift;
            ssize <<= 1;
        }
        int segmentShift = 32 - sshift;
        int segmentMask = ssize - 1;
this.segments = Segment.newArray(ssize);

由上面代碼可知，segments數組的長度ssize是通過concurrencyLevel計算得出的。為了能通過按位與的散列算法來定位segments數組的索引，必須保證segments數組的長度是2的N次方，所以必須計算出一個大於或等於concurrencyLevel的最小的2的N次方值來作為segments數組的長度。concurrencyLevel的最大值是65535，這意味著segments數組的長度最大為65536，對應的二進制是16位。

初始化segmentShift和segmentMask：這兩個全局變量需要在定位segment時的散列算法裏使用，sshift等於ssize從1向左移位的次數，在默認情況下concurrencyLevel等於16,1需要向左移位移動4次，所以sshift等於4。segmentShift用於定位參與散列運算的位數，segmentShift等於32減去sshift，所以等於28，這裏之所以用32是因為ConcurrentHashMap裏的hash()方法輸出的最大數是32位的。segmentMask是散列運算的掩碼，等於ssize減1，即15，掩碼的二進制各個位的值都是1.因為ssize的最大長度是65536，所以segmentShift最大值是16，segmentMask最大值是65535，對應的二進制是16位，每個位都是1.

初始化每個segment：輸入參數initalCapacity是ConcurrentHashMap的初始化容量，loadfactor是每個segment的負載因子，在構造方法裏需要通過兩個參數來初始化數組中的每個segment。

定位segment

既然ConcurrentHashMap使用分段鎖Segment來保護不同段的數據，那麽在插入和獲取元素的時候，必須先通過散列算法定位到segment。ConcurrentHashMap會首先使用Wang/Jenkins hash的變種算法對元素的hashCode進行一次再散列。之所以進行再散列，目的是減少散列沖突，使元素能夠均勻地分布在不同的Segment上，從而提高容器的存取效率。假如散列的質量差到極點，那麽所有的元素都在一個Segment中，不僅存取元素緩慢，分段鎖也會失去意義。

默認情況下segmentShift為28，segmentMask為15，再散列後的數量最大是32位二進制數據，向右無符號移動28位，意思是讓高4位參與到散列運算中，（hash>>>segmentShit）&segmentMask的運算結果分別是4,15,7和8，可以看到散列值沒有發生沖突。

ConcurrentHashMap的操作

get操作：Segment的get操作實現非常簡單和高效。先經過一次再散列，然後使用這個散列值通過散列運算定位到Segment，再通過散列算法定位到元素，代碼如下

public V get(Object key){
 int hash = hash(key.hashCode());
 return segmentFor(hash).get(key,hahsh)   
}

get操作的高效之處在於整個get過程不需要加鎖，除非讀到的值是空才會加鎖重讀。它的get方法裏將要使用的共享變量都定義成volatile類型，如用於統計當前Segment大小的count字段和用於存儲值的HashEntry的value，定義成volatile的變量，能夠在線程之間保持可見性，能夠被多線程同時讀，並且保證不會讀到過期的值，但是只能被單線程寫（有一種情況可以被多線程寫，就是寫入的值不依賴與原值），在get操作裏只需要讀不需要寫共享變量count和value，所以可以不用加鎖。之所以不會讀到過期的值，是因為根據Java內存模型的happen before原則，對volatile字段的寫入操作先於讀操作，即使兩個線程同時修改和獲取volatile變量，get操作也能拿到最新的值，這是用volatiel替換鎖的經典應用場景。

transient volatile int count;
volatile V value;

在定位元素的代碼裏我們可以發現，定位HashEntry和定位Segment的散列算法雖然一樣，都與數組的長度減去1再相“與”，但是想“與”的值不一樣，定位segment使用的是元素的hashcode通過再散列後得到的值的高位，而定位HashEntry直接使用的是再散列後的值。其目的是避免兩次散列後的值一樣，雖然元素在Segment裏散列開了，但是卻沒有再HashEntry裏散列開。

put操作：由於put方法裏需要對共享變量進行寫入操作，所以為了線程安全，在操作共享變量時必須加鎖。put方法首先定位到Segment，然後再Segment裏進行插入操作。插入操作需要經歷兩個步驟，第一步判斷是否需要對Segment裏的HashEntry數組進行擴容，第二步添加元素的位置，然後將其放在HashEntry裏。

（1）是否需要擴容：在插入元素前會先判斷Segment裏的HashEntry數組是否超過容量，如果超過闕值，則對數組進行擴容。值得一提的是，Segment的擴容判斷比HashMap更恰當，因為HashMap是在插入元素後判斷元素是否已經到達容量的，如果達到了就進行擴容，但是很有可能擴容之後沒有新元素插入，這時HashMap就進行了一次無效的擴容，

（2）如何擴容：在擴容的時候，首先會創建一個容量是原來容量兩倍的數組，然後將原數組裏的元素進行再散列後插入到新的數組裏。為了高效，ConcurrentHashMap不會對整個容器進行擴容，而只對某個segment進行擴容。

（3）size操作：如果要統計整個ConcurrentHashMap裏元素的大小，就必須統計所有Segment裏元素的大小後求和。Segment裏的全局變量count是一個colatile變量，那麽在多線程場景下，是不是直接把所有的Segment的count相加就可以得到整個ConcurrentHashMap大小呢？不是的，雖然相加時可以獲取每個Segment的count最新值，但是可能累加前使用count發生了變化，那麽統計結果就不準了。所以，最安全的做法是在統計size的時候把所有Segment的put，remone和clean方法全部鎖住，但是這種做法顯然非常抵消。因為在累加count操作過程中，之前累加過的count發生變化的幾率非常小，所以ConcurrentHashMap的做法是先嘗試2次通過不鎖住Segment的方式來統計各個Segment大小，如果統計的過程中，容器的count發生了變化，則再采用加鎖的方式來統計所有Segment大小。ConcurrentHashMap如何判斷統計時候發生了變化呢？使用modCount變量，在put，remove和clean方法裏操作元素前都會將變成modCount進行加1，那麽在統計size前後比較modCount是否發生變化，從而得知容器的大小是否發生變化。

ConcurrentLinkedQueue

在並發編程中，有時候需要使用線程安全的隊列。如果要實現一個線程安全的隊列有兩種方式：一種是使用阻塞算法，另一種是使用非阻塞算法。使用阻塞算法的隊列可以用一個鎖（入隊和出隊用同一把鎖）或兩個鎖（入隊和出隊用於不同的鎖）等方式實現。非阻塞的實現方式則可以使用循環CAS的方式實現。

ConcurrentLinkedQueue是一個基於鏈接節點的無界線程安全隊列，它采用先進先出的規則對節點進行排序，當我們添加一個元素的時候，它會添加到隊列的尾部；當我們獲取一個元素時候，它會返回隊列頭部的元素。它采用了“wait-free”算法（即CAS算法）來實現，概算在Michael&Scoft算法上進行了一些修改。

第六章 Java並發容器和框架