1 前言
上一节我们讲解了HashMap底层的实现，但是HashMap不是线程安全的，存在多个线程并发操作的线程安全的问题。今天我们就来看一下ConcurrentHashMap这个数据结构，并发安全版本的Map结构。
2 ConcurrentHashMap内部源码
2.1 内部属性
// 最大容量上限，2的30次方 private static final int MAXIMUM_CAPACITY = 1 << 30; // 默认的初始化容量16 private static final int DEFAULT_CAPACITY = 16; // 最大的数组长度上限 static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; // 默认并发能力层级，16 // （默认数组长度是16，每个数组元素都可作为一把锁，所以数组长度是多少默认就有多少把锁） private static final int DEFAULT_CONCURRENCY_LEVEL = 16; // 负载因子 private static final float LOAD_FACTOR = 0.75f; // 红黑树转化条件（链表长度达到8需要转成红黑树，提高查询效率） static final int TREEIFY_THRESHOLD = 8; // 非树化条件（红黑树大小小于等于6时候，需要将红黑树转成链表） static final int UNTREEIFY_THRESHOLD = 6; // 红黑树化前最小容量（如果当前容量小于64，首先考虑扩容减少hash冲突，而不是进行树化） static final int MIN_TREEIFY_CAPACITY = 64; private static final int MIN_TRANSFER_STRIDE = 16; private static int RESIZE_STAMP_BITS = 16; private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; // 标志位，说明当前正在扩容操作，该位置数据正在移动，也就是move（移动）的意思 static final int MOVED = -1; // 标志位，说明改位置的数据正在进行树化 static final int TREEBIN = -2; // 保留位置 static final int RESERVED = -3; // 进行hash算法的时候，保留多少位hash结果，这里是保留int类型中的后31位 static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash // 当前机器的cpu核数 static final int NCPU = Runtime.getRuntime().availableProcessors(); // 保存数据的基础数组，使用volatile修饰，保证多线程之间的可见性 transient volatile Node<K,V>[] table; // 扩容期间的另外一个数组，ConcurrentHashMap扩容期间两个数组都在使用 private transient volatile Node<K,V>[] nextTable; // 用作大小统计的基础窗口 // ConcurrentHashMap使用多窗口分段锁机制来进行数量统计，提高并发性能 private transient volatile long baseCount; // 一个标志位，用作数组初始化或者扩容期间的标识 private transient volatile int sizeCtl; // 扩容期间，下一个需要进行数组迁移的数组下标 private transient volatile int transferIndex; // private transient volatile int cellsBusy; // 窗口列表，用多窗口机制来减少锁冲突，提高高并发的计数统计性能 private transient volatile CounterCell[] counterCells;
上面是ConcurrentHashMap内部属性，接下来我们看一下它的put方法，是怎么使用分段锁、cas操作来保证线程安全同时又提高了并发性能的。
2.2 put方法
public V put(K key, V value) { // 内部调用putVal方法 return putVal(key, value, false); }
2.2.1 putVal方法
final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); // 使用spread函数，得出一个新的hash值 // 跟上一章节讲的HashMap的hash函数核心思想一样 // 低16位结果变为高低16位的异或值 int hash = spread(key.hashCode()); int binCount = 0; // 使用一个无限循环，知道putVal操作成功才退出 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // 如果基础数组为空或者长度为零，说明还未初始化 if (tab == null || (n = tab.length) == 0) // 首先进行数组的初始化 tab = initTable(); // 走到这里说明table数组已经初始化了，肯定不为空 // 根据寻址算法 (n-1)&hash 找到数组的table[i]位置 // 如果table[i] == null说明该位置之前没存储元素 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 创建一个新的Node节点，使用cas操作将node设置到table[i]位置 // cas保证原子性，由于table使用volatile关键字修饰，保证可见性、有序性；所以是并发安全的 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) // cas成功则退出循环，否则继续 break; } // 如果f=table[i]的hash值被标记为MOVED，说明当前正在进行扩容 else if ((fh = f.hash) == MOVED) // 则当前线程帮助进行扩容（多线程扩容，缩短扩容时间） // 扩容完成之后再进行重试 tab = helpTransfer(tab, f); else { V oldVal = null; // 注意这里使用的synchronized进行加锁，加锁的元素是f=table[i] // 加锁的是数组的一个元素，通过寻址算法落到哪个位置，就对哪个位置进行加锁 // 保证并发安全的同时，加锁粒度很小，只要并发操作hash冲突不大，锁冲突就不大 // 这里就是分段锁的思想了，table数组的每个元素都可以作为一把锁 // 并发操作时，hash冲突时候才会存在锁竞争 synchronized (f) { // tabAt获取table[i]位置元素，二次确认此Node对象是否还是原来哪个，没被修改过 if (tabAt(tab, i) == f) { // fh=f.hash >= 0 说明未扩容或者红黑树化 // 在扩容或者初始化的时候会把table[i]所在元素的hash改为小于0 // 表示正在进行扩容、移动数据、进行树化等操作 if (fh >= 0) { binCount = 1; // 这里就跟之前HashMap一样，就是从头遍历链表 // 一个个查找节点是否Key与传入一样，如果一样则替换value值即可 // 如果不一样，则在链表尾部插入一个新的节点 // 由于使用了synchronized，是并发安全的 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; } } } // 如果f节点是红黑树 else if (f instanceof TreeBin) { Node<K,V> p; binCount = 2; // 对红黑树进行遍历，如果有节点Key与传入key一直，则替换value值即可 // 否则需要插入一个新的节点 if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { // 达到了树化条件，table[i]位置的链表转成红黑树 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // 这里使用并发多窗口分段统计的思想，进行计数统计 addCount(1L, binCount); return null; }
我们画个图理解一下：
（1）调用initTable方法初始化数组的时候，这个是能保证线程安全的，同一个时间只能有一个线程进行初始化，后面再给你分析一下
（2）helpTransfer方法是能进行多线程扩容的，这里保证并发安全的同时还能多线程协作扩容，提升效率
（3）使用加锁的粒度是f = table[i], synchronized(f)，只是路由到的哪个数组元素，就对那个元素进行加锁，加锁的粒度很小。
（4）操作成功之后，如果是新增元素，调用addCount方法使用多窗口分段锁思想，进行数量统计，这里也能减少所冲突，提升性能。
2.2.2 initTable方法
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; // 无限循环，直到成功才退出 while ((tab = table) == null || tab.length == 0) { // sizeCtl初始值是0，使用volatile修饰保证可见性、有序性 // 第一个进行初始化的线程会将sizeCtl使用cas操作设置为-1 // 后面第二、第三个线程发现sizeCtl < 0 发现已经有线程在进行初始化操作了 // 自己就不需要操作了，直接等待初始化完成即可 if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin // 走到这里，发现自己是第一个初始化的人，设置sizeCtl初始化标志位-1 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代表下一次扩容的阈值，默认是12 sc = n - (n >>> 2); } } finally { // 由于此时只有一个线程有操作权限，是线程安全的，直接修改即可 // 设置下次扩容的阈值为sc，也就是12 sizeCtl = sc; } break; } } return tab; }
上面的核心流程比较简单，就不画图了哈，主要包含以下步骤：
（1）第一个进行扩容的线程会将sizeCtl设置为-1，由于使用了volatile修饰，保证可见性
（2）后面进来的线程读取sizeCtl立即可见，发现为-1，说明有人正在初始化了，自己等待初始化完成即可
（3）初始化操作就是创建一个长度为DEFAULT_CAPACITY的数组出来
2.2.3 addCount方法
private final void addCount(long x, int check) { // as就是备用窗口列表 CounterCell[] as; long b, s; // 如果备用窗口列表不为空，或者在基础窗口竞争失败，则取备用窗口列表操作 if ((as = counterCells) != null || // 这里的意思是基础窗口竞争失败 !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true; // 这里就是从备用窗口列表选一个备用窗口，进行操作了 // 原理跟LongAdder基本一致的，这里就不详细分析了 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; // 这里就是将基础窗口的数量+所有备用窗口的数量，得到容量总和 s = sumCount(); } if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; // 这里判断容量 s > sizeCtl，也就是达到了扩容阈值，需要进行扩容操作 while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n); // 发现sizeCtl说明有线程正在进行扩容 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); } // 走到这里说明sizeCtl >= 0，说明自己是第一个执行扩容的线程 else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) // 真正进行扩容的方法 transfer(tab, null); s = sumCount(); } } }
执行过程大致是：
（1）首先如果是新增操作，需要将容量+1，容量增加为了提升并发的性能，使用多窗口统计的机制，将竞争压力分散到多个窗口中
（2）具体增加的操作是，首先竞争常规窗口，进行cas的自增1操作，如果失败则选择从备用窗口列表选择一个窗口，进行cas自增1操作
（3）这里如果多窗口机制理解有问题，可以回去看看LongAdder的文章，那里有详细的讲解
（4）自增完成后，使用sunCount方法得到当前的容量
（5）判断当前的容量是否达到扩容阈值，如果达到了则进行扩容
（6）扩容的时候可以多个线程进行，每个线程负责一部分数据迁移，有助于提高扩容效率
2.2.4 sumCount方法
final long sumCount() { // counterCells是备用窗口列表 CounterCell[] as = counterCells; CounterCell a; // baseCount就是基础窗口的数量 long sum = baseCount; // 这里就是将基础窗口数量 + 所有备用窗口的数量，就得到总数量了 if (as != null) { for (int i = 0; i < as.length; ++i) { if ((a = as[i]) != null) sum += a.value; } } return sum; }
这个流程很简单，跟之前看的LongAdder里面获得总数也是一样的，流程完全一样。
2.3 get方法
public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // 老规矩，计算一下hash码 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; } // 如果eh < 0,说明正在迁移到新数组，此时就是链表形式 else if (eh < 0) // 调用e.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; }
3 小结
ConcurrentHashMap暂时就先看这么多哈，有理解不对的地方欢迎指正哈。