JDK1.8中的ConcurrentHashMap使用及場(chǎng)景分析

更新時(shí)間：2021年01月05日 14:37:54 作者：吳振照

這篇文章主要介紹了JDK1.8中的ConcurrentHashMap使用及場(chǎng)景分析,本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

ConcurrentHashMap 的初步使用及場(chǎng)景：

　　ConcurrentHashMap 是 J.U.C 包里面提供的一個(gè)線程安全并且高效的 HashMap，所以ConcurrentHashMap 在并發(fā)編程的場(chǎng)景中使用的頻率比較高，那么我們就從ConcurrentHashMap 的使用上以及源碼層面來(lái)分析 ConcurrentHashMap 到底是如何實(shí)現(xiàn)安全性的

api 使用：

　　ConcurrentHashMap 是 Map 的派生類(lèi)，所以 api 基本和 Hashmap 是類(lèi)似，主要就是 put、get 這些方法，接下來(lái)基于 ConcurrentHashMap 的 put 和 get 這兩個(gè)方法作為切入點(diǎn)來(lái)分析 ConcurrentHashMap 的源碼實(shí)現(xiàn)。

　　ConcurrentHashMap 和 HashMap 的實(shí)現(xiàn)原理是差不多的，但是因?yàn)?ConcurrentHashMap需要支持并發(fā)操作，所以在實(shí)現(xiàn)上要比 hashmap 稍微復(fù)雜一些。在 JDK1.7 的實(shí)現(xiàn)上， ConrruentHashMap 由一個(gè)個(gè) Segment 組成，簡(jiǎn)單來(lái)說(shuō)，ConcurrentHashMap 是一個(gè) Segment 數(shù)組，它通過(guò)繼承 ReentrantLock 來(lái)進(jìn)行加鎖，通過(guò)每次鎖住一個(gè) segment來(lái)保證每個(gè) segment內(nèi)的操作的線程安全性從而實(shí)現(xiàn)全局線程安全。整個(gè)結(jié)構(gòu)圖如下：

　　當(dāng)每個(gè)操作分布在不同的 segment 上的時(shí)候，默認(rèn)情況下，理論上可以同時(shí)支持 16 個(gè)線程的并發(fā)寫(xiě)入。相比于 1.7 版本，它做了兩個(gè)改進(jìn)

取消了 segment 分段設(shè)計(jì)，直接使用 Node 數(shù)組來(lái)保存數(shù)據(jù)，并且采用 Node 數(shù)組元素作為鎖來(lái)實(shí)現(xiàn)每一行數(shù)據(jù)進(jìn)行加鎖來(lái)進(jìn)一步減少并發(fā)沖突的概率將原本數(shù)組+單向鏈表的數(shù)據(jù)結(jié)構(gòu)變更為了數(shù)組+單向鏈表+紅黑樹(shù)的結(jié)構(gòu)。為什么要引入紅黑樹(shù)呢？在正常情況下，key hash 之后如果能夠很均勻的分散在數(shù)組中，那么 table 數(shù)組中的每個(gè)隊(duì)列的長(zhǎng)度主要為 0 或者 1.但是實(shí)際情況下，還是會(huì)存在一些隊(duì)列長(zhǎng)度過(guò)長(zhǎng)的情況。如果還采用單向列表方式，那么查詢(xún)某個(gè)節(jié)點(diǎn)的時(shí)間復(fù)雜度就變?yōu)?O(n); 因此對(duì)于隊(duì)列長(zhǎng)度超過(guò) 8 的列表，JDK1.8 采用了紅黑樹(shù)的結(jié)構(gòu)，那么查詢(xún)的時(shí)間復(fù)雜度就會(huì)降低到O(logN),可以提升查找的性能；

　　這個(gè)結(jié)構(gòu)和 JDK1.8 版本中的 Hashmap 的實(shí)現(xiàn)結(jié)構(gòu)基本一致，但是為了保證線程安全性，ConcurrentHashMap 的實(shí)現(xiàn)會(huì)稍微復(fù)雜一下。接下來(lái)我們從源碼層面來(lái)了解一下它的原理.我們基于 put 和 get 方法來(lái)分析它的實(shí)現(xiàn)即可。

put 方法第一階段：

public V put(K key, V value) {
    return putVal(key, value, false);
  }

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
　　if (key == null || value == null) throw new NullPointerException();
　　int hash = spread(key.hashCode());//計(jì)算 hash 值
　　int binCount = 0;//用來(lái)記錄鏈表的長(zhǎng)度
　　for (Node<K,V>[] tab = table;;) {//這里其實(shí)就是自旋操作，當(dāng)出現(xiàn)線程競(jìng)爭(zhēng)時(shí)不斷自旋
　　　　Node<K,V> f; int n, i, fh;
　　　　if (tab == null || (n = tab.length) == 0)//如果數(shù)組為空，則進(jìn)行數(shù)組初始 化
　　　　　　tab = initTable();//初始化數(shù)組
　　　　　　//通過(guò) hash 值對(duì)應(yīng)的數(shù)組下標(biāo)得到第一個(gè)節(jié)點(diǎn); 以 volatile 讀的方式來(lái)讀取 table 數(shù)
　　　　　　//組中的元素，保證每次拿到的數(shù)據(jù)都是最新的
　　　　　　//(Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
　　　　else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
　　　　　　 //如果該下標(biāo)返回的節(jié)點(diǎn)為空，則直接通過(guò) cas 將新的值封裝成 node 插入即可；
　　　　　　//如果 cas 失敗，說(shuō)明存在競(jìng)爭(zhēng)，則進(jìn)入下一次循環(huán)
　　　　　　if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))
　　　　　　　　break; // no lock when adding to empty bin
　　　　　　}　　　　}
    .......
}

　　假如在上面這段代碼中存在兩個(gè)線程，在不加鎖的情況下：線程 A 成功執(zhí)行 casTabAt 操作后，隨后的線程 B 可以通過(guò) tabAt 方法立刻看到 table[i]的改變。原因如下：線程 A 的casTabAt 操作，具有 volatile 讀寫(xiě)相同的內(nèi)存語(yǔ)義，根據(jù) volatile 的 happens-before 規(guī)則：線程 A 的 casTabAt 操作，一定對(duì)線程 B 的 tabAt 操作可見(jiàn)。

initTable()：

　　數(shù)組初始化方法，這個(gè)方法比較簡(jiǎn)單，就是初始化一個(gè)合適大小的數(shù)組sizeCtl 這個(gè)要單獨(dú)說(shuō)一下，如果沒(méi)搞懂這個(gè)屬性的意義，可能會(huì)被搞暈這個(gè)標(biāo)志是在 Node 數(shù)組初始化或者擴(kuò)容的時(shí)候的一個(gè)控制位標(biāo)識(shí)，負(fù)數(shù)代表正在進(jìn)行初始化或者擴(kuò)容操作

