1.前言node
看完咕泡Jack前輩的有關hashMap的視頻(非宣傳,jack自帶1.5倍嘴速,高效),收益良多,因此記錄一下學習到的東西。算法
2.基礎用法數組
源碼的註釋首先就介紹了哈希表是基於Map接口,因此它的用法和其餘集合的用法差很少。安全
/**
* Hash table based implementation of the <tt>Map</tt> interface. This * 哈希表的實現基於<tt>Map</ tt>接口。
* implementation provides all of the optional map operations, and permits * 此實現提供全部可選的映射操做,
* <tt>null</tt> values and the <tt>null</tt> key. (The <tt>HashMap</tt> * 並容許<tt> null </ tt>值和<tt> null </ tt>鍵。
* class is roughly equivalent to <tt>Hashtable</tt>, except that it is * (<tt> HashMap </ tt>類與<tt> Hashtable </ tt>大體等效,
* unsynchronized and permits nulls.) This class makes no guarantees as to * 除了它是不一樣步的,而且容許爲null。) , 此類不保證映射的順序。
* the order of the map; in particular, it does not guarantee that the order * 特別是不能保證訂單將隨着時間的推移保持不變。
* will remain constant over time.
對應的源碼,如圖所示,它繼承了抽象Map類,實現了Map接口:數據結構
public class HashMap<K,V> extends AbstractMap<K,V> implements Map<K,V>, Cloneable, Serializable { ... }app
至於具體咋用,很少介紹,推一個連接,HashMap的基礎用法:https://blog.csdn.net/lzx_cherry/article/details/98947819dom
3.存儲方式ide
下面就是介紹一個HashMap完成put(key, value)操做以後的存儲流程。性能
(1)HashMap key、value被put後的存儲方式:學習
在JDK1.7及其以前都是用的 數組+鏈表 的方式,JDK1.8以後存儲方式優化成了 數組+鏈表+紅黑樹 的方式。
(JDK1.8後,若是單鏈表存儲的長度大於8則轉換爲紅黑樹存儲,採用這樣的改善有利於解決hash衝突中鏈表過長引起的性能降低問題)
(2)圖解HashMap的主要數據結構:
<1>存儲單元 Node
圖中的每個格子表明每個Node對象。Node的信息主要包含它的存儲位置,key,value,若是在鏈表中則會有下一個Node的信息,若是存儲在紅黑樹中則包含紅黑樹的相關信息。
由上面咱們能夠寫出Node數據結構的僞代碼:
Node[] table; 數組
class Node{ Node next; } 鏈表
TreeNode(left, right, parent, boolean flag = red| black) 紅黑樹
而HashMap源碼中Node的代碼和上面僞代碼的一致:
/** * Basic hash bin node, used for most entries. (See below for * TreeNode subclass, and in LinkedHashMap for its Entry subclass.) */ static class Node<K,V> implements Map.Entry<K,V> { //經過hash算法得出的存儲位置 final int hash; //key final K key; //value V value; //鏈表的下個Node Node<K,V> next;
...
}
HashMap源碼中TreeMap的代碼(建議以前先了解紅黑樹的原理):
static final class TreeNode<K,V> extends LinkedHashMap.Entry<K,V> { TreeNode<K,V> parent; // red-black tree links TreeNode<K,V> left; TreeNode<K,V> right; TreeNode<K,V> prev; // needed to unlink next upon deletion boolean red;
... }
<2>存儲過程
根據HashMap的數據結構,能夠大體推斷出它的存儲過程。
a.先建立一個數組
b.計算出存儲key value的Node的位置
c.若是hash衝突了,判斷衝突數目的長度決定使用鏈表仍是紅黑樹結構
d.數組不夠須要進行擴容
下面對存儲過程進行細緻的分析。
1.計算出存儲key value的Node的位置
先分析這個有助於其餘的理解,這也是理解存儲過程一個比較基礎重要的內容。
計算出Node的位置,就是須要獲得Node在數組中的整型數下標,可是前提是不超出數組的大小。HashMap數組的大小能夠採用默認值,也能夠自行規定,這邊咱們採用默認的值16進行分析。
首先,要保證是個整型數,最好仍是和key value有關聯的,因此最好的方式就是經過key.hashCode()。其次,咱們須要保證Node的index大小在0~15之間。因此咱們能夠先進行一個取餘判斷,判斷: 整型數%16 = [ 0 , 15 ]。
分析取餘:
例如一個數字1,hashCode的值爲49,那麼取餘的操做就爲 49 % 16 = 1。可是這樣的取模方式還能夠進行優化,49的10進制整型數轉化爲32位二進制:
0000 0000 0000 0000 0000 0000 0011 0001 % 16 = 1
對16進行取餘,其實效果就至關於對(16-1)進行與操做:0000 0000 0000 0000 0000 0000 0011 0001 & 0111,由於與操做時候與的時候,最後四位的範圍是[0,15],若是大於15的話,就進位了,這樣能夠更有效控制整形數的範圍。
0000 0000 0000 0000 0000 0000 0011 0001
0 1111 &操做 (數組大小 - 1)
————————————————————————————————————————————————
0001(結果)
最終返回的結果就是Node在數組中的位置index了。index若是相同的話就會產生位置衝突,這時候就須要鏈表和紅黑樹數據結構,但這樣會使得咱們去獲取key value變得更加耗時。因此咱們須要儘可能保證index就是Node的位置不要太容易就出現重複的狀況。
從上的與過程當中咱們能夠看出,能決定Node位置取決在兩個相與的數(暫稱爲key1和key2),這兩個數的後4位決定了Node的位置,若是要保證hash不衝突的話,就要先分析他們。與操做,一方爲0就結果爲0,key2的最後四位值若是一個爲0的話,不管key1對應的是什麼,結果都是0,這樣極其容易致使衝突,因此咱們要儘可能保證key2除了最高爲0外。其餘位置都1。例如:01111(15)、011111(31)、0111111(63),不難看出key2的值,其實就是2的n次冪-1。因此咱們須要儘可能保證數組的大小爲2的n次冪。可是即使保證了後四位都爲1的話,畢竟只有4位,4位進行與操做,仍是很容易出現一個重複狀況,對於這種狀況,HashMap採用了異或(xor)操做( a⊕b = (¬a ∧ b) ∨ (a ∧¬b) )。
具體操做,將32位的二進制數字一分爲二,左邊16位的高16位爲一份,右邊的低16位爲另外一份。二者進行異或運算,下降重複的可能性。
如:
高16位 低16位
0101 0101 0010 1001 | 0001 0001 0001 0110
其實這就是HashMap中的hash算法,源碼:
static final int hash(Object key) { int h;
//若是key爲null則返回0,若是不爲null則返回key的hashCode高低16位異或運算的結果 return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16); }
若是再重複就只能造成鏈表和紅黑樹了。
2.建立數組以及數組的擴容
若是採用默認的大小的話,默認的數組大小爲16。源碼:
/**
* The default initial capacity - MUST be a power of two.
* 默認初始容量爲16,必須爲2的冪
* 表示1,左移4位,變成10000,也就是16 */
static final int DEFAULT_INITIAL_CAPACITY = 1 << 4; // aka 16
至於爲何不採用int DEFAULT_INITIAL_CAPACITY = 16;,一方面是省略了中間一些複雜的轉換過程,直接以二進制形式去運行運算,另外一方面也是配合2次冪的約束條件。
固然咱們知道,HashMap數組的大小也是能夠本身定義的。本身定義和默認有啥區別,若是你使用了阿里的checkstyle,初始化HashMap使用了默認的大小,這時候規約就會提示你須要本身定義
HashMap的大小。咱們能夠看一下阿里巴巴的規約:
上面說的很清楚了,若是不指定初始化的大小,容易引發屢次的擴容操做,影響性能。並給出了推薦的初始化值 = (須要存儲的元素個數 / 負載因子) + 1;
負載因子決定了數組擴容的闊值,若是一個數組大小爲20,負載因子爲0.75,那麼數組長度到達15的時候,數組就會進行擴容操做。15也就是數組擴容的闊值,0.75就稱爲負載因子,附上源碼:
源碼中的load factor也就是負載因子,規定的大小爲0.75,也就是3/4。
/** * The load factor used when none specified in constructor. */ static final float DEFAULT_LOAD_FACTOR = 0.75f;
若是本身進行初始化數組的值,那麼是否是就能夠隨意設置值了呢?看一下源碼就知道了:
初始化大小必須大於等於0,且是有最大值的。
/** * Constructs an empty <tt>HashMap</tt> with the specified initial * capacity and load factor. * * @param initialCapacity the initial capacity 初始化大小 * @param loadFactor the load factor 負載因子 * @throws IllegalArgumentException if the initial capacity is negative * or the load factor is nonpositive */ public HashMap(int initialCapacity, float loadFactor) { //若是初始化大小小於0,拋出異常 if (initialCapacity < 0) throw new IllegalArgumentException("Illegal initial capacity: " + initialCapacity); //若是初始化大小大於最大值,則將初始化值設爲最大值 if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; if (loadFactor <= 0 || Float.isNaN(loadFactor)) throw new IllegalArgumentException("Illegal load factor: " + loadFactor); this.loadFactor = loadFactor; //闊值,該方法保證了數組初始化大小爲2的次冪 this.threshold = tableSizeFor(initialCapacity); }
最大值:(1073741824)
static final int MAXIMUM_CAPACITY = 1 << 30;
闊值:
/** * Returns a power of two size for the given target capacity. */ static final int tableSizeFor(int cap) { int n = cap - 1; //位或操做,一步一步保證最後幾位都1 n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16; //若是n小於0返回1,否則返回小於最大值的n+1值 return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1; }
上面就是哈希中數組初始化的內容,接下來講一下數組擴容的方式。
1.首先須要知道的是,擴容是擴成多大的容量。
從源碼中的must be power of two,就是必須是2的n次冪就能夠推斷出擴容的方式是double,也就是將數組大小翻倍。
2.新數組如何建立,以及如何從新散列。(從新散列:把老數組中的Node移到到新的數組。)
新數組的大小咱們已經能夠肯定爲舊數組大小的2倍,如今主要的問題就是從新散列,也就是把舊的數組中Node轉移到新的數組中去。廣泛的作法就是遍歷舊的數組,將非空的Node依次賦值給新的數組。
若是Node節點下面是鏈表就遍歷鏈表,再賦值給新數組的Node節點。若是是紅黑樹,也是同樣打散重排。這樣的作法很簡單,讓咱們一塊兒看一下源碼(較長):
/** * *初始化或增長表大小。 若是爲null,則分配 *符合在現場閾值中保持的初始容量目標。 *不然,由於咱們使用的是二次冪擴展,因此 *每一個bin中的元素必須保持相同的索引或移動 *在新表中具備兩個偏移量的冪。 * * Initializes or doubles table size. If null, allocates in * accord with initial capacity target held in field threshold. * Otherwise, because we are using power-of-two expansion, the * elements from each bin must either stay at same index, or move * with a power of two offset in the new table. * * @return the table */ final Node<K,V>[] resize() { //舊的Node數組 Node<K,V>[] oldTab = table; //獲取舊的數組長度,若是爲null則返回0 int oldCap = (oldTab == null) ? 0 : oldTab.length; //舊數組的闊值 //threshold :The next size value at which to resize (capacity * load factor). //下一個要調整大小的大小值(容量*負載係數)。 int oldThr = threshold; //新的數組和新的闊值 int newCap, newThr = 0; //若是舊數組長度大於0 if (oldCap > 0) { //限制最大值 if (oldCap >= MAXIMUM_CAPACITY) { threshold = Integer.MAX_VALUE; return oldTab; } //限制新的闊值大小 else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY && oldCap >= DEFAULT_INITIAL_CAPACITY) //這邊就是對新的數組闊值進行翻倍 newThr = oldThr << 1; // double threshold } //初始化新數組的值 else if (oldThr > 0) // initial capacity was placed in threshold newCap = oldThr; else { // zero initial threshold signifies using defaults //若是以前的闊值小於0,新的數組大小設置爲16,闊值設置爲 16 * 0.75f newCap = DEFAULT_INITIAL_CAPACITY; newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY); } if (newThr == 0) { //若是以前的闊值=0,賦值給新的闊值 float ft = (float)newCap * loadFactor; newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ? (int)ft : Integer.MAX_VALUE); } //全局變量的闊值變化 threshold = newThr; @SuppressWarnings({"rawtypes","unchecked"}) Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap]; //新的哈希表 table = newTab; if (oldTab != null) { //不爲空的狀況下遍歷 for (int j = 0; j < oldCap; ++j) { Node<K,V> e; //若是Node節點不爲null if ((e = oldTab[j]) != null) { //以前的值刪除(就是設置爲null) oldTab[j] = null; //若是不爲鏈表和紅黑樹 if (e.next == null) //直接賦值給新的哈希表 newTab[e.hash & (newCap - 1)] = e; //若是Node是紅黑樹數據結構,打散重排 else if (e instanceof TreeNode) ((TreeNode<K,V>)e).split(this, newTab, j, oldCap); else { // preserve order //否則就是鏈表,對鏈表進行遍歷 賦值到新的的哈希表 Node<K,V> loHead = null, loTail = null; Node<K,V> hiHead = null, hiTail = null; Node<K,V> next; do { next = e.next; if ((e.hash & oldCap) == 0) { if (loTail == null) loHead = e; else loTail.next = e; loTail = e; } else { if (hiTail == null) hiHead = e; else hiTail.next = e; hiTail = e; } } while ((e = next) != null); if (loTail != null) { loTail.next = null; newTab[j] = loHead; } if (hiTail != null) { hiTail.next = null; newTab[j + oldCap] = hiHead; } } } } } return newTab; }
可是在賦值的過程當中,須要注意全部的位置都進行了新的一輪hash運算,在 【1.計算出存儲key value的Node的位置 】中能夠知道key2的值形式要保持是01111……11的形式。
以前的操做是這樣的:
0000 0000 0000 0000 0000 0000 0011 0001
0 1111 &操做 (數組大小 - 1)
————————————————————————————————————————————————
0001(結果) --- 1
可是咱們如今對key2的值進行了翻倍,那麼隨之與操做的結果也會變化,也就是在新的數組中Node的位置以及發生了變化,具體看下面:
0000 0000 0000 0000 0000 0000 0010 0001 key1的值 第一種狀況
0000 0000 0000 0000 0000 0000 0011 0001 key1的值 第二種狀況
01 1111 &操做 (新數組大小 = 舊數組大小 * 2,比以前左邊多了一位1)
————————————————————————————————————————————————
00 0001(第一種結果) ---1
第一種狀況,key1的值和以前的值同樣,也就是從新散列的位置不變
01 0001(第二種結果) ---17
第二種狀況,key1的值比以前的值大16(數組的長度),也就是從新散列的位置發生了變化
因此,哈希resize後,以前舊數組的Node在新數組中的位置有兩種狀況:1.保持和舊數組同樣 2.舊數組的位置+舊數組的大小
在註釋部分,就交代了「刷新」操做是會重建內部數據結構的。
* <p>An instance of <tt>HashMap</tt> has two parameters that affect its * <p> <tt> HashMap </ tt>的實例有兩個參數會影響其性能:<i>初始容量</ i>和<i>負載係數</ i>。
* performance: <i>initial capacity</i> and <i>load factor</i>. The * 容量只是建立哈希表時的容量。
* <i>capacity</i> is the number of buckets in the hash table, and the initial * <i>負載因子</ i>是衡量哈希表容許填充的程度的度量在容量自動增長以前獲取 。
* capacity is simply the capacity at the time the hash table is created. The * 當哈希表中的條目超過了負載係數和當前容量,
* <i>load factor</i> is a measure of how full the hash table is allowed to * 哈希表被<i>刷新</ i>(即內部數據結構已重建),
* get before its capacity is automatically increased. When the number of * 所以哈希表的大小大約是原來容量的2倍。
* entries in the hash table exceeds the product of the load factor and the
* current capacity, the hash table is <i>rehashed</i> (that is, internal data
* structures are rebuilt) so that the hash table has approximately twice the
* number of buckets.
*
3.key和value的put經歷
上面說的都是關於Node的位置問題,若是Node位置肯定了,那麼剩下的就只剩putNode裏面的key和value了。首先,一個數組裏面put一個Node,咱們須要思考這個位置是不是NULL,若是爲NULL的話,就在該位置new 一個Node ;若是不爲NULL,那麼就須要判斷put的內容是覆蓋原來的value仍是新增一個Node,新增又分爲鏈表新增和紅黑樹新增。具體的源碼以下:
/** * Implements Map.put and related methods 實現Map.put及相關方法 * * @param hash hash for key hash算法算出的Node位置 * @param key the key 鍵 * @param value the value to put 放置的值 * @param onlyIfAbsent if true, don't change existing value onlyIfAbsent若是爲true,請不要更改現有值 * @param evict if false, the table is in creation mode. 退出,若是爲false,則表處於建立模式。 * @return previous value, or null if none 上一個值,若是沒有則返回null */ final V putVal(int hash, K key, V value, boolean onlyIfAbsent, boolean evict) { Node<K,V>[] tab; Node<K,V> p; int n, i; //初始化哈希表 if ((tab = table) == null || (n = tab.length) == 0) n = (tab = resize()).length; //獲取在哈希表中的位置,而且判斷該位置是否爲null,若是是null 直接就建立新的Node if ((p = tab[i = (n - 1) & hash]) == null) tab[i] = newNode(hash, key, value, null); else { //若是該位置不爲null,則可能爲鏈表或者紅黑樹 Node<K,V> e; K k; //若是key值相同,hash也相同,則替換value的值 if (p.hash == hash && ((k = p.key) == key || (key != null && key.equals(k)))) e = p; //紅黑樹存儲 else if (p instanceof TreeNode) e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value); //鏈表存儲 else { //遍歷鏈表 for (int binCount = 0; ; ++binCount) { if ((e = p.next) == null) { //尾部插入 p.next = newNode(hash, key, value, null); //若是長度大於8 (TREEIFY_THRESHOLD = 8) if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st //鏈表轉換爲紅黑樹 treeifyBin(tab, hash); break; } if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) break; p = e; } } if (e != null) { // existing mapping for key V oldValue = e.value; if (!onlyIfAbsent || oldValue == null) e.value = value; afterNodeAccess(e); return oldValue; } } ++modCount; //數組長度大於闊值 if (++size > threshold) resize(); afterNodeInsertion(evict); return null; }
getNode的源碼和上面的一模一樣,也很好理解,貼一下:
/** * Implements Map.get and related methods * * @param hash hash for key * @param key the key * @return the node, or null if none */ final Node<K,V> getNode(int hash, Object key) { Node<K,V>[] tab; Node<K,V> first, e; int n; K k; if ((tab = table) != null && (n = tab.length) > 0 && (first = tab[(n - 1) & hash]) != null) { if (first.hash == hash && // always check first node ((k = first.key) == key || (key != null && key.equals(k)))) return first; if ((e = first.next) != null) { if (first instanceof TreeNode) return ((TreeNode<K,V>)first).getTreeNode(hash, key); do { if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) return e; } while ((e = e.next) != null); } } return null; }
4.ConcurrentHashMap以及線程安全
先看一下一樣是線程安全的HashTable是如何保證線程安全的:
public synchronized V put(K key, V value) { // Make sure the value is not null if (value == null) { throw new NullPointerException(); }
由上面的源碼能夠看出synchronized關鍵字直接約束了整個put方法,這樣線程雖然是安全的,可是效率過於低下。對比之下ConcurrentHashMap的鎖設計就更爲精確化,由於對於一個put方法,後者把它大體分爲幾個步驟,經過對每一個步驟進行線程安全約束來提高效率。(index:數組下標)
大體的put步驟:map.put(K,V)—> new Node[]建立數組 —> index == null(數組位置值爲null,直接建立) —> index!=null(加入鏈表,紅黑樹) —> resize()擴容
1.保證初始化哈希表線程安全
在建立數組的時候,經過樂觀鎖機制(CAS)保證只有一個線程去初始化數組;
初始化的源碼:
//putVal 方法中
if (tab == null || (n = tab.length) == 0)
//初始化 tab = initTable();
/** * Initializes table, using the size recorded in sizeCtl. */ private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { //若是SIZECTL<0,就表明已經有一個線程在執行初始化了,進行線程讓步 if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin //CAS 樂觀鎖機制保證數組初始化線程安全,若是當前對象的值==SIZECTL,則認爲線程安全,返回-1 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = tab = nt; sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; }
2.數組下標index爲null時候
若是數組下標的值爲null,也是經過樂觀鎖機制保證線程安全,源碼:
/** 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()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0) tab = initTable(); //若是數組下標的值爲null,也是經過樂觀鎖機制保證線程安全 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin }
3.數組下標不爲null
數組下標不爲null的話,那麼就是鏈表和紅黑樹結構,若是再用CAS去保證線程安全就須要對鏈表和紅黑樹中的元素依次去進行compareAndSwapInt,很麻煩。因此在這邊,咱們能夠將鏈表或者紅黑樹的頭節點鎖住,就能夠保證一整個鏈表紅黑樹的線程安全,而且影響的範圍很小。
源碼:
經過對頭節點(數組下標)的鎖,保證一整個鏈表和紅黑樹的線程安全。
//若是數組下標的值爲null,也是經過樂觀鎖機制保證線程安全 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; //若是下標不爲null,那麼就是鏈表和紅黑樹結構,若是再用CAS去保證線程安全就須要對鏈表和紅黑樹中的元素依次去進行compareAndSwapInt, //因此在這邊,咱們能夠將鏈表或者紅黑樹的頭節點鎖住,就能夠保證一整個鏈表紅黑樹的線程安全 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; }
完整的putVal源碼:
/** 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()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0) tab = initTable(); //若是數組下標的值爲null,也是經過樂觀鎖機制保證線程安全 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } //若是不是初始化數組的線程的話,就去幫忙從新散列 //MOVED 的值爲-1 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; //若是下標不爲null,那麼就是鏈表和紅黑樹結構,若是再用CAS去保證線程安全就須要對鏈表和紅黑樹中的元素依次去進行compareAndSwapInt, //因此在這邊,咱們能夠將鏈表或者紅黑樹的頭節點鎖住,就能夠保證一整個鏈表紅黑樹的線程安全 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; 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; } } } else if (f instanceof TreeBin) { Node<K,V> p; binCount = 2; if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } //每次put都會統計數組的大小,以肯定是否擴容 addCount(1L, binCount); return null; }
4.擴容的線程安全
一個線程去擴容的時候,其餘的線程必定是不能進行擴容和進行操做的,須要進入等待狀態。這樣等待狀態佔據着CPU可是卻不作事情,因此對此進行了優化,保證只有一個線程能去初始化,剩下等待的線程共同完成從新散列。
好比一個數組tab[]大小16,一個線程去負責擴容成大小32的新數組。剩下的等待線程中,a線程去從數組末尾開始向前領取一個區間的Node進行從新散列,例如區間(tab[13]~tab[15] ),a線程去負責對區間的Node進行從新散列。若是在a完成了,尚未其餘的擴容(或者put)線程進入變成等待線程的話,a就會繼續領取一個區間的任務進行散列,若是有一個線程b要進行擴容,由於擴容已經有線程在作了,b隨之進入等待狀態,這時候b線程就會去幫着a線程去完成剩下區間的散列任務。以此反覆。
分析上面的過程,側重就2個,第一個保證一個線程初始化數組,第二保證剩下的線程去幫助擴容。
源碼實現:
統計源碼:
/** * Adds to count, and if table is too small and not already * resizing, initiates transfer. If already resizing, helps * perform transfer if work is available. Rechecks occupancy * after a transfer to see if another resize is already needed * because resizings are lagging additions. * * @param x the count to add * @param check if <0, don't check resize, if <= 1 only check if uncontended */ private final void addCount(long x, int check) { CounterCell[] as; long b, s; if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true; 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); return; } if (check <= 1) return; //統計的結果s s = sumCount(); } if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; //若是s大於闊值,則須要進行擴容 while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n); //sc = 闊值 if (sc < 0) { 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); } //闊值大於0 進行初始化 樂觀鎖保證線程安全 else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount(); } } }
任務代碼:
/** * Moves and/or copies the nodes in each bin to new table. See * above for explanation. */ private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; //肯定任務的大小=16 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range //初始化數組線程 if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } //非初始化線程 int nextn = nextTab.length; ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); //不斷領取任務 boolean advance = true; //標記從新散列任務是否完成 boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; //領取散列任務 while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound; i = nextIndex - 1; advance = false; } } //執行散列 if (i < 0 || i >= n || i + n >= nextn) { int sc; //完成擴容 if (finishing) { nextTable = null; table = nextTab; //擴展改變 sizeCtl = (n << 1) - (n >>> 1); return; } //沒有完成擴容,彙報本身的完成任務 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; finishing = advance = true; i = n; // recheck before commit } } else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) advance = true; // already processed else { //遷移數據操做,和HashMap一致 synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; if (fh >= 0) { int runBit = fh & n; Node<K,V> lastRun = f; for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) { ln = lastRun; hn = null; } else { hn = lastRun; ln = null; } 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); } setTabAt(nextTab, i, ln); setTabAt(nextTab, i + n, hn); setTabAt(tab, i, fwd); advance = true; } else if (f instanceof TreeBin) { TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> lo = null, loTail = null; 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; } } 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; } } } } } }
再精細的就不會了。