ConcurrentHashMap 与阻塞队列：锁一个桶，而不是锁整张表

• HashMap 三次分别丢了 3.7 万、16 万、12.9 万个用户，没有任何异常。两个线程同时往同一个空桶放节点，后放的盖掉先放的；同时扩容，各自搬出一张新表，最后只留下其中一张（HashMap 页）。
• Collections.synchronizedMap 每个方法都锁整张表：不丢了，但所有线程的读和写排成一队，读也不能并行。
• 缺的那一块：让不同的桶各管各的。改的是不同的桶，就互不等待；读不加锁也能读到完整的数据。

不懂这一点时的典型误用：换成 ConcurrentHashMap 就觉得“线程安全了”，接着写 Integer v = map.get(k); map.put(k, v + 1);。实测 两个线程各 +1 共 200 万次，结果 1970799 和 1954211；换成 merge 才是 2000000（⑥ 坑 1）。

选择规则：要 Map 就用 ConcurrentHashMap，要有序再换跳表；要在线程之间交接任务、让“放的”和“取的”速度自动对齐，用阻塞队列。本页以 ConcurrentHashMap 为主，阻塞队列讲家族和选法，SynchronousQueue 和 LinkedTransferQueue 的内部见专门那一页。

④-1长什么样：和 HashMap 一样的桶数组，多了三种特殊节点

④-2put 走哪条路：点一个场景看看

• 1空桶不加锁：用 casTabAt 把新节点 CAS 进数组。两个线程抢同一个空桶，只有一个成功，失败的那个重新循环，这时桶已经不空了，就走加锁的路。
• 2非空桶锁桶头：synchronized (f)，f 是桶里第一个节点。锁到以后再确认一次桶头没被换掉（期间可能被扩容搬走），然后在链表或树里找 key。只有同一个桶的写才互相等，表越大、哈希越散，冲突越少。
• 3get 完全不加锁：tabAt 是一次带获取语义的读，节点的 val、next 是 volatile。写的一方先把节点构造好再发布，读的一方读到节点就看得到完整内容。
• 4扩容大家一起搬：一个线程发现要扩容，就建一张两倍长的新表，从后往前每次领 16 个桶（MIN_TRANSFER_STRIDE）去搬；别的线程 put 时撞上 ForwardingNode，先去领一段帮忙（helpTransfer），搬完再回来 put。

④-3size 和“只保证单个方法原子”

• 1size 是估算：size() 把 baseCount 和所有 CounterCell 加起来。加的过程中别的线程还在改，所以并发时读到的是“某个时刻附近”的值。实测 两个线程各 put 30 万个以后：baseCount 40954，counterCells 2 格合计 559046，加起来 600000。
• 2单个方法是原子的，两个方法之间不是：get 和 put 各自安全，但“get 完再 put”中间能插进别人。要“读出来改一改再写回”，用 merge、compute、computeIfAbsent，它们在桶锁里一次做完。
• 3不允许 null：key、value 都不能是 null。并发下 get 返回 null，必须能确定是“没有这个 key”；如果允许 null 值，就分不清，而又没法像单线程那样再调一次 containsKey 确认（两次调用之间可能变了）。

④-4阻塞队列：满了就等，空了就等

• 1“等”都是 AQS 的 Condition 加 park（AQS 页机制 5）：取的线程在 notEmpty 上等，放了东西就 signal；所以全景第 9 站的 order-canceller 平时是 WAITING，不占 CPU。
• 2DelayQueue 的 leader：只让一个线程按“堆顶还剩多久”定时睡（TIMED_WAITING），其余取的线程不限时等；堆顶到期，leader 取走，再叫醒下一个当 leader。避免一堆线程同时醒来抢同一个元素。

④-5为什么这样设计：锁的粒度跟着数据走

决定胜负的约束：真正会冲突的是“同一个桶”，不是“同一段”。哈希够散时，两个线程同时写同一个桶的概率很小，所以锁桶头几乎总是无竞争的；再配上 synchronized 在 JDK 6 以后的轻量级锁，比 Segment 里的 ReentrantLock 更省。

ConcurrentHashMap<Long, List<Long>> userOrders = new ConcurrentHashMap<>();
userOrders.computeIfAbsent(1024L, k -> new CopyOnWriteArrayList<>()).add(10017L);  // 没有就建，一步完成

ConcurrentHashMap<Long, Long> sold = new ConcurrentHashMap<>();
sold.merge(88L, 1L, Long::sum);                               // 销量 +1，原子
sold.compute(88L, (k, v) -> v == null ? 1 : v + 1);           // 同上，写法更灵活
long n = sold.mappingCount();                                 // 比 size() 好：返回 long

