1.HashMap
HashMap的ReHash在线程并发情况下会形成链表环。想要避免HashMap的线程安全问题有很多办法，比如改用HashTable或者Collections.synchronizedMap.但是，这两者有着共同的性能问题，无论是读操作还是写操作，他们都会给整个集合加锁，导致同一时间的操作为之阻塞。所以在高并发场景下，我们通常采用另一个集合类ConcurrentHashMap。
2.Java 1.7ConcurrentHashMap
2.1ConcurrentHashMap结构
image
ConcurrentHashMap中有一个Segment数组，而每一个Segment数组包含一个HashEntry数组，数组中的每一个HashEntry既是一个键值对，也是一个链表的头节点。可以说，ConcurrentHashMap是一个二级哈希表。
2.2 ConcurrentHashMap的优势
就是采用了锁分段技术，每个Segment都有一把锁。不同Segment互不影响。不同Segment可以并发写入，同一Segment可以一写一读，而同一Segment不能并发写入。
2.3如何统计ConcurrentHashMap的Size呢？
先是去尝试总的修改次数是否大于上一次的修改次数。如果大于，说明统计过程中有修改。则尝试重新统计。尝试次数到达一定阈值后，对每一个Segment加锁，再重新统计。这里体现了乐观锁悲观锁的思想。先是用了乐观锁的思想，假设统计过程中没有修改。后来用了悲观锁的思想，假设统计过程中肯定有修改，对每一个Segment加锁。
3.Java 1.8ConcurrentHashMap
3.1table初始化
table初始化操作会延缓到第一次put行为。但是put是可以并发执行的，那么是如何实现table只初始化一次的？让我们来看看源码的实现。
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
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;
}
sizeCtl默认为0，如果ConcurrentHashMap实例化时有传参数，sizeCtl会是一个2的幂次方的值。所以执行第一次put操作的线程会执行Unsafe.compareAndSwapInt方法修改sizeCtl为-1，有且只有一个线程能够修改成功，其它线程通过Thread.yield()让出CPU时间片等待table初始化完成。
3.2put操作
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();
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);
...省略部分代码
}
addCount(1L, binCount);
return null;
}
1.计算hash值，在table中定位索引的位置。获取table中对应索引的元素f。采用Unsafe.getObjectVolatile来获取。 在java内存模型中，每个线程都有一个工作内存，里面存储着table的副本，虽然table是volatile修饰的，但不能保证线程每次都拿到table中的最新元素，Unsafe.getObjectVolatile可以直接获取指定内存的数据，保证了每次拿到数据都是最新的。
2.如果f为null，说明table中这个位置第一次插入元素，利用Unsafe.compareAndSwapObject方法插入Node节点。
如果CAS成功，说明Node节点已经插入，随后addCount(1L, binCount)方法会检查当前容量是否需要进行扩容。
如果CAS失败，说明有其它线程提前插入了节点，自旋重新尝试在这个位置插入节点。
3.如果f的hash值为-1，说明当前f是ForwardingNode节点，意味有其它线程正在扩容，则一起进行扩容操作。
4.其余情况把新的Node节点按链表或红黑树的方式插入到合适的位置，这个过程采用同步内置锁实现并发，代码如下:
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;
}
}
}
}
在节点f上进行同步，节点插入之前，再次利用tabAt(tab, i) == f判断，防止被其它线程修改。
如果f.hash >= 0，说明f是链表结构的头结点，遍历链表，如果找到对应的node节点，则修改value，否则在链表尾部加入节点。
如果f是TreeBin类型节点，说明f是红黑树根节点，则在树结构上遍历元素，更新或增加节点。
如果链表中节点数binCount >= TREEIFY_THRESHOLD(默认是8)，则把链表转化为红黑树结构。
3.2get操作
分为两步
1.构建一个nextTable，大小为table的两倍。（通过Unsafe.compareAndSwapInt修改sizeCtl值，保证只有一个线程能够初始化nextTable）
2.把table的数据复制到nextTable中。
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
计算key的hash值，并获取指定table中指定位置的Node节点，通过遍历链表或则树结构找到对应的节点，返回value值。
网友评论