ReZero's Utopia.

Java HashMap & Concurrent Hash Map

Word count: 1.8kReading time: 7 min
2020/08/09 Share
  1. 为啥拉链用尾插不用头插?

    • jdk 1.7 用的头插,但头插的问题就是resize:因为hash取值后放置是从后往前(头插,逆序链表方向)的,但是在resize时会重计算hash,此时是顺序遍历方向,那么就可能出现一开始A头插在B前,rehash时B
      后进行rehash导致B
      头插在A前。这样AB的顺序resize后就不同了。这样就会使得多线程时对引用的指向调整有问题,有可能造成循环链表了。尾插法的顺序和rehash时遍历的顺序一致自然就不会有这个问题。
  2. Collections.synchronizedMap

    内部维护了一个 map 用来引用需要改造的map,通过加 mutex 排他锁的形式来对map进行各种操作。其实就是全部用 synchronized 包裹起来。

  3. fail-safe & fail-fast

    hashtable 使用的是 fail-safe 机制,该机制会在进入遍历前,记录当前集合的内容,然后遍历记录的集合内容,不关心集合在此期间的意外修改情况。类似得还有 copyOnWriteList

    fail-fast: 每次集合变动会使得 modCount 跟着变动,而使用 next 遍历前会检查这个 modCount 是否是期望值。

Concurrent Hash Map

  1. 1.7 版本的分段方式 segment
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
// 先定位 segment,对 segment后进行put操作
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();//这就是为啥他不可以put null值的原因
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null)
s = ensureSegment(j);
return s.put(key, hash, value, false);
}


final V put(K key, int hash, V value, boolean onlyIfAbsent) {
// 将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry
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;
// 遍历该 HashEntry,如果不为空则判断传入的 key 和当前遍历的 key 是否相等,相等则覆盖旧的 value。
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 {
// 不为空则需要新建一个 HashEntry 并加入到 Segment 中,同时会先判断是否需要扩容。
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;
}
  1. 1.8 版本的 CAS + synchronized
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
final V putVal(K key, V value, boolean onlyIfAbsent) {
//可以看到,在并发情况下,key 和 value 都是不支持为空的。
if (key == null || value == null) throw new NullPointerException();
//这里和1.8 HashMap 的hash 方法大同小异,只是多了一个操作,如下
//( h ^ (h >>> 16)) & HASH_BITS; HASH_BITS = 0x7fffffff;
// 0x7fffffff ,二进制为 0111 1111 1111 1111 1111 1111 1111 1111 。
//所以,hash值除了做了高低位异或运算,还多了一步,保证最高位的 1 个 bit 位总是0。
//这里,我并没有明白它的意图,仅仅是保证计算出来的hash值不超过 Integer 最大值,且不为负数吗。
//同 HashMap 的hash 方法对比一下,会发现连源码注释都是相同的,并没有多说明其它的。
//我个人认为意义不大,因为最后 hash 是为了和 capacity -1 做与运算,而 capacity 最大值为 1<<30,
//即 0100 0000 0000 0000 0000 0000 0000 0000 ,减1为 0011 1111 1111 1111 1111 1111 1111 1111。
//即使 hash 最高位为 1(无所谓0),也不影响最后的结果,最高位也总会是0.
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();
//若表已经初始化,则找到当前 key 所在的桶,并且判断是否为空
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//若当前桶为空,则通过 CAS 原子操作,把新节点插入到此位置,
//这保证了只有一个线程可以 CAS 成功,其它线程都会失败。
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//若所在桶不为空,则判断节点的 hash 值是否为 MOVED(值是-1)
else if ((fh = f.hash) == MOVED)
//若为-1,说明当前数组正在进行扩容,则需要当前线程帮忙迁移数据
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//这里用加同步锁的方式,来保证线程安全,给桶中第一个节点对象加锁
synchronized (f) {
//recheck 一下,保证当前桶的第一个节点无变化,后边很多这样类似的操作,不再赘述
if (tabAt(tab, i) == f) {
//如果hash值大于等于0,说明是正常的链表结构
if (fh >= 0) {
binCount = 1;
//从头结点开始遍历,每遍历一次,binCount计数加1
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果找到了和当前 key 相同的节点,则用新值替换旧值
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;
}
}
}
//否则判断是否是树节点。这里提一下,TreeBin只是头结点对TreeNode的再封装
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;
}
}
}
}
//注意下,这个判断是在同步锁外部,因为 treeifyBin内部也有同步锁,并不影响
if (binCount != 0) {
//如果节点个数大于等于 8,则转化为红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
//把旧节点值返回
if (oldVal != null)
return oldVal;
break;
}
}
}
//给元素个数加 1,并有可能会触发扩容,比较复杂,稍后细讲
addCount(1L, binCount);
return null;
}

红黑树

定义
  1. 根节点叶子节点都是黑色 且 叶子都是 NIL 节点

  2. 路径上不能有两个连续的红色节点

  3. 从任一节点到其每个叶子的所有路径都包含相同数目的黑色节点

带来的效益:最长路径不会超过最短路径的两倍

变色 & 旋转

基础:红黑树的插入节点都是红色的

  1. 变色

    • 红变黑或者黑变红,变化过程可能会发生连锁反应
  2. 旋转

    • 逆时针旋转,即父节点抢了右子节点的左儿子作为自己的右儿子,然后自己去认右子节点为爹。如图

    • 顺时针旋转,上面的左右互换即可。即父节点抢了左子节点的右儿子作为自己的左儿子,然后自己去认右子节点为爹。

  3. 应用

CATALOG
  1. 1. Concurrent Hash Map
    1. 1.1. 红黑树
      1. 1.1.1. 定义
      2. 1.1.2. 变色 & 旋转