java并發(fā)包ConcurrentHashMap源碼的示例分析

這篇文章主要介紹java并發(fā)包ConcurrentHashMap源碼的示例分析，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

JDK1.7的實現(xiàn)

整個 ConcurrentHashMap 由一個個 Segment 組成，Segment 代表”部分“或”一段“的意思，所以很多地方都會將其描述為分段鎖。注意，行文中，我很多地方用了“槽”來代表一個 segment。

簡單理解就是，ConcurrentHashMap 是一個 Segment 數(shù)組，Segment 通過繼承 ReentrantLock 來進行加鎖，所以每次需要加鎖的操作鎖住的是一個 segment，這樣只要保證每個 Segment 是線程安全的，也就實現(xiàn)了全局的線程安全。

concurrencyLevel：并行級別、并發(fā)數(shù)、Segment 數(shù)。默認(rèn)是 16，也就是說 ConcurrentHashMap 有 16 個 Segments，所以理論上，這個時候，最多可以同時支持 16 個線程并發(fā)寫，只要它們的操作分別分布在不同的 Segment 上。這個值可以在初始化的時候設(shè)置為其他值，但是一旦初始化以后，它是不可以擴容的。

再具體到每個 Segment 內(nèi)部，其實每個 Segment 很像之前介紹的 HashMap，不過它要保證線程安全，所以處理起來要麻煩些。

初始化
initialCapacity：初始容量，這個值指的是整個 ConcurrentHashMap 的初始容量，實際操作的時候需要平均分給每個 Segment。

loadFactor：負載因子，之前我們說了，Segment 數(shù)組不可以擴容，所以這個負載因子是給每個 Segment 內(nèi)部使用的。

public ConcurrentHashMap(int initialCapacity,
             float loadFactor, int concurrencyLevel) {
  if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
    throw new IllegalArgumentException();
  if (concurrencyLevel > MAX_SEGMENTS)
    concurrencyLevel = MAX_SEGMENTS;
  // Find power-of-two sizes best matching arguments
  int sshift = 0;
  int ssize = 1;
  // 計算并行級別 ssize，因為要保持并行級別是 2 的 n 次方
  while (ssize < concurrencyLevel) {
    ++sshift;
    ssize <<= 1;
  }
  // 我們這里先不要那么燒腦，用默認(rèn)值，concurrencyLevel 為 16，sshift 為 4
  // 那么計算出 segmentShift 為 28，segmentMask 為 15，后面會用到這兩個值
  this.segmentShift = 32 - sshift;
  this.segmentMask = ssize - 1;
  if (initialCapacity > MAXIMUM_CAPACITY)
    initialCapacity = MAXIMUM_CAPACITY;
  // initialCapacity 是設(shè)置整個 map 初始的大小，
  // 這里根據(jù) initialCapacity 計算 Segment 數(shù)組中每個位置可以分到的大小
  // 如 initialCapacity 為 64，那么每個 Segment 或稱之為"槽"可以分到 4 個
  int c = initialCapacity / ssize;
  if (c * ssize < initialCapacity)
    ++c;
  // 默認(rèn) MIN_SEGMENT_TABLE_CAPACITY 是 2，這個值也是有講究的，因為這樣的話，對于具體的槽上，
  // 插入一個元素不至于擴容，插入第二個的時候才會擴容
  int cap = MIN_SEGMENT_TABLE_CAPACITY; 
  while (cap < c)
    cap <<= 1;
  // 創(chuàng)建 Segment 數(shù)組，
  // 并創(chuàng)建數(shù)組的第一個元素 segment[0]
  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];
  // 往數(shù)組寫入 segment[0]
  UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
  this.segments = ss;
}

初始化完成，我們得到了一個 Segment 數(shù)組。

我們就當(dāng)是用 new ConcurrentHashMap() 無參構(gòu)造函數(shù)進行初始化的，那么初始化完成后：

• Segment 數(shù)組長度為 16，不可以擴容

• Segment[i] 的默認(rèn)大小為 2，負載因子是 0.75，得出初始閾值為 1.5，也就是以后插入第一個元素不會觸發(fā)擴容，插入第二個會進行第一次擴容

