HashMap很美好，但线程不安全怎么办？ConcurrentHashMap告诉你答案！
写在开头
在
《耗时2天，写完HashMap》
这篇文章中，我们提到关于HashMap线程不安全的问题，主要存在如下3点风险：
风险1：
put的时候导致元素丢失；如两个线程同时put，且key值相同的情况下，后一个线程put操作覆盖了前一个线程的操作，导致前一个线程的元素丢失。
风险2：
put 和 get 并发时会导致 get 到 null；若一个线程的put操作触发了数组的扩容，这时另外一个线程去get，因为扩容的操作很耗时，这时有可能会卡死或者get到null。
风险3：
多线程下扩容会死循环；多线程下触发扩容时，因为前一个线程已经破坏了原有链表结构，后一个线程再去读取节点，进行链接的时候，很可能发生顺序错乱，从而形成一个环形链表，进而导致死循环。
Hashtable解决线程安全靠谱吗？
那我们怎么办呢？很多小伙伴可能第一时间想到了HashTable，因为它和HashMap拥有者相似的功能，底层也是基于哈希表实现，数组+链表构建，数组容量到达阈值后，同样会自动扩容，Hashtable 默认的初始大小为 11，之后每次扩充，容量变为原来的 2n+1。并且，
Hashtable内部的方法几乎都是synchronized关键字修饰，保证了线程的安全
。
哇！这样一看，Hashtable简直是解决HashMap线程不安全的天选之子啊！但事实上，因为性能的问题，Hashtable已经在被废弃的边缘了，非常不建议在代码中使用它，原因如下接着往下看。
我们先写一个小小的测试类，来感受一下Hashtable的使用。
【代码示例1】
public class Test {
public static void main(String[] args) {
HashMap<Integer, String> map = new HashMap<>();
map.put(1, "I");
map.put(2, "love");
map.put(3, "Java");
Hashtable<Integer, String> hashtable = new Hashtable<>();
hashtable.put(1, "JavaBuild");
for (Map.Entry<Integer, String> entry : hashtable.entrySet()) {
System.out.println(entry.getKey()+":"+entry.getValue());
}
}
}
输出：
1:JavaBuild
然后，我们跟入到put中的原来，去看看它的底层实现
【源码解析1】
public synchronized V put(K key, V value) {
// Make sure the value is not null
if (value == null) {
throw new NullPointerException();
}
// Makes sure the key is not already in the hashtable.
Entry<?,?> tab[] = table;
int hash = key.hashCode();
int index = (hash & 0x7FFFFFFF) % tab.length;
@SuppressWarnings("unchecked")
Entry<K,V> entry = (Entry<K,V>)tab[index];
for(; entry != null ; entry = entry.next) {
if ((entry.hash == hash) && entry.key.equals(key)) {
V old = entry.value;
entry.value = value;
return old;
}
}
addEntry(hash, key, value, index);
return null;
}
通过这段源码我们能够发现
1、Hashtable哈希值的计算，并没有像HashMap那样重新计算，而是直接取key的hashCode()方法，这样一来它的扰动次数明显降低，hash的重合度更高；
2，index的位置计算中，Hashtable采用了%取余运算，而HashMap采用的是&运算，我们知道位运算直接对内存数据进行操作，不需要转成十进制，处理速度非常快，相比之下Hashtable的效率低下。
3，底层大部分的方法都是synchronized修饰，我们知道用synchronized 来保证线程安全的效率非常低下。当一个线程访问同步方法时，其他线程也访问同步方法，可能会进入阻塞或轮询状态，如使用 put 添加元素，另一个线程不能使用 put 添加元素，也不能使用 get，竞争会越来越激烈效率越低。
以上3点足以让我们头也不回的舍弃Hashtable，那么问题来了，除了这个集合类外，我们还有什么选项呢？这时，
ConcurrentHashMap
高高的举起了它的小手！
ConcurrentHashMap
文章写到这些，终于引出了我们今天的主角，ConcurrentHashMap！作为一个效率又高，又能保证线程安全的集合类，它的使用频率非常之高，话不多说，我们先来画一个底层逻辑实现图感受一下它的魅力！
JDK1.8下的ConcurrentHashMap底层实现
哦，对了，虽然我们现在主流的Java版本都是1.8+了，但很多公司在面试的时候，提及ConcurrentHashMap时，有时候还是会问到1.7的底层实现，因此，学有余力的小伙伴，私下里把JDK1.7的底层源码也拿过来读读哈（build哥本地没有安装JDK1.7，就不贴源码解析了）。
JDK1.8中ConcurrentHashMap抛弃了原有的 Segment 分段锁，采用了 CAS + synchronized 来保证并发安全性，底层结构采用Node数组+链表/红黑树，当链表长度达到一定长度后，会转为红黑树，这和HashMap一样。
【PUT源码解析】
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
// key 和 value 不能为空
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
// f = 目标位置元素
Node<K,V> f; int n, i, fh;// fh 后面存放目标位置的元素 hash 值
if (tab == null || (n = tab.length) == 0)
// 数组桶为空，初始化数组桶（自旋+CAS)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 桶内为空，CAS 放入，不加锁，成功了就直接 break 跳出
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;
// 使用 synchronized 加锁加入节点
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;
}
}
}
addCount(1L, binCount);
return null;
}
源码有点长，大致做了如下几点：
- 先根据 key 计算出 hashcode；
- 判断数组桶是否为空，若为空则通过tab = initTable()，初始化数组桶（自旋+CAS)；
- 计算出key的数组桶位置后，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功；
- 如果当前位置的 “hashcode == MOVED == -1”,则需要进行扩容；
- 如果都不满足，则利用 synchronized 锁写入数据；
- 如果数量大于 TREEIFY_THRESHOLD 则要执行树化方法，在 treeifyBin 中会首先判断当前数组长度 ≥64 时才会将链表转换为红黑树。
【源码扩展1】
上面put的时候，若Node数组桶为空时，需要进行初始化，那么我们跟入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 ,说明另外的线程执行CAS 成功，正在进行初始化。
if ((sc = sizeCtl) < 0)
// 让出 CPU 使用权
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;
}
从源码中我们可以看到，它的初始化是通过CAS和自旋完成的，注意其中的
sizeCtl
私有成员变量，当它的值小于0（准确来说等于-1）时，说明另外的线程执行CAS 成功，正在进行初始化。通过Thread.yield()做线程让步动作，让出CPU的使用权，自旋等待，随着获得资源，进入CAS。
知识点补充
CAS
（compare and swap) 译为：比较与交换
// 如果在这个位置(address) 的值等于 这个值(expectedValue),那么交换(newValue)。
boolean CAS(address,expectedValue,newValue) {
if(address 的 value == expectedValue) {
address 的 value = newValue;
return true;
}
}
自旋：
所谓的自旋，旨在线程抢锁失败后进入阻塞状态，放弃 CPU，需要过很久才能再次被调度。但经过测算，大部分情况下，虽然当前抢锁失败，但过不了很久，锁就会被释放。因此，当某个线程抢占 CPU 失败后，保持就绪状态，一旦锁释放，就会继续抢占。
以上这2点内容，在后面的并发多线程中会着重学习，在这里浅浅点名，让大家明白他们的意思和作用即可。
【源码扩展2】
当链表的长度大于8时，会转为红黑树，而红黑树的实现，是通过底层的TreeBin，我们跟进去看一下。
static final class TreeBin<K,V> extends Node<K,V> {
TreeNode<K,V> root;
volatile TreeNode<K,V> first;
volatile Thread waiter;
volatile int lockState;
// values for lockState
static final int WRITER = 1; // set while holding write lock
static final int WAITER = 2; // set when waiting for write lock
static final int READER = 4; // increment value for setting read lock
...
}
TreeBin通过root属性维护红黑树的根结点，因为红黑树在旋转的时候，根结点可能会被它原来的子节点替换掉，在这个时间点，如果有其他线程要写这棵红黑树就会发生线程不安全问题，所以在 ConcurrentHashMap 中TreeBin通过waiter属性维护当前使用这棵红黑树的线程，来防止其他线程的进入。
【Get源码解析】
与put相比，get的源码就简单太多了，大概进行了如下几步操作：
1，根据计算出来的 hash 值寻址，如果在桶上直接返回值；
2，如果是红黑树，按照树的方式获取值；
3，如果是链表，按链表的方式遍历获取值；
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// key 所在的 hash 位置
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 如果指定位置元素存在，头结点hash值相同
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
// key hash 值相等，key值相同，直接返回元素 value
return e.val;
}
else if (eh < 0)
// 头结点hash值小于0，说明正在扩容或者是红黑树，find查找
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;
}
总结
文章写到这里，ConcurrentHashMap的介绍基本讲完了，我们现在来自我总结一下为啥它的效率又高，又能保证线程安全。
以JDK1.8版本阐述：
- Node + CAS + synchronized 保证并发安全，每次上锁的颗粒度细到链表或红黑树的根节点，不会影响其他Node的读写，此外CAS是轻量级的，synchronized 也经过了锁升级；
- JDK1.7的版本里采用的Segment 分段锁，颗粒度粗不说，Segment 的个数一旦初始化就不能改变。 Segment 数组的大小默认是 16，也就是说默认可以同时支持 16 个线程并发写。而1.8的版本中，Node是一个数组，初始默认为16，后续仍然可以以2的幂次方级别进行扩容，因此，它所支持的并发量要看它数组的真实容量；
- 效率高是因为它底层采用了和JDK1.8中HashMap相同的数组+链表/红黑树结构。
结尾彩蛋
如果本篇博客对您有一定的帮助，大家记得
留言+点赞+收藏
呀。原创不易，转载请联系Build哥！
如果您想与Build哥的关系更近一步，还可以关注俺滴公众号“JavaBuild888”，在这里除了看到《Java成长计划》系列博文，还有提升工作效率的小笔记、读书心得、大厂面经、人生感悟等等，欢迎您的加入！