-1 代表正在初始化
-N 代表有 N-1 有二個(gè)線程正在進(jìn)行擴(kuò)容操作，這里不是簡(jiǎn)單的理解成 n 個(gè)線程，sizeCtl 就是-N,這塊后續(xù)在講擴(kuò)容的時(shí)候會(huì)說(shuō)明
0 標(biāo)識(shí) Node 數(shù)組還沒(méi)有被初始化，正數(shù)代表初始化或者下一次擴(kuò)容的大小

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
      if ((sc = sizeCtl) < 0)//被其他線程搶占了初始化的操作,則直接讓出自己的 CPU 時(shí)間片
        Thread.yield(); // lost initialization race; just spin
      //通過(guò) cas 操作，將 sizeCtl 替換為-1，標(biāo)識(shí)當(dāng)前線程搶占到了初始化資格
      //第一次進(jìn)來(lái)初始化一定走這里
      else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
        try {
          if ((tab = table) == null || tab.length == 0) {
            //默認(rèn)初始容量為 16
            int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
            @SuppressWarnings("unchecked")
            //初始化數(shù)組，長(zhǎng)度為 16，或者初始化在構(gòu)造 ConcurrentHashMap 的時(shí)候傳入的長(zhǎng)度
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
            table = tab = nt;//將這個(gè)數(shù)組賦值給 table
            //計(jì)算下次擴(kuò)容的大小，實(shí)際就是當(dāng)前容量的 0.75倍，這里使用了右移來(lái)計(jì)算
            //sc =12
            sc = n - (n >>> 2);
          }
        } finally {
          //設(shè)置 sizeCtl 為 sc, 如果默認(rèn)是 16 的話，那么這個(gè)時(shí)候sc=16*0.75=12
          sizeCtl = sc;
        }
        break;
      }
    }
    return tab;
}

tabAt()：

　　該方法獲取對(duì)象中offset偏移地址對(duì)應(yīng)的對(duì)象field的值。實(shí)際上這段代碼的含義等價(jià)于tab[i],但是為什么不直接使用 tab[i]來(lái)計(jì)算呢？getObjectVolatile，一旦看到 volatile 關(guān)鍵字，就表示可見(jiàn)性。因?yàn)閷?duì) volatile 寫(xiě)操作 happen-before 于 volatile 讀操作，因此其他線程對(duì) table 的修改均對(duì) get 讀取可見(jiàn)；雖然 table 數(shù)組本身是增加了 volatile 屬性，但是“volatile 的數(shù)組只針對(duì)數(shù)組的引用具有volatile 的語(yǔ)義，而不是它的元素”。所以如果有其他線程對(duì)這個(gè)數(shù)組的元素進(jìn)行寫(xiě)操作，那么當(dāng)前線程來(lái)讀的時(shí)候不一定能讀到最新的值。出于性能考慮，Doug Lea 直接通過(guò) Unsafe 類(lèi)來(lái)對(duì) table 進(jìn)行操作。

static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
　　return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}

put 方法第二階段

　　在putVal方法執(zhí)行完成以后，會(huì)通過(guò)addCount來(lái)增加ConcurrentHashMap中的元素個(gè)數(shù)，并且還會(huì)可能觸發(fā)擴(kuò)容操作。這里會(huì)有兩個(gè)非常經(jīng)典的設(shè)計(jì)

高并發(fā)下的擴(kuò)容
如何保證 addCount 的數(shù)據(jù)安全性以及性能

........

　　//將當(dāng)前 ConcurrentHashMap 的元素?cái)?shù)量加 1，有可能觸發(fā) transfer 操作(擴(kuò)容)
　　addCount(1L, binCount);
　　return null;
}

addCount()：

　　在 putVal 最后調(diào)用 addCount 的時(shí)候，傳遞了兩個(gè)參數(shù)，分別是 1 和 binCount(鏈表長(zhǎng)度)，看看 addCount 方法里面做了什么操作。x 表示這次需要在表中增加的元素個(gè)數(shù)，check 參數(shù)表示是否需要進(jìn)行擴(kuò)容檢查，大于等于 0都需要進(jìn)行檢查

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
//    判斷 counterCells 是否為空，
//    1. 如果為空，就通過(guò) cas 操作嘗試修改 baseCount 變量，對(duì)這個(gè)變量進(jìn)行原子累加操
//    作(做這個(gè)操作的意義是：如果在沒(méi)有競(jìng)爭(zhēng)的情況下，仍然采用 baseCount 來(lái)記錄元素個(gè) 數(shù))
//    2. 如果 cas 失敗說(shuō)明存在競(jìng)爭(zhēng)，這個(gè)時(shí)候不能再采用 baseCount 來(lái)累加，而是通過(guò) CounterCell 來(lái)記錄
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
      CounterCell a; long v; int m;
      boolean uncontended = true;//是否沖突標(biāo)識(shí)，默認(rèn)為沒(méi)有沖突
//      這里有幾個(gè)判斷
//      1. 計(jì)數(shù)表為空則直接調(diào)用 fullAddCount
//      2. 從計(jì)數(shù)表中隨機(jī)取出一個(gè)數(shù)組的位置為空，直接調(diào)用 fullAddCount
//      3. 通過(guò) CAS 修改 CounterCell 隨機(jī)位置的值，如果修改失敗說(shuō)明出現(xiàn)并發(fā)情況（這里又
//      用到了一種巧妙的方法），調(diào)用 fullAndCount
//      Random 在線程并發(fā)的時(shí)候會(huì)有性能問(wèn)題以及可能會(huì)產(chǎn)生相同的隨機(jī)
//      數(shù) ,ThreadLocalRandom.getProbe 可以解決這個(gè)問(wèn)題，并且性能要比 Random 高
      if (as == null || (m = as.length - 1) < 0 ||
          (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
          !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
        fullAddCount(x, uncontended);//執(zhí)行 fullAddCount 方法
        return;
      }
      if (check <= 1)//鏈表長(zhǎng)度小于等于 1，不需要考慮擴(kuò)容
        return;
      s = sumCount();//統(tǒng)計(jì) ConcurrentHashMap 元素個(gè)數(shù)
    }
  .......
}

　　CounterCells 解釋?zhuān)?/p>

　　ConcurrentHashMap 是采用 CounterCell 數(shù)組來(lái)記錄元素個(gè)數(shù)的，像一般的集合記錄集合大小，直接定義一個(gè) size 的成員變量即可，當(dāng)出現(xiàn)改變的時(shí)候只要更新這個(gè)變量就行。為什么ConcurrentHashMap 要用這種形式來(lái)處理呢？問(wèn)題還是處在并發(fā)上，ConcurrentHashMap 是并發(fā)集合，如果用一個(gè)成員變量來(lái)統(tǒng)計(jì)元素個(gè)數(shù)的話，為了保證并發(fā)情況下共享變量的的安全性，勢(shì)必會(huì)需要通過(guò)加鎖或者自旋來(lái)實(shí)現(xiàn)，如果競(jìng)爭(zhēng)比較激烈的情況下，size 的設(shè)置上會(huì)出現(xiàn)比較大的沖突反而影響了性能，所以在ConcurrentHashMap 采用了分片的方法來(lái)記錄大小，具體什么意思，我們來(lái)分析下

private transient volatile int cellsBusy;// 標(biāo)識(shí)當(dāng)前 cell 數(shù)組是否在初始化或擴(kuò)容中的CAS 標(biāo)志位
/**
* Table of counter cells. When non-null, size is a power of 2.
*/
private transient volatile CounterCell[] counterCells;// counterCells 數(shù)組，總數(shù)值的分值分別存在每個(gè) cell 中

@sun.misc.Contended static final class CounterCell {
　　volatile long value;
　　CounterCell(long x) { value = x; }
}
//看到這段代碼就能夠明白了，CounterCell 數(shù)組的每個(gè)元素，都存儲(chǔ)一個(gè)元素個(gè)數(shù)，而實(shí)際我們調(diào)用size 方法就是通過(guò)這個(gè)循環(huán)累加來(lái)得到的
//又是一個(gè)設(shè)計(jì)精華，大家可以借鑒； 有了這個(gè)前提，再會(huì)過(guò)去看 addCount 這個(gè)方法，就容易理解一些了
final long sumCount() {
　　CounterCell[] as = counterCells; CounterCell a;
　　long sum = baseCount;
　　if (as != null) {
　　　　for (int i = 0; i < as.length; ++i) {
　　　　　　if ((a = as[i]) != null)
　　　　　　　　sum += a.value;
　　　　　　}
　　　　}
　　return sum;
}