• 這里初始化了 segment[0]，其他位置還是 null，至于為什么要初始化 segment[0]，后面的代碼會介紹

• 當(dāng)前 segmentShift 的值為 32 - 4 = 28，segmentMask 為 16 - 1 = 15，姑且把它們簡單翻譯為移位數(shù)和掩碼，這兩個值馬上就會用到

Segment

static class Segment<K,V> extends ReentrantLock implements Serializable {
  transient volatile HashEntry<K,V>[] table;
  transient int count;
  transient int modCount;
  
}

從上Segment的繼承體系可以看出，Segment實現(xiàn)了ReentrantLock,也就帶有鎖的功能,table使用volatile修飾，保證了內(nèi)存可見性。

put 過程分析

我們先看 put 的主流程，對于其中的一些關(guān)鍵細節(jié)操作，后面會進行詳細介紹。

public V put(K key, V value) {
  Segment<K,V> s;
  if (value == null)
    throw new NullPointerException();
  // 1. 計算 key 的 hash 值
  int hash = hash(key);
  // 2. 根據(jù) hash 值找到 Segment 數(shù)組中的位置 j
  //  hash 是 32 位，無符號右移 segmentShift(28) 位，剩下高 4 位，
  //  然后和 segmentMask(15) 做一次與操作，也就是說 j 是 hash 值的高 4 位，也就是槽的數(shù)組下標(biāo)
  int j = (hash >>> segmentShift) & segmentMask;
  // 剛剛說了，初始化的時候初始化了 segment[0]，但是其他位置還是 null，
  // ensureSegment(j) 對 segment[j] 進行初始化
  if ((s = (Segment<K,V>)UNSAFE.getObject     // nonvolatile; recheck
     (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
    s = ensureSegment(j);
  // 3. 插入新值到 槽 s 中
  return s.put(key, hash, value, false);
}

初始化槽: ensureSegment

ConcurrentHashMap 初始化的時候會初始化第一個槽 segment[0]，對于其他槽來說，在插入第一個值的時候進行初始化。

這里需要考慮并發(fā)，因為很可能會有多個線程同時進來初始化同一個槽 segment[k]，不過只要有一個成功了就可以。

private Segment<K,V> ensureSegment(int k) {
  final Segment<K,V>[] ss = this.segments;
  long u = (k << SSHIFT) + SBASE; // raw offset
  Segment<K,V> seg;
  if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
    // 這里看到為什么之前要初始化 segment[0] 了，
    // 使用當(dāng)前 segment[0] 處的數(shù)組長度和負載因子來初始化 segment[k]
    // 為什么要用“當(dāng)前”，因為 segment[0] 可能早就擴容過了
    Segment<K,V> proto = ss[0];
    int cap = proto.table.length;
    float lf = proto.loadFactor;
    int threshold = (int)(cap * lf);
    // 初始化 segment[k] 內(nèi)部的數(shù)組
    HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
    if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
      == null) { // 再次檢查一遍該槽是否被其他線程初始化了。
      Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
      // 使用 while 循環(huán)，內(nèi)部用 CAS，當(dāng)前線程成功設(shè)值或其他線程成功設(shè)值后，退出,如果其他線程成功設(shè)置后，這里獲取到直接返回
      while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
          == null) {
        if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
          break;
      }
    }
  }
  return seg;
}

總的來說，ensureSegment(int k) 比較簡單，對于并發(fā)操作使用 CAS 進行控制。

第一層很簡單，根據(jù) hash 值很快就能找到相應(yīng)的 Segment，之后就是 Segment 內(nèi)部的 put 操作了。

