Java 容器類 - Map
sschrodinger
2019/03/24
參考
《算法》第四版 - Robert.S 著，謝路云 譯
AbstractMap
EntrySet
類似的，AbstractMap 實(shí)現(xiàn)了 Map 的基礎(chǔ)框架，在這個(gè)框架中，最重要的類是 EntrySet 結(jié)構(gòu)。Map 實(shí)現(xiàn)不支持迭代器，EntrySet 將鍵值對(duì)包裝成一個(gè) Set，這樣就可以對(duì) Set 做迭代，實(shí)現(xiàn)對(duì) Map 的遍歷。
舉例來說，containsValue(Object value) 函數(shù)在內(nèi)部使用 entrySet 的迭代器遍歷整個(gè) Map，如果找到就返回 true，否則返回 false。
public boolean containsValue(Object value) {
Iterator<Entry<K,V>> i = entrySet().iterator();
if (value==null) {
while (i.hasNext()) {
Entry<K,V> e = i.next();
if (e.getValue()==null)
return true;
}
} else {
while (i.hasNext()) {
Entry<K,V> e = i.next();
if (value.equals(e.getValue()))
return true;
}
}
return false;
}
鍵視圖和值視圖
和 List 的 subList 類似，Map 支持返回鍵視圖和值視圖。因?yàn)?Map 的值要求唯一，而 value 不做要求，所以 Map 的鍵視圖采用 Set 存儲(chǔ)，值視圖采用 Collection 存儲(chǔ)，兩個(gè)視圖如下所示：
transient Set<K> keySet;
transient Collection<V> values;
懶加載模式
所謂的懶加載模式，即在創(chuàng)建類的時(shí)候不創(chuàng)建視圖，而是在使用視圖的時(shí)候，如調(diào)用 keySet() 函數(shù)時(shí)才創(chuàng)建視圖。
如下，當(dāng) keySet 不為空時(shí)，直接返回 keySet；如果為空，才創(chuàng)建 keySet。
public Set<K> keySet() {
Set<K> ks = keySet;
if (ks == null) {
ks = new AbstractSet<K>() {
//...
};
keySet = ks;
}
return ks;
}
受限制的視圖
和 AbstractList 的 subList 所不同的是，AbstractMap 提供的鍵視圖和值視圖是受限制的視圖。我們可以在 subList 中刪除元素，增加元素，取決于他的原 List 可以支持哪些操作，但是 AbstractMap 的鍵視圖和值視圖只支持部分操作。
KeySet 的實(shí)現(xiàn)重寫了 AbstractSet，但是在 AbstractSet 中，所有的操作都被設(shè)置成了拋出 OperationNotSupportException，所以只有其重寫的函數(shù)和該 Map 的實(shí)現(xiàn)所綁定。
下面是 AbstractMap 的 keySet 實(shí)現(xiàn)。
new AbstractSet<K>() {
public Iterator<K> iterator() {
return new Iterator<K>() {
private Iterator<Entry<K,V>> i = entrySet().iterator();
public boolean hasNext() {
return i.hasNext();
}
public K next() {
return i.next().getKey();
}
public void remove() {
i.remove();
}
};
}
public int size() {
return AbstractMap.this.size();
}
public boolean isEmpty() {
return AbstractMap.this.isEmpty();
}
public void clear() {
AbstractMap.this.clear();
}
public boolean contains(Object k) {
return AbstractMap.this.containsKey(k);
}
};
HashMap
特性
HashMap 是平常用的最多的一個(gè) Map 實(shí)現(xiàn)，我們列出其幾大特性：
特性
- 基于 Hash table（哈希表）的實(shí)現(xiàn)
- 提供所有 map 接口，允許 null 的鍵和 null 的值
- 非線程安全
- 不保證讀取鍵值對(duì)的順序，不僅僅包括鍵值對(duì) put 和 get 順序不同，也包括隨時(shí)間變化，迭代的順序也可能不同。
- 對(duì)于基本的操作（get 和 put），可以在常數(shù)時(shí)間上完成。
HashMap 實(shí)現(xiàn)原理
散列表 的作用相當(dāng)于索引，通過散列表可以快速的定位元素的位置。散列表依據(jù)如何解決沖突可以劃分為多種散列表，在 HashMap 的實(shí)現(xiàn)中，采用單獨(dú)鏈表法解決沖突。
首先看 Hash table 的原理，簡要的概括， hash table 的思想就是分區(qū)。
一個(gè)簡化版的 hash map 如下圖所示：
對(duì)于需要保存的每一個(gè) entry，求出其 hash 值，并與 hash table 的大小做取模運(yùn)算（保證每一個(gè) hash 值都可以映射到 hash table 中），然后，將取模運(yùn)算之后的 hash 值對(duì)應(yīng)的 entry 連接到 hash table 對(duì)應(yīng)的項(xiàng)中，如果已經(jīng)有元素，則將其連接到已有元素的后面。
hashMap 采用 Node 類作為其 entry 節(jié)點(diǎn)，Node 類如下所示：
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
V value;
Node<K,V> next;
Node(int hash, K key, V value, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
public final K getKey() { return key; }
public final V getValue() { return value; }
public final String toString() { return key + "=" + value; }
public final int hashCode() {
return Objects.hashCode(key) ^ Objects.hashCode(value);
}
public final V setValue(V newValue) {
V oldValue = value;
value = newValue;
return oldValue;
}
public final boolean equals(Object o) {
if (o == this)
return true;
if (o instanceof Map.Entry) {
Map.Entry<?,?> e = (Map.Entry<?,?>)o;
if (Objects.equals(key, e.getKey()) &&
Objects.equals(value, e.getValue()))
return true;
}
return false;
}
}
Node 實(shí)際上是一個(gè)鏈表節(jié)點(diǎn)，這樣才能實(shí)現(xiàn) hash table 的鏈表引用。
在 HashMap 的實(shí)現(xiàn)中，采用 Node 數(shù)組作為哈希表結(jié)構(gòu)。
結(jié)構(gòu)如下圖：
transient Node<K,V>[] table;
舉個(gè)例子，我們使用如下程序初始化：
private void mapTest() {
Map<String, String> stringMap = new HashMap<>();
stringMap.put("key_1", "value_1");
stringMap.put("key_2", "value_2");
stringMap.put("key_3", "value_3");
stringMap.put("key_4", "value_4");
}
初始化之后，存儲(chǔ)的結(jié)構(gòu)如下圖所示：
我們來看 HashMap 具體怎么實(shí)現(xiàn)存儲(chǔ)。
put 過程
根據(jù)散列表原理，對(duì)每一個(gè)加入散列表的元素都要求他的哈希值找到對(duì)應(yīng)的存儲(chǔ)位置。在 hashMap 中，我們使用 key 查找值，所以在散列表中，只要能根據(jù) key 的 hash 值快速找到 Node 節(jié)點(diǎn)，就可以定位 value 值，所以 hashMap 使用 key 的 hash 值作為 hash table 的索引。
以下是 put 函數(shù)的部分源代碼：
public V put(K key, V value) {
return putVal(hash(key), key, value, false, true);
}
//onlyIfAbsent:if true, don't change existing value
//evict:if false, the table is in creation mode
final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
boolean evict) {
// step 1: resize if necessary
Node<K,V>[] tab; Node<K,V> p; int n, i;
if ((tab = table) == null || (n = tab.length) == 0)
n = (tab = resize()).length;
// step 2: if no entry in current tab, create a new node
if ((p = tab[i = (n - 1) & hash]) == null)
tab[i] = newNode(hash, key, value, null);
else {
Node<K,V> e; K k;
// step 3: check if first entry key equals key
if (p.hash == hash &&
((k = p.key) == key || (key != null && key.equals(k))))
e = p;
else {
// step 4: this loop state search entry key equals key in s specific hash or create a new entry if no equals entry found
for (int binCount = 0; ; ++binCount) {
if ((e = p.next) == null) {
p.next = newNode(hash, key, value, null);
break;
}
if (e.hash == hash &&
((k = e.key) == key || (key != null && key.equals(k))))
break;
p = e;
}
}
// step 5: modifify if exist
if (e != null) { // existing mapping for key
V oldValue = e.value;
if (!onlyIfAbsent || oldValue == null)
e.value = value;
return oldValue;
}
}
++modCount;
if (++size > threshold)
resize();
return null;
}
最主要的函數(shù)是 putVal 函數(shù)，他把 key 的哈希值，key 值和 value 值最為參數(shù)，總共分為五步，如下所示：
put 步驟
- step 1：如果 table 為空或者 table 長度為 0，重新配置 table
- step 2：利用
(n - 1) & hash找到 entry 所在的鏈表頭。如果鏈表頭為空，直接創(chuàng)建新節(jié)點(diǎn)。
- 用
n - 1和 hash 做與操作，可以將結(jié)果限制在 0 到 n - 1 中，優(yōu)點(diǎn)是速度快，缺點(diǎn)是相當(dāng)于只有低位參加了 hash 的過程，導(dǎo)致碰撞幾率增大。
- step 3：檢查鏈表頭的 key 是否等于所期望的 key。如果一致，記錄當(dāng)前 node
- 需要滿足兩個(gè)條件
p.hash == hash和(k = p.key) == key || (key != null && key.equals(k))
- 第一個(gè)條件指的是同一個(gè)鍵的 hash 結(jié)果應(yīng)該維持不變，所以先檢查 hash 結(jié)果，如果 hash 結(jié)果不一樣，則鍵肯定不一樣（一致性）
- 第二個(gè)條件是指滿足所期望的 key 和鏈表頭的 key 為同一對(duì)象（同一內(nèi)存）或者 equal 函數(shù)相同
- step 4：遍歷鏈表，如果滿足當(dāng)前遍歷的節(jié)點(diǎn)和期望的 key 相同，記錄當(dāng)前 node，break；如果鏈表中沒有滿足要求的 node，新建節(jié)點(diǎn)在鏈表最后。
- step 5：對(duì)于 step 3 和 step 4，如果存在滿足 key 條件的節(jié)點(diǎn)，表明在原來的鏈表中有記錄，根據(jù) onlyIfAbsent 參數(shù)決定是否更新。
get 過程
get 過程比較簡單，根據(jù) key 值遍歷 Map，如果有元素，則返回值，如果沒有，則返回空，代碼如下：
public V get(Object key) {
Node<K,V> e;
return (e = getNode(hash(key), key)) == null ? null : e.value;
}
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;
}
hash 函數(shù)
因?yàn)樵?get 方法時(shí)使用了 (n - 1) & hash，而不是取模運(yùn)算，相當(dāng)與只有低位參加了運(yùn)算，所以碰撞幾率相當(dāng)?shù)母?，為了減少碰撞幾率，hashMap 使用了一個(gè)支持方法 hash ，將 key 值 hash 的高位和低位做與或運(yùn)算，使得高位也參加到 hash 的計(jì)算中，減少碰撞幾率。
static final int hash(Object key) {
int h;
return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}
resize 實(shí)現(xiàn)
每當(dāng)哈希表的使用率大于 $loadFactory * capicity$ 時(shí)，會(huì)自動(dòng)擴(kuò)大哈希表，標(biāo)準(zhǔn)是每次擴(kuò)大兩倍。
擴(kuò)大兩倍可以很好的解決元素重新排列的問題。我們知道 hashMap 使用 $ (n - 1) \& hash$ 作為哈希和表的映射，每一個(gè)元素需要調(diào)整的位置只能是當(dāng)前位置或者是當(dāng)前位置加上原來容量之后的位置，高位決定了是當(dāng)前位置還是兩倍之后的位置。
假設(shè)當(dāng)前容量大小為 $16$($0x10000$)，某一元素的哈希值為 $44$($0x101100$)，不考慮內(nèi)置的 hash 函數(shù)，直接將 15 和 44 做與操作，那么
會(huì)得到 $0x1100$，即在原來應(yīng)該放在 12 號(hào)位。容量擴(kuò)大兩倍，實(shí)際上是對(duì) 16 向左移動(dòng)了 1 位，得到 32，當(dāng)和 44 做與操作時(shí)，實(shí)際上前 4 位不會(huì)變動(dòng)，只有第五位可能有區(qū)別，在這個(gè)例子中，做與操作仍然是放在 12 號(hào)位，當(dāng)然第五位可能是 1，變成當(dāng)前位置加上原來容量之后的位置。這樣就不用重新計(jì)算每一個(gè)的位置，而只需要計(jì)算高位不同的 hash 值并將存放位置加上當(dāng)前容量的偏移。
note
- 最大 table 長度為
$1 << 30$
代碼如下：
final Node<K,V>[] resize() {
Node<K,V>[] oldTab = table;
int oldCap = (oldTab == null) ? 0 : oldTab.length;
int oldThr = threshold;
int newCap, newThr = 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
newCap = DEFAULT_INITIAL_CAPACITY;
newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY);
}
if (newThr == 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;
if ((e = oldTab[j]) != null) {
oldTab[j] = null;
if (e.next == null)
newTab[e.hash & (newCap - 1)] = e;
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;
}
迭代器構(gòu)建
HashMap 的迭代器都基于抽象類 HashIterator，都是快速失敗迭代器。
HashIterator 基于深度遍歷的思想，首先遍歷一個(gè)鏈表結(jié)構(gòu)，然后遍歷下一個(gè)鏈表結(jié)構(gòu)。
代碼如下：
abstract class HashIterator {
Node<K,V> next; // next entry to return
Node<K,V> current; // current entry
int expectedModCount; // for fast-fail
int index; // current slot
HashIterator() {
expectedModCount = modCount;
Node<K,V>[] t = table;
current = next = null;
index = 0;
if (t != null && size > 0) { // advance to first entry
// 將 next 指向第一個(gè)存在的（不為空）的鏈表
do {} while (index < t.length && (next = t[index++]) == null);
}
}
public final boolean hasNext() {
return next != null;
}
final Node<K,V> nextNode() {
Node<K,V>[] t;
Node<K,V> e = next;
if (modCount != expectedModCount)
throw new ConcurrentModificationException();
if (e == null)
throw new NoSuchElementException();
if ((next = (current = e).next) == null && (t = table) != null) {
do {} while (index < t.length && (next = t[index++]) == null);
}
return e;
}
public final void remove() {
Node<K,V> p = current;
if (p == null)
throw new IllegalStateException();
if (modCount != expectedModCount)
throw new ConcurrentModificationException();
current = null;
K key = p.key;
removeNode(hash(key), key, null, false, false);
expectedModCount = modCount;
}
}
在 HashIterator 的基礎(chǔ)上，hashMap 實(shí)現(xiàn)了自己的兩個(gè)迭代器，如下：
final class KeyIterator extends HashIterator
implements Iterator<K> {
public final K next() { return nextNode().key; }
}
final class ValueIterator extends HashIterator
implements Iterator<V> {
public final V next() { return nextNode().value; }
}
final class EntryIterator extends HashIterator
implements Iterator<Map.Entry<K,V>> {
public final Map.Entry<K,V> next() { return nextNode(); }
}
Java 1.8 性能改進(jìn)
HashMap 最重要的三個(gè)參數(shù)， threshold 、 loadFactor 和 capacity。
capcity 指的是 table 的大小。
loadFactory 是一個(gè)比例，指 table 的填充比例，即當(dāng) table 中的元素填充個(gè)數(shù)大于 $ capacity * loadFactor$ 時(shí)（數(shù)組中有 $ capacity * loadFactor$ 個(gè)項(xiàng)不為空），則擴(kuò)充節(jié)點(diǎn)。
不同于低版本的 HashMap 實(shí)現(xiàn)， 1.8 增加了 threshold 指的是當(dāng)單鏈表的長度大于 threshold 時(shí)，將單鏈表重新組合成紅黑樹形式的存儲(chǔ)結(jié)構(gòu)，增加讀取效率。詳細(xì)原理可參見 treeMap
treeMap
特性
TreeMap 是基于紅黑樹實(shí)現(xiàn)的一個(gè) map，有幾大基本特性。
特性
- 基于紅黑樹實(shí)現(xiàn)
- 提供所有 map 接口。
- 非線程安全
- 迭代時(shí)的順序按照鍵值排列，即存儲(chǔ)順序是有限的（在這個(gè)基礎(chǔ)上，同時(shí)實(shí)現(xiàn)了
NavigableMap接口，用于查找某個(gè) key 之前的所有元素等操作）- 對(duì)于 put 操作，時(shí)間復(fù)雜度是
$O(logn)$，對(duì)于增加和刪除操作，時(shí)間復(fù)雜度不能保證- 支持
SortedMap接口，如 firstKey()，lastKey()，取得最大最小的key，或sub(fromKey, toKey), tailMap(fromKey) 剪取Map的某一段
TreeMap 實(shí)現(xiàn)原理
2-3 查找樹
利用樹進(jìn)行查找時(shí)，希望樹盡量的平衡，這樣才能夠保證在每一次的查找保證 $O(logn)$ 的復(fù)雜度。在動(dòng)態(tài)插入的過程種要維持平衡二叉樹的代價(jià)太高，所以使用一種新型的平衡樹 - 2-3 查找樹。
對(duì)于一個(gè)二叉查找樹，他的每一個(gè)節(jié)點(diǎn)有一個(gè)值和兩條連接，左連接指向的二叉查找樹的值都小于該節(jié)點(diǎn)，右連接指向的二叉查找樹的值都大于該節(jié)點(diǎn)，對(duì)于一個(gè)整數(shù)類型的數(shù)組 int[] a = new int[] {1,2,3,4,5,6,7}，他所構(gòu)成的平衡二叉查找樹如下所示：
現(xiàn)引入 2-3 查找樹，定義如下：
定義
- 為一棵空樹或由以下節(jié)點(diǎn)組成
- 2-節(jié)點(diǎn)，含有一個(gè)值和兩條連接，左連接指向的 2-3 樹中的值都小于該節(jié)點(diǎn)，右連接指向的 2-3 樹的值都大于該節(jié)點(diǎn)（類似于查找二叉樹）
- 3-節(jié)點(diǎn)，含有兩個(gè)值和三條連接，左連接指向的 2-3 樹中的值都小于該節(jié)點(diǎn)，中連接指向的 2-3 樹的值位于該節(jié)點(diǎn)的兩個(gè)值之間，右連接指向的 2-3 樹的值都大于該節(jié)點(diǎn)
對(duì)于一個(gè)字符數(shù)組 char[] chars = new char[] {A,C,H,L,P,S,X,E,J,R,M}，他的平衡 2-3 樹如下所示：
查找
2-3 樹查找過程和二叉樹相似，略。
添加
在一個(gè)只有根節(jié)點(diǎn)且是 2-節(jié)點(diǎn)的樹上添加元素。為了保證平衡，我們需要將該節(jié)點(diǎn)替換成一個(gè) 3-節(jié)點(diǎn)，如下所示：
在一個(gè)只有根節(jié)點(diǎn)且是 3-節(jié)點(diǎn)的樹上添加元素。為了保證平衡，我們需要將該節(jié)點(diǎn)做局部變化，操作如下：首先將該節(jié)點(diǎn)臨時(shí)增加一個(gè)值變成 4-節(jié)點(diǎn)，然后對(duì)四節(jié)點(diǎn)進(jìn)行拆分，變成 3 個(gè) 2-節(jié)點(diǎn)，如下所示：
在一個(gè)父節(jié)點(diǎn)且是 2-節(jié)點(diǎn)，該節(jié)點(diǎn)是3-節(jié)點(diǎn)的樹上添加元素。為了保證平衡，我們需要將該節(jié)點(diǎn)做局部變化，操作如下：首先將該節(jié)點(diǎn)臨時(shí)增加一個(gè)值變成 4-節(jié)點(diǎn)，然后對(duì)四節(jié)點(diǎn)進(jìn)行拆分，變成 3 個(gè) 2-節(jié)點(diǎn)，最后將一個(gè) 2-節(jié)點(diǎn) 和 父節(jié)點(diǎn)合并，如下所示：
在一個(gè)父節(jié)點(diǎn)是 3-節(jié)點(diǎn)，該節(jié)點(diǎn)是3-節(jié)點(diǎn)的樹上添加元素。為了保證平衡，我們需要將該節(jié)點(diǎn)做局部變化，操作如下：首先將該節(jié)點(diǎn)臨時(shí)增加一個(gè)值變成 4-節(jié)點(diǎn)，然后對(duì)四節(jié)點(diǎn)進(jìn)行拆分，變成 3 個(gè) 2-節(jié)點(diǎn)，最后將一個(gè) 2-節(jié)點(diǎn) 和 父節(jié)點(diǎn)合并然后遞歸的對(duì)父節(jié)點(diǎn)進(jìn)行操作，直到根節(jié)點(diǎn)或者父節(jié)點(diǎn)是 2-節(jié)點(diǎn)停止，如下所示：
刪除
參見 mengzhisuoliu 博客
由此可見， 2-3 樹是由下向上生長的，但是刪除操作需要對(duì)樹進(jìn)行從上和從下兩方面的判斷，相對(duì)來說，刪除非常費(fèi)時(shí)。
紅黑樹
紅黑樹是一種 2-3 平衡樹的實(shí)現(xiàn)。不用去定義特殊的新的數(shù)據(jù)結(jié)構(gòu)，只需要一些附加信息，就可以實(shí)現(xiàn) 2-3 樹的構(gòu)建。
在紅黑樹種，利用黑連接表示 2-3 樹的普通節(jié)點(diǎn)， 紅連接將兩個(gè) 2-節(jié)點(diǎn)連接夠成一個(gè)三節(jié)點(diǎn)。
一個(gè) 2-3 樹可以化成一個(gè)等效的紅黑樹，如下圖所示：
定義
- 紅連接均為左連接
- 沒有任何一個(gè)節(jié)點(diǎn)同時(shí)和兩條連接相連
- 該樹是黑色完美平衡的，即任意空連接到根節(jié)點(diǎn)的路徑上的黑連接數(shù)量相同。
我們定義節(jié)點(diǎn)上存在 color 屬性，代表的是指向該節(jié)點(diǎn)的連接是什么顏色。
紅黑樹基本的操作是旋轉(zhuǎn)，在一些實(shí)現(xiàn)中，某些操作可能會(huì)出現(xiàn)紅色右連接或者兩條連續(xù)的紅連接，我們定義左旋轉(zhuǎn)是將一條紅色的右連接旋轉(zhuǎn)得到一條左連接。右連接相反，如下圖所示：
左旋轉(zhuǎn)的偽代碼如下所示：
Node rotateLset(Node h) {
Node x = h.right;
h.right = x.left;
x.left = h;
x.color = h.color;
h.color = red;
}
只需更改他的 color 屬性為紅，并將他原來自身的 color 屬性賦值給右連接節(jié)點(diǎn)就行。
同理，右連接示意圖如下：
NOde rotateRight(Node h) {
node x= h.left;
h.left = x.right;
x.right = h;
x.color = h.color;
h.color = read;
}
對(duì)于每一個(gè)插入，都插入一個(gè)紅連接。
向單個(gè) 2-節(jié)點(diǎn)中插入新鍵。如果新鍵小于老鍵，則增加一個(gè)紅色左連接，否則增加一個(gè)紅色右連接并進(jìn)行左旋轉(zhuǎn)，兩種情況都能產(chǎn)生一個(gè)等效的3-連接，如下所示：
向樹底部的 2-節(jié)點(diǎn)中插入新鍵?？偸窃黾右粋€(gè)新的紅色連接，如果他的父節(jié)點(diǎn)是 2-節(jié)點(diǎn)，那么按照如上兩種方式調(diào)整節(jié)點(diǎn)就行。
向一棵雙鍵樹（3-節(jié)點(diǎn)）中插入新建，分為了三種子情況，第一種情況是新鍵最大，第二種情況是新鍵在中間，第三種情況是新鍵最小。
如下如所示：
對(duì)于一個(gè)節(jié)點(diǎn)和兩個(gè)紅色連接直接相聯(lián)，這種情況等效于一個(gè) 4-節(jié)點(diǎn)，當(dāng)將這兩個(gè)紅色連接變黑時(shí)，需要將父節(jié)點(diǎn)由黑變紅，因?yàn)檫@樣的變換會(huì)在父節(jié)點(diǎn)產(chǎn)生一個(gè) 3-節(jié)點(diǎn)，理由如下：
每產(chǎn)生一個(gè)紅色連接都會(huì)向上傳遞直到根節(jié)點(diǎn)。
具體實(shí)現(xiàn)
如下代碼展示了紅黑樹在 Map 中的存儲(chǔ)節(jié)點(diǎn)。
static final class Entry<K,V> implements Map.Entry<K,V> {
K key;
V value;
Entry<K,V> left;
Entry<K,V> right;
Entry<K,V> parent;
boolean color = BLACK;
Entry(K key, V value, Entry<K,V> parent) {
this.key = key;
this.value = value;
this.parent = parent;
}
public K getKey() {
return key;
}
public V getValue() {
return value;
}
public V setValue(V value) {
V oldValue = this.value;
this.value = value;
return oldValue;
}
public boolean equals(Object o) {
if (!(o instanceof Map.Entry))
return false;
Map.Entry<?,?> e = (Map.Entry<?,?>)o;
return valEquals(key,e.getKey()) && valEquals(value,e.getValue());
}
public int hashCode() {
int keyHash = (key==null ? 0 : key.hashCode());
int valueHash = (value==null ? 0 : value.hashCode());
return keyHash ^ valueHash;
}
public String toString() {
return key + "=" + value;
}
}
以 boolean color 存儲(chǔ)指向該節(jié)點(diǎn)的連接的顏色。
每當(dāng)增加一個(gè)元素后，調(diào)用 fixAfterInsect 函數(shù)對(duì)紅黑樹進(jìn)行修正，如下所示：
private void fixAfterInsertion(Entry<K,V> x) {
x.color = RED;
while (x != null && x != root && x.parent.color == RED) {
if (parentOf(x) == leftOf(parentOf(parentOf(x)))) {
Entry<K,V> y = rightOf(parentOf(parentOf(x)));
if (colorOf(y) == RED) {
setColor(parentOf(x), BLACK);
setColor(y, BLACK);
setColor(parentOf(parentOf(x)), RED);
x = parentOf(parentOf(x));
} else {
if (x == rightOf(parentOf(x))) {
x = parentOf(x);
rotateLeft(x);
}
setColor(parentOf(x), BLACK);
setColor(parentOf(parentOf(x)), RED);
rotateRight(parentOf(parentOf(x)));
}
} else {
Entry<K,V> y = leftOf(parentOf(parentOf(x)));
if (colorOf(y) == RED) {
setColor(parentOf(x), BLACK);
setColor(y, BLACK);
setColor(parentOf(parentOf(x)), RED);
x = parentOf(parentOf(x));
} else {
if (x == leftOf(parentOf(x))) {
x = parentOf(x);
rotateRight(x);
}
setColor(parentOf(x), BLACK);
setColor(parentOf(parentOf(x)), RED);
rotateLeft(parentOf(parentOf(x)));
}
}
}
root.color = BLACK;
}
ConcurrentHashMap
特性
特性
- 基于 hash map 實(shí)現(xiàn)
- 提供所有 map 接口
- 不允許以 null 作為鍵或者值
- 線程安全，檢索操作不需要鎖定整個(gè)表
- 檢索操作通常不會(huì)阻塞，所以有可能和修改操作重疊
- 迭代時(shí)的順序按照鍵值排列，即存儲(chǔ)順序是有限的（在這個(gè)基礎(chǔ)上，同時(shí)實(shí)現(xiàn)了
NavigableMap接口，用于查找某個(gè) key 之前的所有元素等操作）
ConcurrentHashMap 實(shí)現(xiàn)原理
在 JDK 1.7 及其之前的版本中，ConcurrentHashMap 使用 Segement 數(shù)據(jù)結(jié)構(gòu)作為上鎖的最小單元，每一個(gè) Segment 容納了一個(gè) table，結(jié)構(gòu)如圖所示（引用自 JOE-1992 博客）：
每當(dāng)進(jìn)行 get 操作時(shí)，定位到對(duì)應(yīng)的 segment，上鎖，并執(zhí)行 put 操作。
hash 操作
計(jì)算 hash 時(shí)，同樣使用了內(nèi)置的 hash 函數(shù)對(duì) hash 進(jìn)行再次求解。不過相比于 HashMap，ConcurrentHashMap 使用了single-word Wang/Jenkins hash 算法的變種。代碼如下：
private static int hash(int h) {
// Spread bits to regularize both segment and index locations,
// using variant of single-word Wang/Jenkins hash.
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}
note：Wang/Jenkins Hash 算法關(guān)鍵特性
- 雪崩性（更改輸入?yún)?shù)的任何一位，就將引起輸出有一半以上的位發(fā)生變化）
- 可逆性（input ==> hash ==> inverse_hash ==> input）
因此，使用 Wang/Jenkins Hash 更加能夠獲得沖突更小的 hash。
存儲(chǔ)
在每一個(gè) Segment 中，使用 HashEntry 作為存儲(chǔ)結(jié)構(gòu)，HashEntry 的定義如下：
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
//保證set對(duì)所有線程可見（volatile 語義）
final void setNext(HashEntry<K,V> n) {
UNSAFE.putOrderedObject(this, nextOffset, n);
}
static final sun.misc.Unsafe UNSAFE;
static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class k = HashEntry.class;
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
其中，UNSAFE 為 Java 直接訪問內(nèi)存的函數(shù)，objectFieldOffset 為獲得某一變量的偏移量， putOrderedObject(this, nextOffset, n) 是將 n 放在 this 偏移量 nextOffset 的位置，并且是一個(gè)具有 volatile 語義的修改，對(duì)所有的線程可見。
Segment 直接繼承 ReentrantLock，可以簡化鎖或者一些單獨(dú)的構(gòu)造器，使得其可以單獨(dú)的當(dāng)成一個(gè)鎖。結(jié)構(gòu)如下:
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
//真實(shí)的存儲(chǔ)結(jié)構(gòu)
transient volatile HashEntry<K,V>[] table;
//對(duì)一個(gè) segment 所有元素計(jì)數(shù)
transient int count;
transient int modCount;
//當(dāng) table 中包含的 HashEntry 元素的個(gè)數(shù)超過本變量值時(shí)，觸發(fā) table 的再散列
transient int threshold;
final float loadFactor;
//... method
}
整個(gè)類 維持了一個(gè) Segment 數(shù)組，和必要的信息，如下：
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
implements ConcurrentMap<K, V>, Serializable {
/**
* segments 的掩碼值
* key 的散列碼的高位用來選擇具體的 segment
*/
final int segmentMask;
/**
* segment 外偏移量，和segment 維持多少個(gè) table 有關(guān)
*/
final int segmentShift;
/**
* 由 Segment 對(duì)象組成的數(shù)組，每個(gè)都是一個(gè)特別的Hash Table
*/
final Segment<K,V>[] segments;
// 根據(jù) hash 找到對(duì)應(yīng) segment
private Segment<K,V> segmentForHash(int h) {
// 重點(diǎn)在 (h >>> segmentShift) & segmentMask 函數(shù)
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
// 在 volatile 的環(huán)境下讀取 segments 的第 u 個(gè)元素
return (Segment<K,V>) UNSAFE.getObjectVolatile(segments, u);
}
}
在更老的版本中，采用直接加鎖的方式修改數(shù)據(jù)，代碼如下：
// ConcurrentHashMap 方法
public V put(K key, V value) {
if (value == null) //ConcurrentHashMap 中不允許用 null 作為映射值
throw new NullPointerException();
int hash = hash(key.hashCode()); //計(jì)算鍵對(duì)應(yīng)的散列碼
//根據(jù)散列碼找到對(duì)應(yīng)的 Segment
return segmentFor(hash).put(key, hash, value, false);
}
// Segment 方法
V put(K key, int hash, V value, boolean onlyIfAbsent) {
lock(); //當(dāng)前的segment加鎖
try {
int c = count;
if (c++ > threshold) //如果超過再散列的閾值
rehash(); //執(zhí)行再散列，table 數(shù)組的長度將擴(kuò)充一倍
HashEntry<K,V>[] tab = table;
//把散列碼值與 table 數(shù)組的長度減 1 的值相“與”
//得到該散列碼對(duì)應(yīng)的 table 數(shù)組的下標(biāo)值
int index = hash & (tab.length - 1);
//找到散列碼對(duì)應(yīng)的具體的那個(gè)桶
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;
V oldValue;
if (e != null) { //如果鍵/值對(duì)以經(jīng)存在
oldValue = e.value;
if (!onlyIfAbsent)
e.value = value; // 設(shè)置 value 值
}
else { //鍵/值對(duì)不存在
oldValue = null;
++modCount; //添加新節(jié)點(diǎn)到鏈表中，modCont 要加 1
// 創(chuàng)建新節(jié)點(diǎn)，并添加到鏈表的頭部
tab[index] = new HashEntry<K,V>(key, hash, first, value);
count = c; //寫 count 變量
}
return oldValue;
} finally {
unlock(); //解鎖
}
}
// get
V get(Object key, int hash) {
if(count != 0) { // 首先讀 count 變量
HashEntry<K,V> e = getFirst(hash);
while(e != null) {
if(e.hash == hash && key.equals(e.key)) {
V v = e.value;
if(v != null)
return v;
// 如果讀到 value 域?yàn)?null，說明發(fā)生了重排序，加鎖后重新讀取
return readValueUnderLock(e);
}
e = e.next;
}
}
return null;
}
在如上的實(shí)現(xiàn)中，最重要的是 get 函數(shù)的 count = c 操作和 put 函數(shù)的 count != 0 操作。這時(shí)實(shí)現(xiàn)可見性的關(guān)鍵。
當(dāng)一個(gè)線程上鎖修改之后，會(huì)調(diào)用 count = c。 count 是 volatile 類型的變量，這時(shí)就會(huì)將修改之后的 map 對(duì)所有線程可見。
當(dāng)另一個(gè)線程讀取時(shí)，會(huì)調(diào)用 count != 0，這個(gè)操作保證了更新之后的值對(duì)當(dāng)前線程可見。
對(duì)于 1.7 版本，參見技術(shù)世界
對(duì)于讀操作，獲取Key所在的Segment時(shí)，需要保證可見性。具體實(shí)現(xiàn)上可以使用volatile關(guān)鍵字，也可使用鎖。但使用鎖開銷太大，而使用volatile時(shí)每次寫操作都會(huì)讓所有CPU內(nèi)緩存無效，也有一定開銷。ConcurrentHashMap使用如下方法保證可見性，取得最新的Segment。
Segment<K,V> s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)
獲取Segment中的HashEntry時(shí)也使用了類似方法
HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE)
獲取鎖時(shí)，并不直接使用lock來獲取，因?yàn)樵摲椒ǐ@取鎖失敗時(shí)會(huì)掛起。事實(shí)上，它使用了自旋鎖，如果tryLock獲取鎖失敗，說明鎖被其它線程占用，此時(shí)通過循環(huán)再次以tryLock的方式申請(qǐng)鎖。如果在循環(huán)過程中該Key所對(duì)應(yīng)的鏈表頭被修改，則重置retry次數(shù)。如果retry次數(shù)超過一定值，則使用lock方法申請(qǐng)鎖。
這里使用自旋鎖是因?yàn)樽孕i的效率比較高，但是它消耗CPU資源比較多，因此在自旋次數(shù)超過閾值時(shí)切換為互斥鎖。
為更好支持并發(fā)操作，ConcurrentHashMap會(huì)在不上鎖的前提逐個(gè)Segment計(jì)算3次size，如果某相鄰兩次計(jì)算獲取的所有Segment的更新次數(shù)（每個(gè)Segment都與HashMap一樣通過modCount跟蹤自己的修改次數(shù)，Segment每修改一次其modCount加一）相等，說明這兩次計(jì)算過程中無更新操作，則這兩次計(jì)算出的總size相等，可直接作為最終結(jié)果返回。如果這三次計(jì)算過程中Map有更新，則對(duì)所有Segment加鎖重新計(jì)算Size。該計(jì)算方法代碼如下
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;
}
對(duì)于 1.8 版本，參見技術(shù)世界 ，使用 CAS 直接完成任務(wù)。
對(duì)于put操作，如果Key對(duì)應(yīng)的數(shù)組元素為null，則通過CAS操作將其設(shè)置為當(dāng)前值。如果Key對(duì)應(yīng)的數(shù)組元素（也即鏈表表頭或者樹的根元素）不為null，則對(duì)該元素使用synchronized關(guān)鍵字申請(qǐng)鎖，然后進(jìn)行操作。如果該put操作使得當(dāng)前鏈表長度超過一定閾值，則將該鏈表轉(zhuǎn)換為樹，從而提高尋址效率。
對(duì)于讀操作，由于數(shù)組被volatile關(guān)鍵字修飾，因此不用擔(dān)心數(shù)組的可見性問題。同時(shí)每個(gè)元素是一個(gè)Node實(shí)例（Java 7中每個(gè)元素是一個(gè)HashEntry），它的Key值和hash值都由final修飾，不可變更，無須關(guān)心它們被修改后的可見性問題。而其Value及對(duì)下一個(gè)元素的引用由volatile修飾，可見性也有保障。
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
}
對(duì)于Key對(duì)應(yīng)的數(shù)組元素的可見性，由Unsafe的getObjectVolatile方法保證。
put、remove和get操作只需要關(guān)心一個(gè)Segment，而size操作需要遍歷所有的Segment才能算出整個(gè)Map的大小。一個(gè)簡單的方案是，先鎖住所有Sgment，計(jì)算完后再解鎖。但這樣做，在做size操作時(shí)，不僅無法對(duì)Map進(jìn)行寫操作，同時(shí)也無法進(jìn)行讀操作，不利于對(duì)Map的并行操作。
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);
}
size操作
put方法和remove方法都會(huì)通過addCount方法維護(hù)Map的size。size方法通過sumCount獲取由addCount方法維護(hù)的Map的size。
ConcurrentSkipListMap
使用跳表實(shí)現(xiàn)。
跳躍表原理參見漫畫算法：什么是跳躍表