fullAddCount()：

　　fullAddCount 主要是用來(lái)初始化 CounterCell，來(lái)記錄元素個(gè)數(shù)，里面包含擴(kuò)容，初始化等操作

private final void fullAddCount(long x, boolean wasUncontended) {
    int h;
    //獲取當(dāng)前線程的 probe 的值，如果值為 0，則初始化當(dāng)前線程的 probe 的值,probe 就是隨機(jī)數(shù)
    if ((h = ThreadLocalRandom.getProbe()) == 0) {
      ThreadLocalRandom.localInit();   // force initialization
      h = ThreadLocalRandom.getProbe();
      wasUncontended = true; // 由于重新生成了 probe，未沖突標(biāo)志位設(shè)置為 true
    }
    boolean collide = false;        // True if last slot nonempty
    for (;;) {//自旋
      CounterCell[] as; CounterCell a; int n; long v;
      //說(shuō)明 counterCells 已經(jīng)被初始化過(guò)了，我們先跳過(guò)這個(gè)代碼，先看初始化部分
      if ((as = counterCells) != null && (n = as.length) > 0) {
        // 通過(guò)該值與當(dāng)前線程 probe 求與，獲得cells 的下標(biāo)元素，和 hash 表獲取索引是一樣的
        if ((a = as[(n - 1) & h]) == null) {
          //cellsBusy=0 表示 counterCells 不在初始化或者擴(kuò)容狀態(tài)下
          if (cellsBusy == 0) {// Try to attach new Cell
            //構(gòu)造一個(gè) CounterCell 的值，傳入元素個(gè)數(shù)
            CounterCell r = new CounterCell(x); // Optimistic create
            //通過(guò) cas 設(shè)置 cellsBusy 標(biāo)識(shí)，防止其他線程來(lái)對(duì) counterCells 并發(fā)處理
            if (cellsBusy == 0 &&
                U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
              boolean created = false;
              try {        // Recheck under lock
                CounterCell[] rs; int m, j;
                //將初始化的 r 對(duì)象的元素個(gè)數(shù)放在對(duì)應(yīng)下標(biāo)的位置
                if ((rs = counterCells) != null &&
                    (m = rs.length) > 0 &&
                    rs[j = (m - 1) & h] == null) {
                  rs[j] = r;
                  created = true;
                }
              } finally {//恢復(fù)標(biāo)志位
                cellsBusy = 0;
              }
              if (created)//創(chuàng)建成功，退出循環(huán)
                break;
              //說(shuō)明指定 cells 下標(biāo)位置的數(shù)據(jù)不為空，則進(jìn)行下一次循環(huán)
              continue;      // Slot is now non-empty
            }
          }
          collide = false;
        }
        //說(shuō)明在 addCount 方法中 cas 失敗了，并且獲取 probe 的值不為空
        else if (!wasUncontended)    // CAS already known to fail
          // 設(shè)置為未沖突標(biāo)識(shí)，進(jìn)入下一次自旋
          wasUncontended = true;   // Continue after rehash
        // 由于指定下標(biāo)位置的 cell 值不為空，則直接通過(guò) cas 進(jìn)行原子累加，如果成功，則直接退出
        else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
          break;
        // 如果已經(jīng)有其他線程建立了新的 counterCells
        // 或者 CounterCells 大于 CPU 核心數(shù)（很巧妙，線程的并發(fā)數(shù)不會(huì)超過(guò) cpu 核心數(shù)）
        else if (counterCells != as || n >= NCPU)
          //設(shè)置當(dāng)前線程的循環(huán)失敗不進(jìn)行擴(kuò)容
          collide = false;      // At max size or stale
        else if (!collide)//恢復(fù) collide 狀態(tài)，標(biāo)識(shí)下次循環(huán)會(huì)進(jìn)行擴(kuò)容
          collide = true;
        //進(jìn)入這個(gè)步驟，說(shuō)明 CounterCell 數(shù)組容量不夠，線程競(jìng)爭(zhēng)較大，所以先設(shè)置一個(gè)標(biāo)識(shí)表示為正在擴(kuò)容
        else if (cellsBusy == 0 &&
            U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
          try {
            if (counterCells == as) {// Expand table unless stale
              // 擴(kuò)容一倍 2 變成 4 ，這個(gè)擴(kuò)容比較簡(jiǎn)單
              CounterCell[] rs = new CounterCell[n << 1];
              for (int i = 0; i < n; ++i)
                rs[i] = as[i];
              counterCells = rs;
            }
          } finally {
            cellsBusy = 0;//恢復(fù)標(biāo)識(shí)
          }
          collide = false;
          // 繼續(xù)下一次自旋
          continue;          // Retry with expanded table
        }//更新隨機(jī)數(shù)的值
        h = ThreadLocalRandom.advanceProbe(h);
      }
      //cellsBusy=0 表示沒(méi)有在做初始化，通過(guò) cas 更新 cellsbusy 的值標(biāo)注當(dāng)前線程正在做初始化操作
      else if (cellsBusy == 0 && counterCells == as &&
          U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
        boolean init = false;
        try {              // Initialize table
          if (counterCells == as) { //初始化容量為 2
            CounterCell[] rs = new CounterCell[2];
            //將 x 也就是元素的個(gè)數(shù)放在指定的數(shù)組下標(biāo)位置
            rs[h & 1] = new CounterCell(x);
            counterCells = rs;//賦值給 counterCells
            init = true;//設(shè)置初始化完成標(biāo)識(shí)
          }
        } finally {//恢復(fù)標(biāo)識(shí)
          cellsBusy = 0;
        }
        if (init)
          break;
      }
      //競(jìng)爭(zhēng)激烈，其它線程占據(jù) cell 數(shù)組，直接累加在 base 變量中
      else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
        break;             // Fall back on using base
    }
  }

　　CounterCells 初始化圖解

　　初始化長(zhǎng)度為 2 的數(shù)組，然后隨機(jī)得到指定的一個(gè)數(shù)組下標(biāo)，將需要新增的值加入到對(duì)應(yīng)下標(biāo)位置處

transfer() 擴(kuò)容階段:

　　回到addCount(long x, int check)。判斷是否需要擴(kuò)容，也就是當(dāng)更新后的鍵值對(duì)總數(shù) baseCount >= 閾值 sizeCtl 時(shí)，進(jìn)行rehash，這里面會(huì)有兩個(gè)邏輯。

如果當(dāng)前正在處于擴(kuò)容階段，則當(dāng)前線程會(huì)加入并且協(xié)助擴(kuò)容
如果當(dāng)前沒(méi)有在擴(kuò)容，則直接觸發(fā)擴(kuò)容操作

