首先常用三種HashMap包括HashMap,HashTable和CocurrentHashMap：

• HashMap在併發程式設計過程中使用可能導致死迴圈，因為插入過程不是原子操作，每個HashEntry是一個連結串列節點，很可能在插入的過程中，已經設定了後節點，實際還未插入，最終反而插入在後節點之後，造成鏈中出現環，破壞了連結串列的性質，失去了尾節點，出現死迴圈。
• HashTable因為內部是採用synchronized來保證執行緒安全的，但線上程競爭激烈的情況下HashTable的效率下降得很快因為synchronized關鍵字會造成程式碼塊或方法成為為臨界區(對同一個物件加互斥鎖)，當一個執行緒訪問臨界區的程式碼時，其他執行緒也訪問同一臨界區時，會進入阻塞或輪詢狀態。究其原因，實際上是有獲取鎖意向的執行緒的數目增加，但是鎖還是隻有單個，導致大量的執行緒處於輪詢或阻塞，導致同一時間段有效執行的執行緒的增量遠不及執行緒總體增量。 
  
  • 在查詢時，尤其能夠體現出CocurrentHashMap在效率上的優勢，HashTable使用Sychronized關鍵字，會導致同時只能有一個查詢在執行，而Cocurrent則不採取加鎖的方法，而是採用volatile關鍵字，雖然也會犧牲效率，但是由於Sychronized，於該文末尾繼續討論。
• CocurrentHashMap利用鎖分段技術增加了鎖的數目，從而使爭奪同一把鎖的執行緒的數目得到控制。 
  
  • 鎖分段技術就是對資料集進行分段，每段競爭一把鎖，不同資料段的資料不存在鎖競爭，從而有效提高 高併發訪問效率。
  • CocurrentHashMap在get方法是無需加鎖的，因為用到的共享變數都採用volatile關鍵字修飾，巴證共享變數線上程之間的可見性(每次讀取都先同步快取和記憶體，直接從記憶體中獲取值，雖然不是原子操作，但根據JAVA記憶體模型的happen before原則，對volatile欄位的寫入操作先於讀操作，能夠保證不會髒讀),volatile為了讓變數提供執行緒之間的記憶體可見性，會禁止程式執行結果的重排序（導致快取優化的效果降低）

CocurrentHashMap的結構

• CocurrentHashMap是由Segment陣列和HashEntry陣列組成。
• Segment是重入鎖(ReentrantLock),作為一個數據段競爭鎖，每個HashEntry一個連結串列結構的元素，利用Hash演算法得到索引確定歸屬的資料段，也就是對應到在修改時需要競爭獲取的鎖。

segments陣列的初始化

• 首先簡單描述一下原始碼中變數的含義:

• 首先計算出segment陣列的長度ssize，並且計算出與ssize關聯的segmentShift和segmentMask

if ( cocurrencyLevel > MAX_SEGMENTS )
    cocurrencyLevel = MAX_SEGMENTS;
//如果cocurrencyLevel大於上限，那麼取值為上限，
//上限定義為65535，決定了重入鎖的segments的數目
int sshift = 0;
int ssize = 1;
//找到大於等於concurrencyLevel的最小的2^n作為segments的大小
while ( ssize < concurrencyLevel ){
    ++sshift;//記錄偏移量，為了以後通過與運算獲取segment的索引
    ssize <<=1;
}
segmentShift = 32 - sshift;//說明只有高sshift位作為segment的索引
segmentMask = ssize - 1;//能直接通過與運算獲取segment的索引
this.segments = Segment.newArray(ssize);
//靜態工廠方法構造ssize大小的segment陣列

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15

• 初始化每個segment,因為已經知道segment陣列的規模，將ConcurrentHashMap的邏輯上持有的HashEntry均分到每個Segment上，因為是雜湊，所以要loadFactor來定義負載率，來保證segment適時的拓充，來避免散列表和資料規模相近時，衝突加重的風險。

if ( initialCapacity > MAXIMUM_CAPACITY )
    initialCapacity = MAXIMUM_CAPACITY;
int c = initCapacity / ssize;
//整個cocurrentHashMap的容量由所有的segment均攤
if ( c * ssize < initCapacity )
    ++c;
int cap = 1;//segment中的hashEntry陣列的長度
while ( cap < c )
    cap <<= 1;
//設定loadFactor,保證雜湊的高效性的同時也保證空間浪費相對有限
for ( int i = 0 ; i < this.segments.length ; i++ )
    this.segments[i] =  new Segments<K,V>(cap , loadFactor);
//最終計算出segment的容量threshold=(int)cap*loadFactor;

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13