Segment 內(nèi)部是由 數(shù)組+鏈表 組成的

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
  // 在往該 segment 寫入前，需要先獲取該 segment 的獨占鎖
  //  先看主流程，后面還會具體介紹這部分內(nèi)容
  HashEntry<K,V> node = tryLock() ? null :
    scanAndLockForPut(key, hash, value);
  V oldValue;
  try {
    // 這個是 segment 內(nèi)部的數(shù)組
    HashEntry<K,V>[] tab = table;
    // 再利用 hash 值，求應(yīng)該放置的數(shù)組下標(biāo)
    int index = (tab.length - 1) & hash;
    // first 是數(shù)組該位置處的鏈表的表頭
    HashEntry<K,V> first = entryAt(tab, index);

    // 下面這串 for 循環(huán)雖然很長，不過也很好理解，想想該位置沒有任何元素和已經(jīng)存在一個鏈表這兩種情況
    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;
        }
        // 繼續(xù)順著鏈表走
        e = e.next;
      }
      else {
        // node 到底是不是 null，這個要看獲取鎖的過程，不過和這里都沒有關(guān)系。
        // 如果不為 null，那就直接將它設(shè)置為鏈表表頭；如果是null，初始化并設(shè)置為鏈表表頭。
        if (node != null)
          node.setNext(first);
        else
          node = new HashEntry<K,V>(hash, key, value, first);

        int c = count + 1;
        // 如果超過了該 segment 的閾值，這個 segment 需要擴容
        if (c > threshold && tab.length < MAXIMUM_CAPACITY)
          rehash(node); // 擴容后面也會具體分析
        else
          // 沒有達到閾值，將 node 放到數(shù)組 tab 的 index 位置，
          // 其實就是將新的節(jié)點設(shè)置成原鏈表的表頭
          setEntryAt(tab, index, node);
        ++modCount;
        count = c;
        oldValue = null;
        break;
      }
    }
  } finally {
    // 解鎖
    unlock();
  }
  return oldValue;
}

整體流程還是比較簡單的，由于有獨占鎖的保護，所以 segment 內(nèi)部的操作并不復(fù)雜。至于這里面的并發(fā)問題，我們稍后再進行介紹。

到這里 put 操作就結(jié)束了，接下來，我們說一說其中幾步關(guān)鍵的操作。

獲取寫入鎖: scanAndLockForPut

前面我們看到，在往某個 segment 中 put 的時候，首先會調(diào)用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是說先進行一次 tryLock() 快速獲取該 segment 的獨占鎖，如果失敗，那么進入到 scanAndLockForPut 這個方法來獲取鎖。

下面我們來具體分析這個方法中是怎么控制加鎖的。

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
  HashEntry<K,V> first = entryForHash(this, hash);
  HashEntry<K,V> e = first;
  HashEntry<K,V> node = null;
  int retries = -1; // negative while locating node
  // 循環(huán)獲取鎖
  while (!tryLock()) {
    HashEntry<K,V> f; // to recheck first below
    if (retries < 0) {
      if (e == null) {
        if (node == null) // speculatively create node
          // 進到這里說明數(shù)組該位置的鏈表是空的，沒有任何元素
          // 當(dāng)然，進到這里的另一個原因是 tryLock() 失敗，所以該槽存在并發(fā)，不一定是該位置
          node = new HashEntry<K,V>(hash, key, value, null);
        retries = 0;
      }
      else if (key.equals(e.key))
        retries = 0;
      else
        // 順著鏈表往下走
        e = e.next;
    }
    // 重試次數(shù)如果超過 MAX_SCAN_RETRIES（單核1多核64），那么不搶了，進入到阻塞隊列等待鎖
    //  lock() 是阻塞方法，直到獲取鎖后返回
    else if (++retries > MAX_SCAN_RETRIES) {
      lock();
      break;
    }
    else if ((retries & 1) == 0 &&
         // 這個時候是有大問題了，那就是有新的元素進到了鏈表，成為了新的表頭
         //   所以這邊的策略是，相當(dāng)于重新走一遍這個 scanAndLockForPut 方法
         (f = entryForHash(this, hash)) != first) {
      e = first = f; // re-traverse if entry changed
      retries = -1;
    }
  }
  return node;
}

這個方法有兩個出口，一個是 tryLock() 成功了，循環(huán)終止，另一個就是重試次數(shù)超過了 MAX_SCAN_RETRIES，進到 lock() 方法，此方法會阻塞等待，直到成功拿到獨占鎖。