private final void addCount(long x, int check) {
　　　　　........if (check >= 0) {//如果 binCount>=0，標(biāo)識(shí)需要檢查擴(kuò)容
      Node<K,V>[] tab, nt; int n, sc;
      //s 標(biāo)識(shí)集合大小，如果集合大小大于或等于擴(kuò)容閾值（默認(rèn)值的 0.75）
      //并且 table 不為空并且 table 的長(zhǎng)度小于最大容量
      while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
          (n = tab.length) < MAXIMUM_CAPACITY) {
        //這里是生成一個(gè)唯一的擴(kuò)容戳，這個(gè)是干嘛用的呢？
        int rs = resizeStamp(n);
        if (sc < 0) {//sc<0，也就是 sizeCtl<0，說(shuō)明已經(jīng)有別的線程正在擴(kuò)容了
          // 這 5 個(gè)條件只要有一個(gè)條件為 true，說(shuō)明當(dāng)前線程不能幫助進(jìn)行此次的擴(kuò)容，直接跳出循環(huán)
          // sc >>> RESIZE_STAMP_SHIFT!=rs 表示比較高 RESIZE_STAMP_BITS 位生成戳和 rs 是否相等，相同
          // sc=rs+1 表示擴(kuò)容結(jié)束
          // sc==rs+MAX_RESIZERS 表示幫助線程線程已經(jīng)達(dá)到最大值了
          // nt=nextTable -> 表示擴(kuò)容已經(jīng)結(jié)束
          // transferIndex<=0 表示所有的 transfer 任務(wù)都被領(lǐng)取完了，
          // 沒(méi)有剩余的hash 桶來(lái)給自己自己好這個(gè)線程來(lái)做 transfer
          if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
              sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
              transferIndex <= 0)
            break;
          //當(dāng)前線程嘗試幫助此次擴(kuò)容，如果成功，則調(diào)用 transfer
          if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
            transfer(tab, nt);
        }
        // 如果當(dāng)前沒(méi)有在擴(kuò)容，那么 rs 肯定是一個(gè)正數(shù)，
        // 通過(guò) rs<<RESIZE_STAMP_SHIFT 將 sc 設(shè)置為一個(gè)負(fù)數(shù)，+2 表示有一個(gè)線程在執(zhí)行擴(kuò)容
        else if (U.compareAndSwapInt(this, SIZECTL, sc,
            (rs << RESIZE_STAMP_SHIFT) + 2))
          transfer(tab, null);
        s = sumCount();// 重新計(jì)數(shù)，判斷是否需要開(kāi)啟下一輪擴(kuò)容
      }
    }
  }

　　resizeStamp:

　　這塊邏輯要理解起來(lái)，也有一點(diǎn)復(fù)雜。resizeStamp 用來(lái)生成一個(gè)和擴(kuò)容有關(guān)的擴(kuò)容戳，具體有什么作用呢？我們基于它的實(shí)現(xiàn)來(lái)做一個(gè)分析

static final int resizeStamp(int n) {
　　return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}

　　Integer.numberOfLeadingZeros 這個(gè)方法是返回?zé)o符號(hào)整數(shù) n 最高位非 0 位前面的 0 的個(gè)數(shù)，比如 10 的二進(jìn)制是 0000 0000 0000 0000 0000 0000 0000 1010，那么這個(gè)方法返回的值就是 28

　　根據(jù) resizeStamp 的運(yùn)算邏輯，我們來(lái)推演一下，假如 n=16，那么 resizeStamp(16)=32796。轉(zhuǎn)化為二進(jìn)制是[0000 0000 0000 0000 1000 0000 0001 1100]

　　接著再來(lái)看,當(dāng)?shù)谝粋€(gè)線程嘗試進(jìn)行擴(kuò)容的時(shí)候，會(huì)執(zhí)行下面這段代碼：U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)

　　rs 左移 16 位，相當(dāng)于原本的二進(jìn)制低位變成了高位 1000 0000 0001 1100 0000 0000 00000000

　　然后再+2 =1000 0000 0001 1100 0000 0000 0000 0000+10=1000 0000 0001 1100 0000 00000000 0010

　　高 RESIZE_STAMP_BITS 16 位代表擴(kuò)容的標(biāo)記、低RESIZE_STAMP_BITS 16 位代表并行擴(kuò)容的線程數(shù)。這樣來(lái)存儲(chǔ)有什么好處呢？

首先在 CHM 中是支持并發(fā)擴(kuò)容的，也就是說(shuō)如果當(dāng)前的數(shù)組需要進(jìn)行擴(kuò)容操作，可以由多個(gè)線程來(lái)共同負(fù)責(zé)，這塊后續(xù)會(huì)單獨(dú)講
可以保證每次擴(kuò)容都生成唯一的生成戳，每次新的擴(kuò)容，都有一個(gè)不同的 n，這個(gè)生成戳就是根據(jù) n 來(lái)計(jì)算出來(lái)的一個(gè)數(shù)字，n 不同，這個(gè)數(shù)字也不同

　　第一個(gè)線程嘗試擴(kuò)容的時(shí)候，為什么是+2 ？

　　因?yàn)?1 表示初始化，2 表示一個(gè)線程在執(zhí)行擴(kuò)容，而且對(duì) sizeCtl 的操作都是基于位運(yùn)算的，所以不會(huì)關(guān)心它本身的數(shù)值是多少，只關(guān)心它在二進(jìn)制上的數(shù)值，而 sc + 1 會(huì)在低 16 位上加 1。

　　擴(kuò)容是 ConcurrentHashMap 的精華之一，擴(kuò)容操作的核心在于數(shù)據(jù)的轉(zhuǎn)移，在單線程環(huán)境下數(shù)據(jù)的轉(zhuǎn)移很簡(jiǎn)單，無(wú)非就是把舊數(shù)組中的數(shù)據(jù)遷移到新的數(shù)組。但是這在多線程環(huán)境下，在擴(kuò)容的時(shí)候其他線程也可能正在添加元素，這時(shí)又觸發(fā)了擴(kuò)容怎么辦？可能大家想到的第一個(gè)解決方案是加互斥鎖，把轉(zhuǎn)移過(guò)程鎖住，雖然是可行的解決方案，但是會(huì)帶來(lái)較大的性能開(kāi)銷(xiāo)。因?yàn)榛コ怄i會(huì)導(dǎo)致所有訪問(wèn)臨界區(qū)的線程陷入到阻塞狀態(tài)，持有鎖的線程耗時(shí)越長(zhǎng)，其他競(jìng)爭(zhēng)線程就會(huì)一直被阻塞，導(dǎo)致吞吐量較低。而且還可能導(dǎo)致死鎖。而 ConcurrentHashMap 并沒(méi)有直接加鎖，而是采用 CAS 實(shí)現(xiàn)無(wú)鎖的并發(fā)同步策略，最精華的部分是它可以利用多線程來(lái)進(jìn)行協(xié)同擴(kuò)容簡(jiǎn)單來(lái)說(shuō)，它把 Node 數(shù)組當(dāng)作多個(gè)線程之間共享的任務(wù)隊(duì)列，然后通過(guò)維護(hù)一個(gè)指針來(lái)劃分每個(gè)線程鎖負(fù)責(zé)的區(qū)間，每個(gè)線程通過(guò)區(qū)間逆向遍歷來(lái)實(shí)現(xiàn)擴(kuò)容，一個(gè)已經(jīng)遷移完的bucket會(huì)被替換為一個(gè)ForwardingNode節(jié)點(diǎn)，標(biāo)記當(dāng)前bucket已經(jīng)被其他線程遷移完了。接下來(lái)分析一下它的源碼實(shí)現(xiàn)