• 定位segment，定位segment尤其重要，如果太多元素集中在少數幾個segment中會導致CocurrentHashMap的效率得不到優化，因為同一個segment中的修改還是要競爭鎖，所以選擇合適的hash演算法儘可能地將元素分到不同的segment中是目標。 
  
  • CocurrentHashMap採用的是對元素的HashCode進行再Hash來減少衝突
  • CocurrentHashMap採用的是根據Wang/JenkinsHash改進的hash演算法，該演算法具有雪崩性(引數一位數字變化，結果都有半數左右(二進位制位)發生變化)

final Segment<K,V> segmentFor ( int hash ){
    //首先根據segmentShift無符號右移，得到表示segment所以的高位，
    //然後與掩碼邏輯與得到segment的索引，定位到segment
    return segments[(hash>>>segmentShift)&segmentMask];
}

• 1
• 2
• 3
• 4
• 5

CocurrentHashMap的操作

• Segment的get操作是不需要加鎖的。因為volatile修飾的變數保證了執行緒之間的可見性
• Segment的put操作是需要加鎖的，在插入時會先判斷Segment裡的HashEntry陣列是否會超過容量(threshold),如果超過需要對陣列擴容，翻一倍。然後在新的陣列中重新hash，為了高效，CocurrentHashMap只會對需要擴容的單個Segment進行擴容
• CocurrentHashMap獲取size的時候要統計Segments中的HashEntry的和，如果不對他們都加鎖的話，無法避免資料的修改造成的錯誤，但是如果都加鎖的話，效率又很低。所以CoccurentHashMap在實現的時候，巧妙地利用了在累加過程中發生變化的機率很小的客觀條件，在獲取count時，不加鎖的計算兩次，如果兩次不相同，在採用加鎖的計算方法。採用了一個高效率的剪枝防止很大概率地減少了不必要額加鎖。