這個方法就是看似復(fù)雜，但是其實就是做了一件事，那就是獲取該 segment 的獨占鎖，如果需要的話順便實例化了一下 node。

獲取鎖時，并不直接使用lock來獲取，因為該方法獲取鎖失敗時會掛起。事實上，它使用了自旋鎖，如果tryLock獲取鎖失敗，說明鎖被其它線程占用，此時通過循環(huán)再次以tryLock的方式申請鎖。如果在循環(huán)過程中該Key所對應(yīng)的鏈表頭被修改，則重置retry次數(shù)。如果retry次數(shù)超過一定值，則使用lock方法申請鎖。

這里使用自旋鎖是因為自旋鎖的效率比較高，但是它消耗CPU資源比較多，因此在自旋次數(shù)超過閾值時切換為互斥鎖。

擴容: rehash

重復(fù)一下，segment 數(shù)組不能擴容，擴容是 segment 數(shù)組某個位置內(nèi)部的數(shù)組 HashEntry\<k,v>[] 進行擴容，擴容后，容量為原來的 2 倍。

首先，我們要回顧一下觸發(fā)擴容的地方，put 的時候，如果判斷該值的插入會導(dǎo)致該 segment 的元素個數(shù)超過閾值，那么先進行擴容，再插值，讀者這個時候可以回去 put 方法看一眼。

該方法不需要考慮并發(fā)，因為到這里的時候，是持有該 segment 的獨占鎖的。

// 方法參數(shù)上的 node 是這次擴容后，需要添加到新的數(shù)組中的數(shù)據(jù)。
private void rehash(HashEntry<K,V> node) {
  HashEntry<K,V>[] oldTable = table;
  int oldCapacity = oldTable.length;
  // 2 倍
  int newCapacity = oldCapacity << 1;
  threshold = (int)(newCapacity * loadFactor);
  // 創(chuàng)建新數(shù)組
  HashEntry<K,V>[] newTable =
    (HashEntry<K,V>[]) new HashEntry[newCapacity];
  // 新的掩碼，如從 16 擴容到 32，那么 sizeMask 為 31，對應(yīng)二進制 ‘000...00011111'
  int sizeMask = newCapacity - 1;

  // 遍歷原數(shù)組，老套路，將原數(shù)組位置 i 處的鏈表拆分到 新數(shù)組位置 i 和 i+oldCap 兩個位置
  for (int i = 0; i < oldCapacity ; i++) {
    // e 是鏈表的第一個元素
    HashEntry<K,V> e = oldTable[i];
    if (e != null) {
      HashEntry<K,V> next = e.next;
      // 計算應(yīng)該放置在新數(shù)組中的位置，
      // 假設(shè)原數(shù)組長度為 16，e 在 oldTable[3] 處，那么 idx 只可能是 3 或者是 3 + 16 = 19
      int idx = e.hash & sizeMask;
      if (next == null)  // 該位置處只有一個元素，那比較好辦
        newTable[idx] = e;
      else { // Reuse consecutive sequence at same slot
        // e 是鏈表表頭
        HashEntry<K,V> lastRun = e;
        // idx 是當(dāng)前鏈表的頭結(jié)點 e 的新位置
        int lastIdx = idx;

        // 下面這個 for 循環(huán)會找到一個 lastRun 節(jié)點，這個節(jié)點之后的所有元素是將要放到一起的
        for (HashEntry<K,V> last = next;
           last != null;
           last = last.next) {
          int k = last.hash & sizeMask;
          if (k != lastIdx) {
            lastIdx = k;
            lastRun = last;
          }
        }
        // 將 lastRun 及其之后的所有節(jié)點組成的這個鏈表放到 lastIdx 這個位置
        newTable[lastIdx] = lastRun;
        // 下面的操作是處理 lastRun 之前的節(jié)點，
        //  這些節(jié)點可能分配在另一個鏈表中，也可能分配到上面的那個鏈表中
        for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
          V v = p.value;
          int h = p.hash;
          int k = h & sizeMask;
          HashEntry<K,V> n = newTable[k];
          newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
        }
      }
    }
  }
  // 將新來的 node 放到新數(shù)組中剛剛的 兩個鏈表之一 的 頭部
  int nodeIndex = node.hash & sizeMask; // add the new node
  node.setNext(newTable[nodeIndex]);
  newTable[nodeIndex] = node;
  table = newTable;
}