fwd:這個(gè)類(lèi)是個(gè)標(biāo)識(shí)類(lèi)，用于指向新表用的，其他線程遇到這個(gè)類(lèi)會(huì)主動(dòng)跳過(guò)這個(gè)類(lèi)，因?yàn)檫@個(gè)類(lèi)要么就是擴(kuò)容遷移正在進(jìn)行，要么就是已經(jīng)完成擴(kuò)容遷移，也就是這個(gè)類(lèi)要保證線程安全，再進(jìn)行操作。
advance:這個(gè)變量是用于提示代碼是否進(jìn)行推進(jìn)處理，也就是當(dāng)前桶處理完，處理下一個(gè)桶的標(biāo)識(shí)
finishing:這個(gè)變量用于提示擴(kuò)容是否結(jié)束用的

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    //將 (n>>>3 相當(dāng)于 n/8) 然后除以 CPU 核心數(shù)。如果得到的結(jié)果小于 16，那么就使用 16
    // 這里的目的是讓每個(gè) CPU 處理的桶一樣多，避免出現(xiàn)轉(zhuǎn)移任務(wù)不均勻的現(xiàn)象，如果桶較少的話，
    // 默認(rèn)一個(gè) CPU（一個(gè)線程）處理 16 個(gè)桶，也就是長(zhǎng)度為 16 的時(shí)候，擴(kuò)容的時(shí)候只會(huì)有一個(gè)線程來(lái)擴(kuò)容
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
      stride = MIN_TRANSFER_STRIDE; // subdivide range
    //nextTab 未初始化， nextTab 是用來(lái)擴(kuò)容的 node 數(shù)組
    if (nextTab == null) {      // initiating
      try {
        @SuppressWarnings("unchecked")
        //新建一個(gè) n<<1 原始 table 大小的 nextTab,也就是 32
        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
        nextTab = nt;//賦值給 nextTab
      } catch (Throwable ex) {   // try to cope with OOME
        //擴(kuò)容失敗，sizeCtl 使用 int 的最大值
        sizeCtl = Integer.MAX_VALUE;
        return;
      }
      nextTable = nextTab;//更新成員變量
      transferIndex = n;//更新轉(zhuǎn)移下標(biāo)，表示轉(zhuǎn)移時(shí)的下標(biāo)
    }//新的 tab 的長(zhǎng)度
    int nextn = nextTab.length;
    // 創(chuàng)建一個(gè) fwd 節(jié)點(diǎn)，表示一個(gè)正在被遷移的 Node，并且它的 hash 值為-1(MOVED)，也
    // 就是前面我們?cè)谥v putval 方法的時(shí)候，會(huì)有一個(gè)判斷 MOVED 的邏輯。它的作用是用來(lái)占位，表示
    // 原數(shù)組中位置 i 處的節(jié)點(diǎn)完成遷移以后，就會(huì)在 i 位置設(shè)置一個(gè) fwd 來(lái)告訴其他線程這個(gè)位置已經(jīng)處理過(guò)了，
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    // 首次推進(jìn)為 true，如果等于 true，說(shuō)明需要再次推進(jìn)一個(gè)下標(biāo)（i--），反之，
    // 如果是false，那么就不能推進(jìn)下標(biāo)，需要將當(dāng)前的下標(biāo)處理完畢才能繼續(xù)推進(jìn)
    boolean advance = true;
    //判斷是否已經(jīng)擴(kuò)容完成，完成就 return，退出循環(huán)
    boolean finishing = false; // to ensure sweep before committing nextTab
    // 通過(guò) for 自循環(huán)處理每個(gè)槽位中的鏈表元素，默認(rèn) advace 為真，通過(guò) CAS 設(shè)置
    // transferIndex 屬性值，并初始化 i 和 bound 值， i 指當(dāng)前處理的槽位序號(hào)， bound 指需要處理
    // 的槽位邊界，先處理槽位 15 的節(jié)點(diǎn)；
    for (int i = 0, bound = 0;;) {
      // 這個(gè)循環(huán)使用 CAS 不斷嘗試為當(dāng)前線程分配任務(wù)
      // 直到分配成功或任務(wù)隊(duì)列已經(jīng)被全部分配完畢
      // 如果當(dāng)前線程已經(jīng)被分配過(guò) bucket 區(qū)域
      // 那么會(huì)通過(guò)--i 指向下一個(gè)待處理 bucket 然后退出該循環(huán)
      Node<K,V> f; int fh;
      while (advance) {
        int nextIndex, nextBound;
        //--i 表示下一個(gè)待處理的 bucket，如果它>=bound,表示當(dāng)前線程已經(jīng)分配過(guò)bucket 區(qū)域
        if (--i >= bound || finishing)
          advance = false;
        else if ((nextIndex = transferIndex) <= 0) {//表示所有 bucket 已經(jīng)被分配完畢
          i = -1;
          advance = false;
        }
        //通過(guò) cas 來(lái)修改 TRANSFERINDEX,為當(dāng)前線程分配任務(wù)，處理的節(jié)點(diǎn)區(qū)間為(nextBound,nextIndex)->(0,15)
        else if (U.compareAndSwapInt
            (this, TRANSFERINDEX, nextIndex,
                nextBound = (nextIndex > stride ?
                    nextIndex - stride : 0))) {
          bound = nextBound;//0
          i = nextIndex - 1;//15
          advance = false;
        }
      }
      //i<0 說(shuō)明已經(jīng)遍歷完舊的數(shù)組，也就是當(dāng)前線程已經(jīng)處理完所有負(fù)責(zé)的 bucket
      if (i < 0 || i >= n || i + n >= nextn) {
        int sc;
        if (finishing) {//如果完成了擴(kuò)容
          nextTable = null;//刪除成員變量
          table = nextTab;//更新 table 數(shù)組
          sizeCtl = (n << 1) - (n >>> 1);//更新閾值(32*0.75=24)
          return;
        }
        // sizeCtl 在遷移前會(huì)設(shè)置為 (rs << RESIZE_STAMP_SHIFT) + 2
        // 然后，每增加一個(gè)線程參與遷移就會(huì)將 sizeCtl 加 1，
        // 這里使用 CAS 操作對(duì) sizeCtl 的低 16 位進(jìn)行減 1，代表做完了屬于自己的任務(wù)
        // if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
        // 第一個(gè)擴(kuò)容的線程，執(zhí)行 transfer 方法之前，會(huì)設(shè)置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)
        // 后續(xù)幫其擴(kuò)容的線程，執(zhí)行 transfer 方法之前，會(huì)設(shè)置 sizeCtl = sizeCtl+1
        // 每一個(gè)退出 transfer 的方法的線程，退出之前，會(huì)設(shè)置 sizeCtl = sizeCtl-1 那么最后一個(gè)線程退出時(shí)：必然有
        // sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT
        // 如果 sc - 2 不等于標(biāo)識(shí)符左移 16 位。如果他們相等了，說(shuō)明沒(méi)有線程在幫助他們擴(kuò)容了。也就是說(shuō)，擴(kuò)容結(jié)束了。
        if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
          if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
            return;
          // 如果相等，擴(kuò)容結(jié)束了，更新 finising 變量
          finishing = advance = true;
          // 再次循環(huán)檢查一下整張表
          i = n; // recheck before commit
        }
      }// 如果位置 i 處是空的，沒(méi)有任何節(jié)點(diǎn)，那么放入剛剛初始化的 ForwardingNode ”空節(jié)點(diǎn)“
      else if ((f = tabAt(tab, i)) == null)
        advance = casTabAt(tab, i, null, fwd);
      //表示該位置已經(jīng)完成了遷移，也就是如果線程 A 已經(jīng)處理過(guò)這個(gè)節(jié)點(diǎn)，那么線程 B 處理這個(gè)節(jié)點(diǎn)時(shí)，hash 值一定為 MOVED
      else if ((fh = f.hash) == MOVED)
        advance = true; // already processed
　　　　......
}

　　擴(kuò)容過(guò)程圖解：