• 主要研究ConcurrentHashMap的3種操作——get操作、put操作和size操作.

  5.1 get操作

  Segment的get操作實現非常簡單和高效. 
  - 先經過一次再雜湊 
  - 然後使用這個雜湊值通過雜湊運算定位到Segment 
  - 再通過雜湊演算法定位到元素.

  public V get(Object key) {
      Segment<K,V> s; 
      HashEntry<K,V>[] tab;
      int h = hash(key);
  //找到segment的地址 long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
  //取出segment，並找到其hashtable 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;
  }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

  整個get方法不需要加鎖，只需要計算兩次hash值，然後遍歷一個單向連結串列（此連結串列長度平均小於2），因此get效能很高。 
  高效之處在於整個過程不需要加鎖,除非讀到的值是空才會加鎖重讀. 
  HashTable容器的get方法是需要加鎖的,那ConcurrentHashMap的get操作是如何做到不加鎖的呢? 
  原因是它的get方法將要使用的共享變數都定義成了volatile型別, 
  如用於統計當前Segement大小的count欄位和用於儲存值的HashEntry的value.定義成volatile的變數,能夠線上程之間保持可見性,能夠被多執行緒同時讀,並且保證不會讀到過期的值,但是隻能被單執行緒寫(有一種情況可以被多執行緒寫,就是寫入的值不依賴於原值), 
  在get操作裡只需要讀不需要寫共享變數count和value,所以可以不用加鎖. 
  之所以不會讀到過期的值,是因為根據Java記憶體模型的happen before原則,對volatile欄位的寫操作先於讀操作,即使兩個執行緒同時修改和獲取 
  volatile變數,get操作也能拿到最新的值, 
  這是用volatile替換鎖的經典應用場景.

  transient volatile int count;
  volatile V value;

  • 1
  • 2

  在定位元素的程式碼裡可以發現,定位HashEntry和定位Segment的雜湊演算法雖然一樣,都與陣列的長度減去1再相“與”,但是相“與”的值不一樣

  • 定位Segment使用的是元素的hashcode再雜湊後得到的值的高位
  • 定位HashEntry直接使用再雜湊後的值.

  其目的是避免兩次雜湊後的值一樣,雖然元素在Segment裡雜湊開了,但是卻沒有在HashEntry裡雜湊開.

  hash >>> segmentShift & segmentMask　　 // 定位Segment所使用的hash演算法
  int index = hash & (tab.length - 1);　　 // 定位HashEntry所使用的hash演算法

  • 1
  • 2

  5.2 put操作

  由於需要對共享變數進行寫操作,所以為了執行緒安全,在操作共享變數時必須加鎖. 
  put方法首先定位到Segment,然後在Segment裡進行插入操作. 
  插入操作需要經歷兩個步驟

  • 判斷是否需要對Segment裡的HashEntry陣列進行擴容

  • 定位新增元素的位置,然後將其放在HashEntry數組裡

    1. 是否需要擴容 
       在插入元素前會先判斷Segment裡的HashEntry陣列是否超過容量(threshold),如果超過閾值,則對陣列進行擴容. 
       值得一提的是,Segment的擴容判斷比HashMap更恰當,因為HashMap是在插入元素後判斷元素是否已經到達容量的,如果到達了就進行擴容,但是很有可能擴容之後沒有新元素插入,這時HashMap就進行了一次無效的擴容.
    2. 如何擴容 
       在擴容的時候,首先會創建一個容量是原來兩倍的陣列,然後將原數組裡的元素進行再雜湊後插入到新的數組裡. 
       為了高效,ConcurrentHashMap不會對整個容器進行擴容,而只對某個segment擴容.

  put方法的第一步，計算segment陣列的索引，並找到該segment，然後呼叫該segment的put方法。

  public V put(K key, V value) {
      Segment<K,V> s;
      if (value == null)
          throw new NullPointerException();
      int hash = hash(key);
  //計算segment陣列的索引，並找到該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);
  //呼叫該segment的put方法 return s.put(key, hash, value, false);
  }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11

  put方法第二步，在Segment的put方法中進行操作。

  final V put(K key, int hash, V value, boolean onlyIfAbsent) {
  //呼叫tryLock()嘗試加鎖，若失敗則呼叫scanAndLockForPut進行加鎖，同時尋找key相應的節點node
      HashEntry<K,V> node = tryLock() ? null :
          scanAndLockForPut(key, hash, value);
  //以下的程式碼都執行在加鎖狀態
      V oldValue;
      try {
          HashEntry<K,V>[] tab = table;
  //計算hash表的索引值，並取出HashEntry int index = (tab.length - 1) & hash;
          HashEntry<K,V> first = entryAt(tab, index);
  //遍歷此連結串列 for (HashEntry<K,V> e = first;;) {
  //如果連結串列不為空，在連結串列中尋找對應的node，找到後進行賦值，並退出迴圈 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;
              }
  //如果在連結串列中沒有找到對應的node else {
  //如果scanAndLockForPut方法中已經返回的對應的node，則將其插入first之前 if (node != null)
                      node.setNext(first);
                  else //否則，new一個新的HashEntry
                      node = new HashEntry<K,V>(hash, key, value, first);
                  int c = count + 1;
  //測試是否需要自動擴容 if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                      rehash(node);
                  else //設定node到Hash表的index索引處
                      setEntryAt(tab, index, node);
                  ++modCount;
                  count = c;
                  oldValue = null;
                  break;
              }
          }
      } finally {
          unlock();
      }
      return oldValue;
  }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45

  5.3 size操作

  如果要統計整個ConcurrentHashMap裡元素的大小,就必須統計所有Segment裡元素的大小後求和. 
  Segment裡的全域性變數count是一個volatile變數,那在多執行緒場景下,是不是直接把所有Segment的count相加就可以得到整個ConcurrentHashMap大小了呢?不是的，雖然相加時可以獲取每個Segment的count的最新值,但是可能累加前使用的count發生了變化,那麼統計結果就不準了.所以,最安全的做法是在統計size的時候把所有Segment的put、remove和clean方法全部鎖住,但是這種做法顯然非常低效.

  因為在累加count操作過程中,之前累加過的count發生變化的機率非常小,所以 
  ConcurrentHashMap的做法是先嚐試2次通過不鎖住Segment的方式來統計各個Segment大小,如果統計的過程中,容器的count發生了變化,則再採用加鎖的方式來統計所有Segment的大小.

  那麼ConcurrentHashMap又是如何判斷在統計的時候容器是否發生了變化呢? 
  使用modCount變數,在put、remove和clean方法裡操作元素前都會將變數modCount進行加1,那麼在統計size前後比較modCount是否發生變化,從而得知容器的大小是否發生變化.

  
  

一點理解

synchronized(其實感覺是可以被重入鎖和Condition完全取代的)和volatile的取捨：

• 首先的區別就在於是否是原子操作，也就是單一的不可分割的操作，在多執行緒中，原子操作能夠保證不受到其他執行緒的影響
• synchonized就實現了原子性操作，不同的執行緒互斥地進入臨界程式碼區，而且是記憶體可見的，也就是每個執行緒進入臨界區時，都是從記憶體中獲取的值，不會因為快取而出現髒讀。
• volatile實現了記憶體可見性，會將修改的值直接寫入內容，並且登出掉之前對於該變數的快取，而且禁止了指令的排序。但是它不是原子操作！！