上期我们提到了HashMap是线程不安全的，因此在并发环境下HashMap会带来问题。代码如下：
public static void main(String[] args) {
HashMap<Integer, Integer> map = new HashMap<>();
//线程一
new Thread(() -> {
for (int i = 0; i < 30; i++) {
map.put(i, i);
System.out.println(map.toString());
}
}).start();
//线程二
new Thread(() -> {
for (int i = 0; i < 30; i++) {
map.put(i, i);
System.out.println(map.toString());
}
}).start();
//线程三
new Thread(() -> {
for (int i = 0; i < 30; i++) {
map.put(i, i);
System.out.println(map.toString());
}
}).start();
}
我们创建三个线程来同时向HashMap中添加元素，就会产生问题，抛出java.util.ConcurrentModificationException的异常。
Exception in thread "Thread-1" Exception in thread "Thread-2" java.util.ConcurrentModificationException
at java.util.HashMap$HashIterator.nextNode(HashMap.java:1445)
at java.util.HashMap$EntryIterator.next(HashMap.java:1479)
at java.util.HashMap$EntryIterator.next(HashMap.java:1477)
at java.util.AbstractMap.toString(AbstractMap.java:554)
at com.study.map.HashMapDemo.lambda$main$1(HashMapDemo.java:24)
at java.lang.Thread.run(Thread.java:748)
那如何来解决这个问题呢？
①我们很容易想到，使用HashTable，因为HashTable是线程安全的，但HashTable在每个方法上都加了synchronized关键字，效率很低。
②使用Collections.synchronizedMap()方法，将HashMap转化为SynchronizedMap，其实还是添加了synchronized关键字，没什么本质区别。
③使用ConcurrentHashMap，它是JUC包下的一个类，用于处理HashMap多线程下引发的线程安全问题，并且性能很高。
下面我们就来深入学习ConcurrentHashMap，研究一下他为什么高效。
一、ConcurrentHashMap简介
在JDK1.7中，ConcurrentHashMap采用了Segment+HashEntry的数据结构来存储，大致结构是下边这个样子：
每一个ConcurrentHashMap都有一个segment数组，每个segment又包含了一个table数据，每一个结点的类型是HashEntry。每次put操作时会先寻找Segment，然后再寻找HashEntry，也就是两次寻找桶的过程。Segment部分代码如下：
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
/**
* put时候会尝试获取锁，如果获取失败肯定就有其他线程存在竞争，
* 则利用scanAndLockForPut()自旋获取锁,如果重试的次数达到了
* MAX_SCAN_RETRIES 则改为阻塞锁获取，保证能获取成功。
*/
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
//每一个segment都含有一个table,通过entryAt/setEntryAt访问
transient volatile HashEntry<K,V>[] table;
//元素个数
transient int count;
//修改的次数
transient int modCount;
//阈值，超过时会rehash
transient int threshold;
//负载因子
final float loadFactor;
}
由于Segment继承了ReentrantLock，所以在put操作时，会将当前Segment进行上锁，默认为16个segment。
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;
}
/**
* Sets next field with volatile write semantics. (See above
* about use of putOrderedObject.)
*/
final void setNext(HashEntry<K,V> n) {
UNSAFE.putOrderedObject(this, nextOffset, n);
}
// Unsafe mechanics
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);
}
}
}
而在JDK1.8之后，著名的Doug Lea大师可能觉得两次寻找桶的过程和锁住segment太慢了，于是做出了更加细粒度的上锁控制，采用了锁桶的方式（后续详细代码会有看到），并且摒弃了Segment（ReentrantLock）的方式，采用了Synchronized+CAS来控制并发。数据结构上采用了的数组+链表+红黑树，与HashMap相同。
二、ConcurrentHashMap的部分源码说明
由于现在主流的还是JDK1.8，所以还是基于JDK1.8来进行。源码如下：
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable {
private static final long serialVersionUID = 7249069246763182397L;
/**
* 最大容量
*/
private static final int MAXIMUM_CAPACITY = 1 << 30;
/**
* 默认容量
*/
private static final int DEFAULT_CAPACITY = 16;
/**
* 数组可能最大额容量，在toArray或者相关方法中使用
*/
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
/**
* 默认的并发级别，没有使用，是为了和之前的版本兼容
*/
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
/**
* 负载因子
*/
private static final float LOAD_FACTOR = 0.75f;
/**
* 树化阈值
*/
static final int TREEIFY_THRESHOLD = 8;
/**
* 链化阈值
*/
static final int UNTREEIFY_THRESHOLD = 6;
/**
* 最小树化桶个数
*/
static final int MIN_TREEIFY_CAPACITY = 64;
/**
* 每次进行转移的最小值
*/
private static final int MIN_TRANSFER_STRIDE = 16;
/**
* sizeCtl中记录扩容线程的位数，默认16位
*/
private static int RESIZE_STAMP_BITS = 16;
/**
* 进行扩容所允许的最大线程数
*/
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
/**
* sizeCtl中记录容量的位数，默认16位
*/
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
/*
* 一些HashCode值的含义
*/
static final int MOVED = -1; // hash for forwarding nodes
static final int TREEBIN = -2; // hash for roots of trees
static final int RESERVED = -3; // hash for transient reservations
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
/** CPU个数 */
static final int NCPU = Runtime.getRuntime().availableProcessors();
/**
* node数组，用于存储元素
*/
transient volatile Node<K,V>[] table;
/**
* 扩容后的下一个数组
*/
private transient volatile Node<K,V>[] nextTable;
/**
* 基本计数器值
*/
private transient volatile long baseCount;
/**
* 当sizeCrl为-1时，表明table数组正在被初始化。
* 当sizeCtl为其他负数时，其高16位代表此次扩容操作数组的一个标记
*（与table数组的大小有关），低16位代表(1+正在参与扩容的线程数)，
* 当sizeCtl为0时，表明此时table数组为null，使用默认的初始化大小，即16。
* 当sizeCtl为正数，且table数组为null时，sizeCtl代表table的初始化大小。
* 当sizeCtl为正数，且table数组不为null时，sizeCtl代表下一次扩容的扩容阈值
*/
private transient volatile int sizeCtl;
/**
* 扩容下另一个表的索引
*/
private transient volatile int transferIndex;
/**
* 自旋锁标记
* 用于保护初始化CounterCell、初始化CounterCell数组以及
* 对CounterCell数组进行扩容时的安全
*/
private transient volatile int cellsBusy;
/**
* 初始大小为2，每次扩容翻倍，存储CounterCell对象，
* 该对象有个value变量，用来存储个数
* 该数组的大小上限与当前机器的CPU数量有关，它不会被主动初始化，
* 只有在调用fullAddCount()函数时才会进行初始化.
*/
private transient volatile CounterCell[] counterCells;
// 一些视图
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
}
CounterCell内部类，用于记录元素个数。使用时是一个CounterCell数组，分片记录个数。
补充小知识，不想看可忽略：@sun.misc.Contended 是 Java 8 新增的一个注解，对某字段加上该注解则表示该字段会单独占用一个缓存行（Cache Line），不然会造成伪共享问题，比如当一个 CPU 要修改某共享变量 A 时会先锁定A 所在的缓存行，并且把其他 CPU 缓存上相关的缓存行设置为无效。但如果被锁定或失效的缓存行里，还存储了变量 B，其他线程此时就访问不了 B，或者由于缓存行失效需要重新从内存中读取加载到缓存里，这就造成了开销。所以让共享变量 A 单独使用一个缓存行就不会影响到其他线程的访问。
/**
* A padded cell for distributing counts. Adapted from LongAdder
* and Striped64. See their internal docs for explanation.
*/
@sun.misc.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
}
三、ConcurrentHashMap的一些重要方法
ConcurrentHashMap中几乎所有的操作都是基于sun.misc.Unsafe中原子操作实现多并发的无锁化操作（CAS）。
①put方法，添加元素
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
//这里说明了ConcurrentHashMap不支持null键和null值
if (key == null || value == null)
throw new NullPointerException();
//计算hash值
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
//f当前桶 n桶个数 i桶下标 fh桶哈希
Node<K,V> f; int n, i, fh;
//如果table为空，则对table进行初始化，
//这也是ConcurrentHashMap的懒加载
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//寻找桶，如果桶未初始化，通过CAS进行赋值
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);
//赋值
else {
V oldVal = null;
//对当前桶上锁
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
//记录桶中元素个数 超过8进行树化
binCount = 1;
//向链表c插入元素的过程
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;
}
}
}
//判断是否需要扩容，总容量加1
addCount(1L, binCount);
return null;
}
②initTable方法，对table进行一下初始化，还有一些字段的默认值
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//负数表示初始化或resize，这里只能是初始化
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) {
//sizeCtl此时为-1,走默认容量
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//设置为n大小的0.75倍，这就是位运算的魅力吗？
sc = n - (n >>> 2);
}
} finally {
//将sizeCtl设置为阈值
sizeCtl = sc;
}
break;
}
}
return tab;
}
③addCount方法，添加元素
private final void addCount(long x, int check) {
//b当前总大小，s添加后大小
CounterCell[] as; long b, s;
//CAS给baseCount + 1
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
//ThreadLocalRandom.getProbe() 得到当前线程的探针哈希值
if (as == null || (m = as.length - 1) < 0 ||
//寻找桶
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
//对CounterCell中的数目累加
s = sumCount();
}
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
//检查是否需要扩容
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
④fullAndCount()方法，用来初始化CounterCell，来记录元素个数，里面包含扩容，初始化等操作。当前CounterCell为null、长度-1小于0、CounterCell某个位置为null、给当前CounterCell进行CAS赋值产生冲突都会执行此方法。
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
//判断ThreadLocalRandom.getProbe()是否为0，为0则初始化，并
//设置未冲突标志位为true
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit(); // force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
//自旋锁
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
//判断counterCells是否被初始化过了
if ((as = counterCells) != null && (n = as.length) > 0) {
//找桶
if ((a = as[(n - 1) & h]) == null) {
//cellsBusy=0 没有线程对CounterCells进行操作
if (cellsBusy == 0) { // Try to attach new Cell
CounterCell r = new CounterCell(x); // Optimistic create
//通过cas设置cellsBusy标识，防止其他线程来对CounterCells并发处理
if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try { // Recheck under lock
CounterCell[] rs; int m, j;
if ((rs = counterCells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
//将初始化的r对象的元素赋值到对应的位置
rs[j] = r;
created = true;
}
} finally {
//恢复标记位
cellsBusy = 0;
}
if (created)
break;
//continue表示当前CounterCell不是空，自旋再处理
continue; // Slot is now non-empty
}
}
collide = false;
}
//CAS失败
else if (!wasUncontended) // CAS already known to fail
//设置为未冲突标识，进入下一次自旋
wasUncontended = true; // Continue after rehash
//由于指定下标位置的CounterCell值不为空，则直接通过cas进行累加，成功直接退出
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
break;
//如果已经有其他线程建立了新的CounterCells或者CounterCells大于CPU核心数
else if (counterCells != as || n >= NCPU)
collide = false; // At max size or stale
else if (!collide)
collide = true;
//进行扩容
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
try {
if (counterCells == as) {// Expand table unless stale
CounterCell[] rs = new CounterCell[n << 1];
for (int i = 0; i < n; ++i)
rs[i] = as[i];
counterCells = rs;
}
} finally {
cellsBusy = 0;
}
collide = false;
continue; // Retry with expanded table
}
h = ThreadLocalRandom.advanceProbe(h);
}
//进行初始化
else if (cellsBusy == 0 && counterCells == as &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean init = false;
try { // Initialize table
if (counterCells == as) {
//默认两个在这儿
CounterCell[] rs = new CounterCell[2];
rs[h & 1] = new CounterCell(x);
counterCells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
//CounterCells被占用，直接累加在baseCount变量中，防止一直自旋
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
break; // Fall back on using base
}
}
核心的方法就介绍这么多吧，剩下的大家有兴趣自己阅读即可，阅读源码的目的就是为了理解并学习他们的思路，以后可以运用到我们的代码当中。最后想说一下，阅读源码，大家请务必坚持，坚持就是胜利，Doug Lea yyds。
今天的分享就到此结束啦，喜欢的小伙伴记得点赞呦。
下期预告：ArrayList + CopyOnWriteList。
关注公众号JavaGrowUp，下期不迷路，获取更多精彩内容。