　　ConcurrentHashMap 支持并發(fā)擴(kuò)容，實(shí)現(xiàn)方式是，把 Node 數(shù)組進(jìn)行拆分，讓每個(gè)線程處理自己的區(qū)域，假設(shè) table 數(shù)組總長(zhǎng)度是 64，默認(rèn)情況下，那么每個(gè)線程可以分到 16 個(gè) bucket。然后每個(gè)線程處理的范圍，按照倒序來(lái)做遷移通過(guò) for 自循環(huán)處理每個(gè)槽位中的鏈表元素，默認(rèn) advace 為真，通過(guò) CAS 設(shè)置 transferIndex屬性值，并初始化 i 和 bound 值，i 指當(dāng)前處理的槽位序號(hào)，bound 指需要處理的槽位邊界，先處理槽位 31 的節(jié)點(diǎn)；（bound,i） =(16,31) 從 31 的位置往前推動(dòng)。

　　假設(shè)這個(gè)時(shí)候 ThreadA 在進(jìn)行 transfer，那么邏輯圖表示如下

　　在當(dāng)前假設(shè)條件下，槽位 15 中沒(méi)有節(jié)點(diǎn)，則通過(guò) CAS 插入在第二步中初始化的ForwardingNode 節(jié)點(diǎn)，用于告訴其它線程該槽位已經(jīng)處理過(guò)了；

　　sizeCtl 擴(kuò)容退出機(jī)制：

　　在擴(kuò)容操作 transfer 的第 2414 行，代碼如下

if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
　　if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)

　　每存在一個(gè)線程執(zhí)行完擴(kuò)容操作，就通過(guò) cas 執(zhí)行 sc-1。接著判斷(sc-2) !=resizeStamp(n) << RESIZE_STAMP_SHIFT ; 如果相等，表示當(dāng)前為整個(gè)擴(kuò)容操作的最后一個(gè)線程，那么意味著整個(gè)擴(kuò)容操作就結(jié)束了；如果不想等，說(shuō)明還得繼續(xù)這么做的目的，一方面是防止不同擴(kuò)容之間出現(xiàn)相同的 sizeCtl，另外一方面，還可以避免sizeCtl 的 ABA 問(wèn)題導(dǎo)致的擴(kuò)容重疊的情況。

數(shù)據(jù)遷移階段的實(shí)現(xiàn)分析

　　通過(guò)分配好遷移的區(qū)間之后，開(kāi)始對(duì)數(shù)據(jù)進(jìn)行遷移。在看這段代碼之前，先來(lái)了解一下原理

高低位原理分析

　　ConcurrentHashMap 在做鏈表遷移時(shí)，會(huì)用高低位來(lái)實(shí)現(xiàn)，這里有兩個(gè)問(wèn)題要分析一下

　　如何實(shí)現(xiàn)高低位鏈表的區(qū)分，假如我們有這樣一個(gè)隊(duì)列：

　　第 14 個(gè)槽位插入新節(jié)點(diǎn)之后，鏈表元素個(gè)數(shù)已經(jīng)達(dá)到了 8，且數(shù)組長(zhǎng)度為 16，優(yōu)先通過(guò)擴(kuò)容來(lái)緩解鏈表過(guò)長(zhǎng)的問(wèn)題，擴(kuò)容這塊的圖解稍后再分析，先分析高低位擴(kuò)容的原理。假如當(dāng)前線程正在處理槽位為 14 的節(jié)點(diǎn)，它是一個(gè)鏈表結(jié)構(gòu)，在代碼中，首先定義兩個(gè)變量節(jié)點(diǎn) ln 和 hn，實(shí)際就是 lowNode 和 HighNode，分別保存 hash 值的第 x 位為 0 和不等于0 的節(jié)點(diǎn)。通過(guò) fn&n 可以把這個(gè)鏈表中的元素分為兩類(lèi)，A 類(lèi)是 hash 值的第 X 位為 0，B 類(lèi)是 hash 值的第 x 位為不等于 0（至于為什么要這么區(qū)分，稍后分析），并且通過(guò) lastRun 記錄最后要處理的節(jié)點(diǎn)。最終要達(dá)到的目的是，A 類(lèi)的鏈表保持位置不動(dòng)，B 類(lèi)的鏈表為 14+16(擴(kuò)容增加的長(zhǎng)度)=30我們把 14 槽位的鏈表單獨(dú)伶出來(lái)，我們用藍(lán)色表示 fn&n=0 的節(jié)點(diǎn)，假如鏈表的分類(lèi)是這樣：

for (Node<K,V> p = f.next; p != null; p = p.next) {
  int b = p.hash & n;
  if (b != runBit) {
    runBit = b;
    lastRun = p;
  }
}

　　通過(guò)上面這段代碼遍歷，會(huì)記錄 runBit 以及 lastRun，按照上面這個(gè)結(jié)構(gòu)，那么 runBit 應(yīng)該是藍(lán)色節(jié)點(diǎn)，lastRun 應(yīng)該是第 6 個(gè)節(jié)點(diǎn)接著，再通過(guò)這段代碼進(jìn)行遍歷，生成 ln 鏈以及 hn 鏈

　　接著，通過(guò) CAS 操作，把 hn 鏈放在 i+n 也就是 14+16 的位置，ln 鏈保持原來(lái)的位置不動(dòng)。并且設(shè)置當(dāng)前節(jié)點(diǎn)為 fwd，表示已經(jīng)被當(dāng)前線程遷移完了

setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);

　　遷移完成以后的數(shù)據(jù)分布如下

　　為什么要做高低位的劃分？

　　要想了解這么設(shè)計(jì)的目的，我們需要從 ConcurrentHashMap 的根據(jù)下標(biāo)獲取對(duì)象的算法來(lái)看，在 putVal 方法中：(f = tabAt(tab, i = (n - 1) & hash)) == null。通過(guò)(n-1) & hash 來(lái)獲得在 table 中的數(shù)組下標(biāo)來(lái)獲取節(jié)點(diǎn)數(shù)據(jù)，【&運(yùn)算是二進(jìn)制運(yùn)算符，1& 1=1，其他都為 0】

　　假設(shè)我們的 table 長(zhǎng)度是 16，二進(jìn)制是【0001 0000】，減一以后的二進(jìn)制是【0000 1111】。假如某個(gè) key 的 hash 值是 20，對(duì)應(yīng)的二進(jìn)制是【0001 0100】，仍然按照(n-1) & hash。算法，分別在 16 長(zhǎng)度和 32 長(zhǎng)度下的計(jì)算結(jié)果

　　16 ： 0000 1111 & 0001 0100=0000 0100 。32 ： 0001 1111 & 0001 0100 =0001 0100 。從結(jié)果來(lái)看，同樣一個(gè) hash 值，在擴(kuò)容前和擴(kuò)容之后，得到的下標(biāo)位置是不一樣的，這種情況當(dāng)然是不允許出現(xiàn)的，所以在擴(kuò)容的時(shí)候就需要考慮，而使用高低位的遷移方式，就是解決這個(gè)問(wèn)題.

　　大家可以看到，16 位的結(jié)果到 32 位的結(jié)果，正好增加了 16.

　　比如 20 & 15=4 、20 & 31=20 ； 4-20 =16

　　比如 60 & 15=12 、60 & 31=28； 12-28=16

　　所以對(duì)于高位，直接增加擴(kuò)容的長(zhǎng)度，當(dāng)下次 hash 獲取數(shù)組位置的時(shí)候，可以直接定位到對(duì)應(yīng)的位置。這個(gè)地方又是一個(gè)很巧妙的設(shè)計(jì)，直接通過(guò)高低位分類(lèi)以后，就使得不需要在每次擴(kuò)容的時(shí)候來(lái)重新計(jì)算 hash，極大提升了效率。