BlockingQueue<Long> notify = new ArrayBlockingQueue<>(10_000);  // 有界
if (!notify.offer(orderId, 50, TimeUnit.MILLISECONDS)) degrade();  // 满了：降级，不要无限等
DelayQueue<CancelTask> cancel = new DelayQueue<>();
CancelTask t = cancel.take();                                 // 到期才返回

初始容量的算法、computeIfAbsent 里的递归、弱一致的迭代器、队列选型的坑，这些放到第⑥步。

$ java -cp … MapRace
== 1. 两个线程各 put 50 万个不同的 key，应有 1000000 ==
第 1 次  HashMap 962723，218 ms   synchronizedMap 1000000，241 ms   ConcurrentHashMap 1000000，235 ms
第 2 次  HashMap 839013，147 ms   synchronizedMap 1000000，228 ms   ConcurrentHashMap 1000000，87 ms
第 3 次  HashMap 871315，172 ms   synchronizedMap 1000000，234 ms   ConcurrentHashMap 1000000，144 ms
== 2. 销量计数：两个线程各 +1 共 100 万次，应有 2000000 ==
第 1 次  get 再 put：1970799    merge：2000000
第 2 次  get 再 put：1954211    merge：2000000

$ java --add-opens java.base/java.util.concurrent=ALL-UNNAMED -cp … ChmPeek
== 1. 扩容：table 长度和 sizeCtl ==
刚 new：        table 还没建，sizeCtl=0
放了   1 个：  table 长 16，sizeCtl=12
放了  12 个：  table 长 32，sizeCtl=24        ← 第 12 个就扩了（HashMap 是第 13 个）
放了  24 个：  table 长 64，sizeCtl=48
放了  48 个：  table 长 128，sizeCtl=96
放了 100 个：  table 长 256，sizeCtl=192
== 2. 一个桶里挤满：链表 → 树 ==
同一个桶里 8 个：表长 128，桶 88 的头节点是 Node     ← new ConcurrentHashMap(64) 建出来是 128
同一个桶里 9 个：表长 128，桶 88 的头节点是 TreeBin
== 3. 计数器：两个线程同时 put ==
size()=600000，baseCount=40954，counterCells 2 格，格子里合计 559046
== 4. computeIfAbsent 里再改同一个 map ==
抛 IllegalStateException: Recursive update

$ java -cp … QueueDemo
== 1. ArrayBlockingQueue(3) ==
队列满了，offer(10004, 200ms) = false，等了 200 ms；add 会抛异常，put 会一直等
== 2. DelayQueue：到期才能取 ==
刚放进去，poll() = null（没到期，取不到）
第 1 个 take：订单 10018，在第 100 ms
第 2 个 take：订单 10019，在第 200 ms
第 3 个 take：订单 10017，在第 300 ms
== 3. 2 生产 2 消费，传 400 万个订单号 ==
第 1 轮  ArrayBlockingQueue 304 ms   LinkedBlockingQueue 650 ms
第 2 轮  ArrayBlockingQueue 385 ms   LinkedBlockingQueue 576 ms

• 1HashMap 三次都丢、每次丢得不一样、从不报错：这就是 J01 全景第 8 站预言的事。ConcurrentHashMap 不丢，三次都比整表加锁的 synchronizedMap 快（第 1 次 235 对 241 ms，只快一点）。
• 2ConcurrentHashMap 在数量达到门槛（12）时就扩容，HashMap 是超过门槛（第 13 个）才扩。new ConcurrentHashMap(64) 在 JDK 17 里按 64 ÷ 0.75 + 1 = 86 再向上取 2 的幂，建出 128 长的表；new HashMap(64) 是 64（⑥ 坑 5）。
• 3两个线程同时 put，大部分计数落进了 counterCells，只有约 7% 加在 baseCount 上：一冲突就分格，和 LongAdder 一模一样。
• 4这一轮 ArrayBlockingQueue 比 LinkedBlockingQueue 快。按设计 LinkedBlockingQueue 放和取用两把锁、能并行，但每放一个都要 new 一个链表节点；2 核、元素很小时，分配和 GC 的开销盖过了两把锁的好处。普通 main 程序，只看方向。

• ConcurrentHashMap 的 transfer 源码逐行：sizeCtl 的高低位、resizeStamp、最后一个线程的复查
• 批量方法 forEach(parallelismThreshold, …)、reduce、search 用 ForkJoinPool 并行
• ConcurrentSkipListMap：没有锁的有序 Map，跳表的 CAS 插入（和 Redis ZSet 对照）
• Caffeine 本地缓存：在 ConcurrentHashMap 之上做淘汰、过期，读写缓冲为什么用环形队列（J07）
• Disruptor 环形队列：为什么比 ArrayBlockingQueue 快一个数量级

自测