總結(jié)一下put的流程：

當(dāng)執(zhí)行put操作時，會進行第一次key的hash來定位Segment的位置，如果該Segment還沒有初始化，即通過CAS操作進行賦值，然后進行第二次hash操作，找到相應(yīng)的HashEntry的位置，這里會利用繼承過來的鎖的特性，在將數(shù)據(jù)插入指定的HashEntry位置時（鏈表的尾端），會通過繼承ReentrantLock的tryLock（）方法嘗試去獲取鎖，如果獲取成功就直接插入相應(yīng)的位置，如果已經(jīng)有線程獲取該Segment的鎖，那當(dāng)前線程會以自旋的方式去繼續(xù)的調(diào)用tryLock（）方法去獲取鎖，超過指定次數(shù)就掛起，等待喚醒。

get 過程分析

相對于 put 來說，get 真的不要太簡單。

1.計算 hash 值，找到 segment 數(shù)組中的具體位置，或我們前面用的“槽”

2.槽中也是一個數(shù)組，根據(jù) hash 找到數(shù)組中具體的位置

3.到這里是鏈表了，順著鏈表進行查找即可

public V get(Object key) {
  Segment<K,V> s; // manually integrate access methods to reduce overhead
  HashEntry<K,V>[] tab;
  // 1. hash 值
  int h = hash(key);
  long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
  // 2. 根據(jù) hash 找到對應(yīng)的 segment
  if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
    (tab = s.table) != null) {
    // 3. 找到segment 內(nèi)部數(shù)組相應(yīng)位置的鏈表，遍歷
    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;
}

size操作

put、remove和get操作只需要關(guān)心一個Segment，而size操作需要遍歷所有的Segment才能算出整個Map的大小。一個簡單的方案是，先鎖住所有Sgment，計算完后再解鎖。但這樣做，在做size操作時，不僅無法對Map進行寫操作，同時也無法進行讀操作，不利于對Map的并行操作。

為更好支持并發(fā)操作，ConcurrentHashMap會在不上鎖的前提逐個Segment計算3次size，如果某相鄰兩次計算獲取的所有Segment的更新次數(shù)（每個Segment都與HashMap一樣通過modCount跟蹤自己的修改次數(shù)，Segment每修改一次其modCount加一）相等，說明這兩次計算過程中無更新操作，則這兩次計算出的總size相等，可直接作為最終結(jié)果返回。如果這三次計算過程中Map有更新，則對所有Segment加鎖重新計算Size。該計算方法代碼如下

public int size() {
 final Segment<K,V>[] segments = this.segments;
 int size;
 boolean overflow; // true if size overflows 32 bits
 long sum;     // sum of modCounts
 long last = 0L;  // previous sum
 int retries = -1; // first iteration isn't retry
 try {
  for (;;) {
   if (retries++ == RETRIES_BEFORE_LOCK) {
    for (int j = 0; j < segments.length; ++j)
     ensureSegment(j).lock(); // force creation
   }
   sum = 0L;
   size = 0;
   overflow = false;
   for (int j = 0; j < segments.length; ++j) {
    Segment<K,V> seg = segmentAt(segments, j);
    if (seg != null) {
     sum += seg.modCount;
     int c = seg.count;
     if (c < 0 || (size += c) < 0)
      overflow = true;
    }
   }
   if (sum == last)
    break;
   last = sum;
  }
 } finally {
  if (retries > RETRIES_BEFORE_LOCK) {
   for (int j = 0; j < segments.length; ++j)
    segmentAt(segments, j).unlock();
  }
 }
 return overflow ? Integer.MAX_VALUE : size;
}

ConcurrentHashMap的Size方法是一個嵌套循環(huán)，大體邏輯如下：