　　接下來(lái)回到鏈表的擴(kuò)容代碼：

else {
        synchronized (f) {//對(duì)數(shù)組該節(jié)點(diǎn)位置加鎖，開(kāi)始處理數(shù)組該位置的遷移工作
          if (tabAt(tab, i) == f) {//再做一次校驗(yàn)
            //ln 表示低位， hn 表示高位;接下來(lái)這段代碼的作用是把鏈表拆分成兩部分，0 在低位，1 在高位
            Node<K,V> ln, hn;
            if (fh >= 0) {
              //下面部分代碼原理在上面分析了
              int runBit = fh & n;
              Node<K,V> lastRun = f;
              //遍歷當(dāng)前 bucket 的鏈表，目的是盡量重用 Node 鏈表尾部的一部分
              for (Node<K,V> p = f.next; p != null; p = p.next) {
                int b = p.hash & n;
                if (b != runBit) {
                  runBit = b;
                  lastRun = p;
                }
              }//如果最后更新的 runBit 是 0，設(shè)置低位節(jié)點(diǎn)
              if (runBit == 0) {
                ln = lastRun;
                hn = null;
              }//否則，設(shè)置高位節(jié)點(diǎn)
              else {
                hn = lastRun;
                ln = null;
              }//構(gòu)造高位以及低位的鏈表
              for (Node<K,V> p = f; p != lastRun; p = p.next) {
                int ph = p.hash; K pk = p.key; V pv = p.val;
                if ((ph & n) == 0)
                  ln = new Node<K,V>(ph, pk, pv, ln);
                else
                  hn = new Node<K,V>(ph, pk, pv, hn);
              }
              //將低位的鏈表放在 i 位置也就是不動(dòng)
              setTabAt(nextTab, i, ln);
              //將高位鏈表放在 i+n 位置
              setTabAt(nextTab, i + n, hn);
              // 把舊 table 的 hash 桶中放置轉(zhuǎn)發(fā)節(jié)點(diǎn)，表明此 hash 桶已經(jīng)被處理
              setTabAt(tab, i, fwd);
              advance = true;
            }//f 是紅黑樹(shù)的根節(jié)點(diǎn)
            else if (f instanceof TreeBin) {
              TreeBin<K,V> t = (TreeBin<K,V>)f;
              //創(chuàng)建低位的樹(shù)節(jié)點(diǎn)
              TreeNode<K,V> lo = null, loTail = null;
              //創(chuàng)建高位的樹(shù)節(jié)點(diǎn)
              TreeNode<K,V> hi = null, hiTail = null;
              int lc = 0, hc = 0;
              for (Node<K,V> e = t.first; e != null; e = e.next) {
                int h = e.hash;
                TreeNode<K,V> p = new TreeNode<K,V>
                    (h, e.key, e.val, null, null);
                if ((h & n) == 0) {
                  if ((p.prev = loTail) == null)
                    lo = p;
                  else
                    loTail.next = p;
                  loTail = p;
                  ++lc;
                }
                else {
                  if ((p.prev = hiTail) == null)
                    hi = p;
                  else
                    hiTail.next = p;
                  hiTail = p;
                  ++hc;
                }
              }
              //判斷是否長(zhǎng)度< 6 轉(zhuǎn)成鏈表
              ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                  (hc != 0) ? new TreeBin<K,V>(lo) : t;
              hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                  (lc != 0) ? new TreeBin<K,V>(hi) : t;
              setTabAt(nextTab, i, ln);
              setTabAt(nextTab, i + n, hn);
              setTabAt(tab, i, fwd);
              advance = true;
            }
          }
        }
      }

　　擴(kuò)容結(jié)束以后的退出機(jī)制：

　　如果線程擴(kuò)容結(jié)束，那么需要退出，就會(huì)執(zhí)行 transfer 方法的如下代碼

//i<0 說(shuō)明已經(jīng)遍歷完舊的數(shù)組，也就是當(dāng)前線程已經(jīng)處理完所有負(fù)責(zé)的 bucket
      if (i < 0 || i >= n || i + n >= nextn) {
        int sc;
        if (finishing) {//如果完成了擴(kuò)容
          nextTable = null;//刪除成員變量
          table = nextTab;//更新 table 數(shù)組
          sizeCtl = (n << 1) - (n >>> 1);//更新閾值(32*0.75=24)
          return;
        }
        // sizeCtl 在遷移前會(huì)設(shè)置為 (rs << RESIZE_STAMP_SHIFT) + 2
        // 然后，每增加一個(gè)線程參與遷移就會(huì)將 sizeCtl 加 1，
        // 這里使用 CAS 操作對(duì) sizeCtl 的低 16 位進(jìn)行減 1，代表做完了屬于自己的任務(wù)
        if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
          // 第一個(gè)擴(kuò)容的線程，執(zhí)行 transfer 方法之前，會(huì)設(shè)置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)
          // 后續(xù)幫其擴(kuò)容的線程，執(zhí)行 transfer 方法之前，會(huì)設(shè)置 sizeCtl = sizeCtl+1
          // 每一個(gè)退出 transfer 的方法的線程，退出之前，會(huì)設(shè)置 sizeCtl = sizeCtl-1 那么最后一個(gè)線程退出時(shí)：必然有
          // sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)，即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT
          // 如果 sc - 2 不等于標(biāo)識(shí)符左移 16 位。如果他們相等了，說(shuō)明沒(méi)有線程在幫助他們擴(kuò)容了。也就是說(shuō)，擴(kuò)容結(jié)束了。
          if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
            return;
          // 如果相等，擴(kuò)容結(jié)束了，更新 finising 變量
          finishing = advance = true;
          // 再次循環(huán)檢查一下整張表
          i = n; // recheck before commit
        }
      }

put 方法第三階段:

　　如果對(duì)應(yīng)的節(jié)點(diǎn)存在，判斷這個(gè)節(jié)點(diǎn)的 hash 是不是等于 MOVED(-1)，說(shuō)明當(dāng)前節(jié)點(diǎn)是ForwardingNode 節(jié)點(diǎn)，意味著有其他線程正在進(jìn)行擴(kuò)容，那么當(dāng)前現(xiàn)在直接幫助它進(jìn)行擴(kuò)容，因此調(diào)用 helpTransfer方法

else if ((fh = f.hash) == MOVED)
  tab = helpTransfer(tab, f);

　　Helps transfer if a resize is in progress.

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
    Node<K,V>[] nextTab; int sc;
    // 判斷此時(shí)是否仍然在執(zhí)行擴(kuò)容,nextTab=null 的時(shí)候說(shuō)明擴(kuò)容已經(jīng)結(jié)束了
    if (tab != null && (f instanceof ForwardingNode) &&
        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
      int rs = resizeStamp(tab.length);//生成擴(kuò)容戳
      while (nextTab == nextTable && table == tab &&
          (sc = sizeCtl) < 0) {//說(shuō)明擴(kuò)容還未完成的情況下不斷循環(huán)來(lái)嘗試將當(dāng)前線程加入到擴(kuò)容操作中
        // 下面部分的整個(gè)代碼表示擴(kuò)容結(jié)束，直接退出循環(huán)
        // transferIndex<=0 表示所有的 Node 都已經(jīng)分配了線程
        // sc=rs+MAX_RESIZERS 表示擴(kuò)容線程數(shù)達(dá)到最大擴(kuò)容線程數(shù)
        // sc >>> RESIZE_STAMP_SHIFT !=rs， 如果在同一輪擴(kuò)容中，
        // 那么 sc 無(wú)符號(hào)右移比較高位和 rs 的值，那么應(yīng)該是相等的。如果不相等，說(shuō)明擴(kuò)容結(jié)束了
        //sc==rs+1 表示擴(kuò)容結(jié)束
        if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
            sc == rs + MAX_RESIZERS || transferIndex <= 0)
          break;
        //在低 16 位上增加擴(kuò)容線程數(shù)
        if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
          transfer(tab, nextTab);//幫助擴(kuò)容
          break;
        }
      }
      return nextTab;//返回新的數(shù)組
    }
    return table;
  }

