ConcurrentHashMap底层实现原理
ConcurrentHashMap
ConcurrentHashMap最早出现在 JDK 1.5中。底层基于散列算法实现，它是一个key-value结构的容器，使用Hash算法来获取值的地址，时间复杂度是O(1)。查询非常快。
- 是一个key-value的映射容器，key不重复
- jdk8中的ConcurrentHashMap基于数组+链表+红黑树实现
- 不保证键值的顺序
- key、value都不可以存入null值
- 线程安全。
ConcurrentHashMap并非锁住整个方法，而是通过原子操作和局部加锁的方法保证了多线程的线程安全，且尽可能减少了性能损耗。
- 继承了AbstractMap，实现了ConcurrentMap接口，提供了key,value结构格式访问的方法
- 实现了Serializable接口，表示HashMap支持序列化
ConcurrentHashMap的整体架构
ConcurrentHashMap在JDK1.8中的存储结构是由数组、单向链表、红黑树组成。
当我们初始化一个ConcurrentHashMap实例时，默认会初始化一个长度为16的数组。由于ConcurrentHashMap它的核心仍然是hash表，所以必然会存在hash冲突问题。
ConcurrentHashMap采用链式寻址法来解决hash冲突。
当hash冲突比较多的时候，会造成链表长度较长，这种情况会使得ConcurrentHashMap中数据元素的查询复杂度变成O(~n~)。因此在JDK1.8中，引入了红黑树的机制。
当数组长度大于64并且链表长度大于等于8的时候，单项链表就会转换为红黑树。
另外，随着ConcurrentHashMap的动态扩容，一旦链表长度小于8，红黑树会退化成单向链表。
ConcurrentHashMap的基本功能
ConcurrentHashMap本质上是一个HashMap，因此功能和HashMap一样，但是ConcurrentHashMap在HashMap的基础上，提供了并发安全的实现。
并发安全的主要实现是通过对指定的Node节点加锁，来保证数据更新的安全性。
ConcurrentHashMap在性能方面的优化
如果在并发性能和数据安全性之间做好平衡，在很多地方都有类似的设计，比如cpu的三级缓存、mysql的buffer_pool、Synchronized的锁升级等等。
ConcurrentHashMap也做了类似的优化，主要体现在以下几个方面：
- 在JDK1.8中，ConcurrentHashMap锁的粒度是数组中的某一个节点，而在JDK1.7，锁定的是Segment，锁的范围要更大，因此性能上会更低。
- 引入红黑树，降低了数据查询的时间复杂度，红黑树的时间复杂度是O(logn)。
当数组长度不够时，ConcurrentHashMap需要对数组进行扩容，在扩容的实现上，ConcurrentHashMap引入了多线程并发扩容的机制，简单来说就是多个线程对原始数组进行分片后，每个线程负责一个分片的数据迁移，从而提升了扩容过程中数据迁移的效率。
ConcurrentHashMap中有一个size()方法来获取总的元素个数，而在多线程并发场景中，在保证原子性的前提下来实现元素个数的累加，性能是非常低的。ConcurrentHashMap在这个方面的优化主要体现在两个点：
- 当线程竞争不激烈时，直接采用CAS来实现元素个数的原子递增。
- 如果线程竞争激烈，使用一个数组来维护元素个数，如果要增加总的元素个数，则直接从数组中随机选择一个，再通过CAS实现原子递增。它的核心思想是引入了数组来实现对并发更新的负载。
源码剖析
-
成员变量
private static final int MAXIMUM_CAPACITY = 1 << 30; //最大容量。因为32位哈希字段的前两位用于控制，所以长度为2的30次方
private static final int DEFAULT_CAPACITY = 16; //默认初始容量，一定是2的幂 (即至少1个)，最多MAXIMUM_CAPACITY
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; //最大数组大小。 需要toArray和相关方法
private static final int DEFAULT_CONCURRENCY_LEVEL = 16; //默认并发级别
private static final float LOAD_FACTOR = 0.75f; //负载因子。n - (n >>> 2)
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; //每个传输步骤的最小重新绑定数。 范围被细分以允许多个调整大小线程。 此值用作下限，以避免调整大小器遇到过多的内存争用。 该值至少为DEFAULT_CAPACITY。
private static int RESIZE_STAMP_BITS = 16; //在sizeCtl中用于生成戳记的位数。 对于32位数组，必须至少为6。
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; //可以帮助调整大小的最大线程数。 必须符合32 - RESIZE_STAMP_BITS位。
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; //在sizeCtl中记录尺寸戳的位移位
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
static final int NCPU = Runtime.getRuntime().availableProcessors(); //cpu的数量，以设置某些大小的界限
-
put方法
public V put(@NotNull K key, @NotNull V value) {
return putVal(key, value, false);
}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) {
// Cas操作
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) {
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;
}锁Node。
-
casTabAt方法
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}调用sun.misc包Unsafe类的CAS方法compareAndSwapObject。
CAS(Compare-And-Swap)：比较并交换。通过一个原子操作，用预期值去和实际值做对比，如果实际值和预期相同，则做更新操作。如果预期值和实际不同，就认为其他线程更新了这个值，不做更新操作。
总结
- 做插入操作时，首先进入乐观锁，
- 然后，在乐观锁中判断容器是否初始化，
- 如果没初始化则初始化容器，
- 如果已经初始化，则判断该hash位置的节点是否为空，如果为空，则通过CAS操作进行插入。
- 如果该节点不为空，再判断容器是否在扩容中，如果在扩容，则帮助其扩容。
- 如果没有扩容，则进行最后一步，先加锁，然后找到hash值相同的那个节点(hash冲突)，
- 循环判断这个节点上的链表，决定做覆盖操作还是插入操作。
- 循环结束，插入完毕。