ConcurrentHashMap源碼分析

構造方法

/**
 * Creates a new, empty map with the default initial table size (16). * 默認大小16 */ public ConcurrentHashMap() { } /** * 自定義大小的構造函數 */ public ConcurrentHashMap(int initialCapacity) { // 初始大小小於0拋出異常 if (initialCapacity < 0) throw new IllegalArgumentException(); int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); this.sizeCtl = cap; } /** * Returns a power of two table size for the given desired capacity. * 返回最接近傳參值的2的次方數 * 參考 Hackers Delight 第3.2節 */ private static final int tableSizeFor(int c) { int n = c - 1; n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16; return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1; } public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) // 合法性判斷 throw new IllegalArgumentException(); if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; }

 

 

初始化table

初始化table的源碼以下，關於爲何代碼13行中，爲何要重複判斷table是否爲空，舉例說明：

①當有兩個線程A和B同時觸發initTable時，此時的sizeCtl要麼是初始值，要麼是已經初始化成功後的值

②當線程A和B都進入循環時，判斷執行U.compareAndSwapInt(this, SIZECTL, sc, -1)【其中U是sun.misc.Unsafe，Unsafe是藉助C調用Cpu底層指令的類，SIZECTL是sizeCtl的偏移量】，此時線程A和B只有一個返回true，而且sc置爲-1，這一步爲原子操做

③假設線程A成功執行，sc置爲-1，那麼線程B則返回false並繼續循環，此時因爲線程A還未執行完，sc爲-1，B則一直讓出CPU

④此時線程A執行判斷sc是否大於0，大於0則賦值n爲sc，不然置爲默認值16，而後建立一個大小爲n的Node數組，並賦值到table

⑤sc=n-(n>>>2)，sizeCtl置爲新的sc值，此時線程A結束初始化，線程B可能觸發運行，這時候在代碼第13行判斷table是否爲空，就能防止從新建立table了

/**
 * Initializes table, using the size recorded in sizeCtl. */ private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; // table爲空或者長度爲0才初始化 while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) Thread.yield(); // sc用於併發的線程判斷，小於0讓出CPU else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { // 這裏重複判空，是爲了防止多線程重複建立table 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; }

 

添加元素

 

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) { // key和value爲空的拋出空指針異常 if (key == null || value == null) throw new NullPointerException(); // 重算hashcode，將hashcode的高位也應用到hashcode的取值，增長hashcode的複雜度，下降碰撞 int hash = spread(key.hashCode()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // table爲空的狀況或者長度爲0，初始化table if (tab == null || (n = tab.length) == 0) tab = initTable(); // 若是table中查詢不到該hash值的數據，則直接建立節點 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 } // 若是f.hash值爲MOVED ,即等於-1，表明有線程在處理 else if ((fh = f.hash) == MOVED) // 幫助擴容，後面分析這個方法 tab = helpTransfer(tab, f); else { V oldVal = null; // 對首個節點進行加鎖，減小線程衝突 synchronized (f) { // 這裏再次獲取hash對應的table節點，判斷節點是否一致，防止table發生改變 if (tabAt(tab, i) == f) { // fh>=0,證實是正常的節點插入，不須要考慮紅黑樹 if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; // 若是在鏈表中找到值爲key的節點e，直接設置e.val = value便可。 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } // 若是沒有找到值爲key的節點，直接新建Node並加入鏈表便可。 Node<K,V> pred = e; // put到末尾節點 if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } // 若是首節點爲TreeBin類型，說明爲紅黑樹結構，執行putTreeVal操做。 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; } } } // 計數加1，可能會觸發擴容 addCount(1L, binCount); return null; } // 查找元素是否存在，原子操做，支持併發 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); } // 新增一個元素，偏移量的計算都是((long)i << ASHIFT) + ABASE static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); }

 

附加說明

在 Node 的子類 ForwardingNode 的構造方法中，能夠看到MOVED做爲 hash 值進行了初始化。

ForwardingNode(Node<K,V>[] tab) { super(MOVED, null, null, null); this.nextTable = tab; }

而這個構造方法只在一個地方調用了，即 transfer（擴容） 方法。