put 方法第四階段：

　　這個(gè)方法的主要作用是，如果被添加的節(jié)點(diǎn)的位置已經(jīng)存在節(jié)點(diǎn)的時(shí)候，需要以鏈表的方式加入到節(jié)點(diǎn)中。如果當(dāng)前節(jié)點(diǎn)已經(jīng)是一顆紅黑樹(shù)，那么就會(huì)按照紅黑樹(shù)的規(guī)則將當(dāng)前節(jié)點(diǎn)加入到紅黑樹(shù)中。

　　如果是鏈表，添加完以后判斷鏈表的長(zhǎng)度是否已經(jīng)達(dá)到臨界值 8. 如果達(dá)到了臨界值，這個(gè)時(shí)候會(huì)根據(jù)當(dāng)前數(shù)組的長(zhǎng)度來(lái)決定是擴(kuò)容還是將鏈表轉(zhuǎn)化為紅黑樹(shù)。也就是說(shuō)如果當(dāng)前數(shù)組的長(zhǎng)度小于 64，就會(huì)先擴(kuò)容。否則，會(huì)把當(dāng)前鏈表轉(zhuǎn)化為紅黑樹(shù)

else {//進(jìn)入到這個(gè)分支，說(shuō)明 f 是當(dāng)前 nodes 數(shù)組對(duì)應(yīng)位置節(jié)點(diǎn)的頭節(jié)點(diǎn)，并且不為空
        V oldVal = null;
        synchronized (f) { //給對(duì)應(yīng)的頭結(jié)點(diǎn)加鎖
          if (tabAt(tab, i) == f) {//再次判斷對(duì)應(yīng)下標(biāo)位置是否為 f 節(jié)點(diǎn)
            if (fh >= 0) {//頭結(jié)點(diǎn)的 hash 值大于 0，說(shuō)明是鏈表
              binCount = 1;//用來(lái)記錄鏈表的長(zhǎng)度
              for (Node<K,V> e = f;; ++binCount) {//遍歷鏈表
                K ek;
                //如果發(fā)現(xiàn)相同的 key，則判斷是否需要進(jìn)行值的覆蓋
                if (e.hash == hash &&
                    ((ek = e.key) == key ||
                        (ek != null && key.equals(ek)))) {
                  oldVal = e.val;
                  //默認(rèn)情況下，直接覆蓋舊的值
                  if (!onlyIfAbsent)
                    e.val = value;
                  break;
                }
                //一直遍歷到鏈表的最末端，直接把新的值加入到鏈表的最后面
                Node<K,V> pred = e;
                if ((e = e.next) == null) {
                  pred.next = new Node<K,V>(hash, key,
                      value, null);
                  break;
                }
              }
            }
            //如果當(dāng)前的 f 節(jié)點(diǎn)是一顆紅黑樹(shù)
            else if (f instanceof TreeBin) {
              Node<K,V> p;
              binCount = 2;
              //則調(diào)用紅黑樹(shù)的插入方法插入新的值
              if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                  value)) != null) {
                oldVal = p.val;
                //同樣，如果值已經(jīng)存在，則直接替換
                if (!onlyIfAbsent)
                  p.val = value;
              }
            }
          }
        }//說(shuō)明上面在做鏈表操作
        if (binCount != 0) {
          if (binCount >= TREEIFY_THRESHOLD)
            //如果鏈表長(zhǎng)度已經(jīng)達(dá)到臨界值 8 就需要把鏈表轉(zhuǎn)換為樹(shù)結(jié)構(gòu)
            treeifyBin(tab, i);
          if (oldVal != null)
            //如果 val 是被替換的，則返回替換之前的值
            return oldVal;
          break;
        }
      }

treeifyBin():

　　在 putVal 的最后部分，有一個(gè)判斷，如果鏈表長(zhǎng)度大于 8，那么就會(huì)觸發(fā)擴(kuò)容或者紅黑樹(shù)的轉(zhuǎn)化操作。

private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    if (tab != null) {//tab 的長(zhǎng)度是不是小于 64，如果是，則執(zhí)行擴(kuò)容
      if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
        tryPresize(n << 1);
      //否則，將當(dāng)前鏈表轉(zhuǎn)化為紅黑樹(shù)結(jié)構(gòu)存儲(chǔ)
      else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
        synchronized (b) {// 將鏈表轉(zhuǎn)換成紅黑樹(shù)
          if (tabAt(tab, index) == b) {
            TreeNode<K,V> hd = null, tl = null;
            for (Node<K,V> e = b; e != null; e = e.next) {
              TreeNode<K,V> p =
                  new TreeNode<K,V>(e.hash, e.key, e.val,
                      null, null);
              if ((p.prev = tl) == null)
                hd = p;
              else
                tl.next = p;
              tl = p;
            }
            setTabAt(tab, index, new TreeBin<K,V>(hd));
          }
        }
      }
    }
  }

tryPresize():

　　tryPresize 里面部分代碼和 addCount 的部分代碼類(lèi)似，看起來(lái)會(huì)稍微簡(jiǎn)單一些

private final void tryPresize(int size) {
    //對(duì) size 進(jìn)行修復(fù),主要目的是防止傳入的值不是一個(gè) 2 次冪的整數(shù)，
    // 然后通過(guò)tableSizeFor 來(lái)將入?yún)⑥D(zhuǎn)化為離該整數(shù)最近的 2 次冪
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
        tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    while ((sc = sizeCtl) >= 0) {
      Node<K,V>[] tab = table; int n;
      //下面這段代碼和 initTable 是一樣的，如果 table 沒(méi)有初始化，則開(kāi)始初始化
      if (tab == null || (n = tab.length) == 0) {
        n = (sc > c) ? sc : c;
        if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
          try {
            if (table == tab) {
              @SuppressWarnings("unchecked")
              Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
              table = nt;
              sc = n - (n >>> 2);
            }
          } finally {
            sizeCtl = sc;
          }
        }
      }
      else if (c <= sc || n >= MAXIMUM_CAPACITY)
        break;
      else if (tab == table) {//這段代碼和 addCount 后部分代碼是一樣的，做輔助擴(kuò)容操作
        int rs = resizeStamp(n);
        if (sc < 0) {
          Node<K,V>[] nt;
          if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
              sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
              transferIndex <= 0)
            break;
          if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
            transfer(tab, nt);
        }
        else if (U.compareAndSwapInt(this, SIZECTL, sc,
            (rs << RESIZE_STAMP_SHIFT) + 2))
          transfer(tab, null);
      }
    }
  }

　　就這樣基于CHM的put()方法我們基本上就分析完了，多過(guò)幾遍源碼理解應(yīng)該不難。對(duì)于有些計(jì)算模糊的，可以通過(guò)寫(xiě)測(cè)試類(lèi)進(jìn)行佐證。

到此這篇關(guān)于JDK1.8中的ConcurrentHashMap使用及場(chǎng)景分析的文章就介紹到這了,更多相關(guān)JDK1.8之ConcurrentHashMap內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: