HashMap很美好，但线程不安全怎么办？ConcurrentHashMap告诉你答案！

本文主要是介绍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 nullif (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 spinelse 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 lockStatestatic final int WRITER = 1; // set while holding write lockstatic final int WAITER = 2; // set when waiting for write lockstatic 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值相同，直接返回元素 valuereturn 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版本阐述：

1. Node + CAS + synchronized 保证并发安全，每次上锁的颗粒度细到链表或红黑树的根节点，不会影响其他Node的读写，此外CAS是轻量级的，synchronized 也经过了锁升级；
2. JDK1.7的版本里采用的Segment 分段锁，颗粒度粗不说，Segment 的个数一旦初始化就不能改变。 Segment 数组的大小默认是 16，也就是说默认可以同时支持 16 个线程并发写。而1.8的版本中，Node是一个数组，初始默认为16，后续仍然可以以2的幂次方级别进行扩容，因此，它所支持的并发量要看它数组的真实容量；
3. 效率高是因为它底层采用了和JDK1.8中HashMap相同的数组+链表/红黑树结构。

结尾彩蛋

如果本篇博客对您有一定的帮助，大家记得留言+点赞+收藏呀。原创不易，转载请联系Build哥！

如果您想与Build哥的关系更近一步，还可以关注俺滴公众号“JavaBuild888”，在这里除了看到《Java成长计划》系列博文，还有提升工作效率的小笔记、读书心得、大厂面经、人生感悟等等，欢迎您的加入！

这篇关于HashMap很美好，但线程不安全怎么办？ConcurrentHashMap告诉你答案！的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---