JDK1.7的ConcurrentHashMap的put源碼到底是怎麼回事
-
- 相關基本概念
-
- ConcurrentHashMap的資料結構存放原理
- 了解UNSAFE操作
- 源碼中用UNSAFE操作取值
- put方法的流程
- 源碼分析
-
- ConcurrentHashMap的構造方法
- ConcurrentHashMap的put方法
- ConcurrentHashMap的ensureSegment方法
- Segment中的scanAndLockForPut方法
- Segment中的put方法
- Segment中的rehash擴容方法
JDK1.7中的HashMap是采用數組加連結清單的結構進行存儲，并且是線程不安全的，而HashTable為了線程安全就在方法上加了synchronized鎖，把整個HashTable鎖了起來，多個線程競争同一把鎖，效率不高。
綜合這兩點兩個問題，ConcurrentHashMap就來了。
相關基本概念
ConcurrentHashMap的資料結構存放原理
JDK1.7中的ConcurrentHashMap的實質也是采用數組加連結清單的方式實作。隻是它進行了一個分段。
在最外層用一個Segment類的數組，每個Segment數組的位置内部又放了HashEntry連結清單數組，而數組就存放在内部的HashEntry中。
這裡其實就可以了解成ConcurrentHashMap把HashMap進行了分段。這樣的話每個線程隻需要跟和它在一個分段上的其他線程競争鎖，而其他分段則互不影響，就提高了效率，也保證了線程的安全。
了解UNSAFE操作
CPU的執行速度要比記憶體讀取資料速度高，是以将需要運算的資料複制一份到CPU的高速緩存中，也就是給目前運作線程的運作記憶體的高速緩沖中放入副本。每個線程第一次會從主存中将變量拿到高速緩沖中，以後每次線程運作所需變量都從該線程自己的高速緩存中取，而不是從主存中取，運算結束後再将高速緩沖中的資料重新整理到主存中。這樣就會導緻每個線程取到的有可能不是主存中最新的值，那麼計算的結果也就是錯誤的。
ConcurrentHashMap源碼中，為了解決上面的這種資料的不一緻性，用了
UNSAFE操作
方法操作。舉個例子【UNSAFE不能直接使用，需要用反射擷取，這裡隻是寫個僞代碼友善了解】：
//有一個數組
private String table[] = {"1", "2", "3", "4"};
// 數組的類型,用于下面arrayIndexScale和arrayBaseOffset的參數
Class arrayClass = String.class;
//擷取數組存儲對象的對象頭大小
int ns = UNSAFE.arrayIndexScale(arrayClass);
//數組中第一個元素的起始位置
int base = UNSAFE.arrayBaseOffset(arrayClass);
//擷取table[1]在記憶體中的最新值
String value = UNSAFE.getObject(table,base+1*ns);
//擷取table[2]在記憶體中的最新值
String value = UNSAFE.getObject(table,base+2*ns);
//擷取table第i個位置在記憶體中的最新值
String value = UNSAFE.getObject(table,base+i*ns);
這個UNSAFE方法其實就和java中的volatile修飾符有着同樣的作用。這上面具體的參數base，ns這些的原理就不作深入了，大概知道這個UNSAFE有什麼用，怎麼用的就行了。
源碼中用UNSAFE操作取值
那麼這個UNSAFE操作在源碼中的用法和上面說的差不多，稍微有些變化。這裡舉例取外層數組Segment[j]位置的值，源碼中是這麼寫的
s = (Segment<K,V>)UNSAFE.getObject(segments, (j << SSHIFT) + SBASE）
前面的(Segment<K,V>)是類型強轉，方法中的第一個參數就是要從中取數的數組對象。主要看後面的UNSAFE.getObject中的最後一個參數，這些參數都會在源碼中的一個靜态代碼塊中賦好值。【sc,ss,SSHIFT就是取外層Segment的相關參數。tc,ts,TSHIFT就是取取内層HashEntry數組的相關參數】
是以UNSAFE.getObject(segments, (j << SSHIFT) + SBASE）中的SBASE就是調用arrayBaseOffset方法擷取的數組中第一個元素的起始位置base，SSHIFT調用的方法中的參數ss就是調用arrayIndexScale擷取的擷取數組存儲對象的對象頭大小ns，這裡翻譯一下就是
UNSAFE.getObject(segments, base +(j<<（31-Integer.numberOfLeadingZeros(ns))））
唯一不同的就是原本我們是用i*ns取第i個位置上的元素
源碼中是用 j<<（31-Integer.numberOfLeadingZeros(ns))
這裡先解釋一下31-Integer.numberOfLeadingZeros(ns)的作用
Integer.numberOfLeadingZeros會傳回某個數字最高位1前面0的個數。以16舉例子：
16的二進制是 0000 0000 0000 0000 0000 0000 0001 0000 它前面共有27個0，以就會傳回27
因為二進制的第一位表示2的0次方，而實際計算左移的時候，左移i位就是乘2的i次方，而不會是乘2的（i-1）次方，是以用整形的32位去掉一位後再減去最高位的1後面的0，剩下的就是小于原來數字最大的二次幂數。
31-Integer.numberOfLeadingZeros(ns) <=> log2(ns)向下取
最後再進行左移 j<<（31-Integer.numberOfLeadingZeros(ns)) <=> j*2[log2(ns)向下取整] <=> j*(小于ns的最大二次幂數)
這個方法分析出來和我們原來的i*ns好像有點差距，這個原因不太了解為啥，最後算出來的ns要往下取小于ns的最大二次幂數
這裡就暫且能大概了解一下這個UNSAFE.getObject(segments, (j << SSHIFT) + SBASE）在源碼中的寫法就是取segments數組中第j位就行了，并且能保證一定是從主拿最新的值，而不是重緩存拿。可能具體的原理還要深入反射，JVM，CAS相關的東西。
put方法的流程
既然ConcurrentHashMap分為了内外兩層，是以肯定先要在外層的Segment數組進行hash算出個值，然後進到Segment這個值下标得位置裡面，進入Segment内部後，先對該分段加鎖，然後再算一次hash，找到内部得Entry數組的下标位置，然後在進行元素的插入。是以大緻流程可以分為下面幾步：
- 外層計算hash值，找到外層Segment數組的對應下标位置
- 進入Segment數組中，對該分段加鎖
- 再次計算hash值，确定内部的Entry數組的下标
- 放入值
大概的流程就這幾步，但肯定不會這麼籠統簡單的完成了，下面看源碼。
源碼分析
ConcurrentHashMap的構造方法
構造方法需要三個參數，initialCapacity可存放元素容量，loadFactor加載因子，concurrencyLevel表示外層數組segment的大小，并規定不能超過定義好的外層數組容量最大值MAX_SEGMENTS。
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;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// create segments and segments[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];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
中間過程先不看，看最後它先把segment的第一個位置上的對象new了出來，這裡的作用就是，在後面每次擴容或相關操作的時候，就可以直接從s0裡面直接取一些需要的資料，比如内層資料大小等，就不用每次再計算一遍。
這裡可以看到，cap作為了内層數組HashEntry的大小，而ssize作為最終外層數組的容量大小，而沒有取concurrencyLevel。那麼也就是說中間的一段代碼都在計算内外層數組的額容量大小。
那就來分析一下中間這段代碼到底做了什麼事情
ConcurrentHashMap和HashMap一樣，都要求數組的長度為2的幂，不管内層還是外層數組容量。是以先要對我們傳入的concurrencyLevel進行處理，while中用ssize與concurrencyLevel比較，小于的話，每次左移動一位，也就是每次都乘以2，這樣就能找到大于等于concurrencyLevel的最小的2次幂值sszie，是以最終外層segment的容量取的是ssize，而不是concurrencyLevel。
而sshift同樣在while中，它每次+1，記錄的是ssize 到底左移了幾次。
中間兩個segmentShift和segmentMask的指派先不看，這裡先明确一下segmentShift是32-ssize左移次數，也就是32-log2(外層數組容量),segmentMask是ssize-1，也就是外層數組容量-1
接着判斷存放元素的容量是否大于的規定的最大容量
int c = initialCapacity / ssize 就是計算了每個segment的内部平均要放幾個元素，由于/除法會舍掉小數，是以還用個if判斷是否需要+1才能保證每個分段中的存放的元素總和大于等于傳入的元素大小。但這裡還不是内部HashEntry的數組容量大小！
因為前面說了，内部的數組容量同樣要為2的n次幂，是以下面又對cap進行了與c比較的左移循環，目的同樣是為了找到大于等于c的最小2次幂作為内部數組的容量。
ConcurrentHashMap的put方法
這裡的ConcurrentHashMap是不允許存放null值的，是以這裡判斷了value==null的話，會抛出個NullPointerException()異常。
接着看到下面的if中先用UNSAFE取了segments中第j個位置的Segment對象值賦給s，并判斷s是否為null。為null的話，調用ensureSegment方法在這個位置上建立一個Segment對象。不為null的話，再調用segments中的put方法進行鍵值對的放入。
既然j是數組下标的話，那麼hash方法和
(hash >>> segmentShift) & segmentMask;
這一行就是用來計算定位外層數組下标位置的方法。
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
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);
}
hash方法不做過多的研究，之前分析HashMap的源碼說過，就是為了讓結果更散列，得到一個hash值。
主要分析
(hash >>> segmentShift) & segmentMask;
這一行代碼
先把前面的hash >>> segmentShift看成某個值x，而segmentMask是2的次幂-1，轉化為二進制後就是低位全是連續的1，比如8-1=7的二進制就是0000 0000 0000 0000 0000 0000 0000 0111
那麼它和任意值進行與&操作後，所得結果的範圍都會在0-7之間，這樣也就保證了計算出的j這個外層數組下标不會越界。
hash是根據key算出來的值，segmentShift是32-log2(外層數組容量),segmentMask是外層數組容量-1
再說的直白一點segmentShift就是數組容量轉為二進制後最高位左面0的個數 + 1。比如8=23，轉為二進制就是0000 0000 0000 0000 0000 0000 0000 1000 ，那麼segmentShift = 28 + 1 = 29
hash右移了segmentShift位之後，就能讓hash的高位轉都到低位，且轉到低位的數量與外層數組容量有關。
這裡為什麼要進行hash >>> segmentShift這樣的操作，也沒太懂，就暫且了解為都是為了讓所計算出的j數組下标更散列，減少哈希沖突。
後面的ensureSegment就是把建立出來的Segment放入指定下标，最後再調用Segment中的put方法把元素放入。
ConcurrentHashMap的ensureSegment方法
這個方法用來在segement的指定下标位置建立一個Segment對象
這裡的u就是前面說到的UNSAFE操作中的數組下标的意思。每次進來都先用UNSAFE操作取最新值去判斷這個u位置上的Segment對象是否為null。
因為錢買你構造方法中說過了在構造ConcurrentHashMap的時候就會把Segment數組中的第一個位置的對象建立出來，友善後面再建立Segment對象的時候，可以減少計算，直接去第一個位置上的對象中取資料。
這裡的proto就是發揮了這個作用
從proto中取出内層HashEntry數組的容量cap，取出加載因子ld，兩者計算得到擴容門檻值thresolad。
接着就開始建立該Segment對象内部的HashEntry數組對象，new出一個tab後，再次确認一下，在上訴過程中，是否有其他線程對該位置上的Segment對象先一步new好，是以再去UNSAFE取一下，如果還為null的話，才是真正的執行
new Segment<K,V>(lf, threshold, tab);
。到這一步，也僅僅是吧這個對象new出來了，并沒有放入Segment數組中指定的位置。
是以new完後的下一步還是通過UNSAFE操作再去擷取一次最新的對象，如果還為null，這裡就執行cas操作
UNSAFE.compareAndSwapObject(ss, u, null, seg = s)
。
這個表示的意思是，判斷ss中第u個位置是否為null，如果是的話，就把這個位置上seg的值更新為s。
這樣就完成了Segment數組上指定下标位置的Segment對象的建立，并且保證了原子性。
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<K,V> proto = ss[0]; // use segment 0 as prototype
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}
Segment中的scanAndLockForPut方法
在看Segment中的put方法之前，會看到它裡面的第一步就用到了scanAndLockForPut()方法，先看這個方法有什麼用。
這裡的entryForHash方法就直接說了，是取内層HashEntry數組中這個hash值算出來對應下标位置的元素
這裡面最主要的就是在一個while循環裡面一直去擷取鎖tryLock()，隻有當擷取到了才傳回。
再嘗試擷取鎖的過程中，他還幹了一些事情，每嘗試擷取一次鎖，它就會往下去周遊該位置上的連結清單，如果enull并且nodenull的話，這裡就表示周遊到尾部了也沒有key相等的元素或者這個位置上根本就沒有元素，那麼這個時候我們就把目前要放入的key-value，new成一個對象node，然後修改retries為0，讓下一次的while嘗試擷取鎖時，進入下面的else if中。
在下面的else if中，他會先判斷目前重試擷取鎖的次數是否達到上限，達到了的話，就會調用lock()阻塞的方法一直等待擷取鎖，不做别的事情。沒達到的話它會再一次的調用entryForHash去擷取該連結清單最新的頭節點，因為擷取不到鎖，并且在每次嘗試擷取和周遊連結清單的過程中，它以頭節點的方式新插入了一個節點，那麼有可能新插入的節點和我目前要插入的節點的key相同，是以又會修改retries回到-1，重新走上面的if中的邏輯。
最終傳回的node不為null的話就是以已經把key和value打包好的Entry對象，如果為null的話，說明周遊的途中擷取到鎖了，提前傳回出來了。
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
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {
if (e == null) {
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
Segment中的put方法
ConcurrentHashMap是線程安全的。該方法進入首先就會嘗試擷取鎖tryLock()，在鎖裡會傳回一個node對象，上面的scanAndLockForPut方法已經分析過，這裡擷取的node是null或者是已經把key和value打包好的Entry對象。如果如果擷取不到的話調用scanAndLockForPut()方法。
進來先計算一下要放在内部數組的下标index，(tab.length - 1) & hash，這種某個值與數組長度-1的與操作都是為了控制計算範圍不越界，并且盡可能的讓計算到每個位置的機率相近。
接着調用entryAt()方法去擷取内部數組index位置的内容，同樣的是用UNSAFE操作去擷取記憶體中的最新值。這裡擷取到的first 就是内層數組指定下标index位置連結清單上的第一個節點。
接着從這個節點開始周遊,每次周遊到的節點賦給e，有下面幾種情況的時候break：
1.e不為null，這個位置上已經有元素了，那麼就判斷目前要放入的key是否和目前該連結清單上周遊到的元素的key相同，如果相同，在根據傳入的onlyIfAbsent判斷，是否需要更新value的值。
2.e為null，這裡表示周遊完了連結清單沒有相同的key，那麼如果前面的node傳回不為null的話，那麼就直接修改node的next屬性為first，否者就new一個Entry對象，同樣的是把next指向first。
接着判斷count+1，即元素的個數是否超過一定值【這裡的thresold擴容門檻值在HashMap源碼中已經解釋了】，如果超過了就rehash方法進行擴容。注意這裡的擴容隻會對内層HashEntry數組進行擴容，對外層的Segment數組大小不會改變，最後用setEntryAt方法，把node放在該數組下标位置上，即完成頭插法的最後一步。最終unlock釋放鎖就完成了整個ConcurrentHashMap的put操作。
static final <K,V> HashEntry<K,V> entryAt(HashEntry<K,V>[] tab, int i) {
return (tab == null) ? null :
(HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)i << TSHIFT) + TBASE);
}
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
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;
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
Segment中的rehash擴容方法
擴容隻對内層數組進行擴容，不是對外層數組！
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // Single node on list
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
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);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
這裡的擴容代碼不細分析，大體看一下，裡面有兩個for循環，并且擴容和HashMap一樣，原本第i個位置上的元素可能放到新數組的第i個位置，也可能放到第i+擴容大小上的位置，就是下面這種情況。
那麼就有一種情況，該位置上從某個節點開始到連結清單結尾的所有節點，都放到的是同一個位置，那麼就隻需要把這一段内的第一個節點放過去即可。類似下面這種情況，那麼我在移動的時候就不需要每個元素都一個個的移動，我隻需要周遊到2元素的時候，把2直接放到0的下面即可。
是以上面的源碼中兩個循環，第一個循環就是在找某一段連續到最後的連結清單區間最後放到的是新數組的同一個位置的開始節點。
第二個循環的時候，就會從0開始周遊，到2結束，把其中的周遊的元素組個放入新數組的對應位置即可。這就通過兩個循環達到了擴容的目的。