1.遍歷所有的Segment。

2.把Segment的元素數(shù)量累加起來。

3.把Segment的修改次數(shù)累加起來。

4.判斷所有Segment的總修改次數(shù)是否大于上一次的總修改次數(shù)。如果大于，說明統(tǒng)計過程中有修改，重新統(tǒng)計，嘗試次數(shù)+1；如果不是。說明沒有修改，統(tǒng)計結(jié)束。

5.如果嘗試次數(shù)超過閾值，則對每一個Segment加鎖，再重新統(tǒng)計。

6.再次判斷所有Segment的總修改次數(shù)是否大于上一次的總修改次數(shù)。由于已經(jīng)加鎖，次數(shù)一定和上次相等。

7.釋放鎖，統(tǒng)計結(jié)束。

并發(fā)問題分析

現(xiàn)在我們已經(jīng)說完了 put 過程和 get 過程，我們可以看到 get 過程中是沒有加鎖的，那自然我們就需要去考慮并發(fā)問題。

添加節(jié)點的操作 put 和刪除節(jié)點的操作 remove 都是要加 segment 上的獨占鎖的，所以它們之間自然不會有問題，我們需要考慮的問題就是 get 的時候在同一個 segment 中發(fā)生了 put 或 remove 操作。

put 操作的線程安全性

• 初始化槽，這個我們之前就說過了，使用了 CAS 來初始化 Segment 中的數(shù)組。

• 添加節(jié)點到鏈表的操作是插入到表頭的，所以，如果這個時候 get 操作在鏈表遍歷的過程已經(jīng)到了中間，是不會影響的。當(dāng)然，另一個并發(fā)問題就是 get 操作在 put 之后，需要保證剛剛插入表頭的節(jié)點被讀取，這個依賴于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。

• 擴容。擴容是新創(chuàng)建了數(shù)組，然后進行遷移數(shù)據(jù)，最后面將 newTable 設(shè)置給屬性 table。所以，如果 get 操作此時也在進行，那么也沒關(guān)系，如果 get 先行，那么就是在舊的 table 上做查詢操作；而 put 先行，那么 put 操作的可見性保證就是 table 使用了 volatile 關(guān)鍵字。

remove 操作的線程安全性

• remove 操作我們沒有分析源碼，所以這里說的讀者感興趣的話還是需要到源碼中去求實一下的。

• get 操作需要遍歷鏈表，但是 remove 操作會"破壞"鏈表。

• 如果 remove 破壞的節(jié)點 get 操作已經(jīng)過去了，那么這里不存在任何問題。

• 如果 remove 先破壞了一個節(jié)點，分兩種情況考慮。 1、如果此節(jié)點是頭結(jié)點，那么需要將頭結(jié)點的 next 設(shè)置為數(shù)組該位置的元素，table 雖然使用了 volatile 修飾，但是 volatile 并不能提供數(shù)組內(nèi)部操作的可見性保證，所以源碼中使用了 UNSAFE 來操作數(shù)組，請看方法 setEntryAt。2、如果要刪除的節(jié)點不是頭結(jié)點，它會將要刪除節(jié)點的后繼節(jié)點接到前驅(qū)節(jié)點中，這里的并發(fā)保證就是 next 屬性是 volatile 的。

最后我們來看看并發(fā)操作示意圖

Case1：不同Segment的并發(fā)寫入

不同Segment的寫入是可以并發(fā)執(zhí)行的。

Case2：同一Segment的一寫一讀

同一Segment的寫和讀是可以并發(fā)執(zhí)行的。

Case3：同一Segment的并發(fā)寫入

Segment的寫入是需要上鎖的，因此對同一Segment的并發(fā)寫入會被阻塞。

由此可見，ConcurrentHashMap當(dāng)中每個Segment各自持有一把鎖。在保證線程安全的同時降低了鎖的粒度，讓并發(fā)操作效率更高。

以上是“java并發(fā)包ConcurrentHashMap源碼的示例分析”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對大家有幫助，更多相關(guān)知識，歡迎關(guān)注億速云行業(yè)資訊頻道！