JDK Concurrent Collections 设计及工作原理与教程

这是关于 JDKjava.util.concurrent包下并发集合（Concurrent Collections）的设计原理、核心机制与实战教程。

内容会从设计哲学深入到具体实现，并附带代码示例。

第一部分：设计哲学与核心概念

并发集合的设计目标是在多线程环境下提供比synchronized包装类（如Collections.synchronizedMap）更高的并发性。

1.1 核心设计原则

• 细粒度锁（Fine-Grained Locking）：不对整个数据结构加锁，而是对数据的一部分（如桶、节点）加锁，允许不同线程操作不同部分。

• 无锁（Lock-Free）与乐观锁：利用 CAS（Compare-And-Swap）CPU 原子指令，避免线程阻塞（如ConcurrentLinkedQueue）。

• 写时复制（Copy-On-Write）：修改时复制一份新数据，读操作永远无需加锁（如CopyOnWriteArrayList）。

• 弱一致性（Weakly Consistent）迭代器：迭代器不抛出ConcurrentModificationException，且能反映迭代开始时的部分状态，但未必是最新状态。

1.2 与遗留同步容器的对比

第二部分：核心并发集合工作原理详解

2.1 ConcurrentHashMap（重点）

设计演进：

• JDK 1.7：分段锁（Segment 数组，每个 Segment 继承 ReentrantLock）。

• JDK 1.8+：废除分段锁，采用CAS + synchronized锁定链表/红黑树的头节点。

核心机制：

1. 初始化：延迟初始化，首次put时通过 CAS 创建数组。

2. 插入（put）流程：

   • 若桶为空：CAS 尝试插入新节点。

   • 若桶不为空：synchronized锁定桶的头节点，进行链表遍历或红黑树插入。

3. 扩容（transfer）：

   • 支持多线程并发扩容（每个线程负责一部分桶的迁移）。

   • 使用sizeCtl控制状态（负数表示正在扩容）。

4. 计数（size/mappingCount）：使用 LongAdder 风格的CounterCell数组分散计数，减少竞争。

注意：ConcurrentHashMap不允许 key/value 为 null，因为这会混淆“未找到”和“值为 null”的场景（在并发下无法用containsKey准确判断）。

2.2 BlockingQueue 实现类（生产者-消费者模型）

• ArrayBlockingQueue：有界，内部使用单锁（ReentrantLock）控制入队出队，支持公平/非公平。

• LinkedBlockingQueue：可选有界，内部使用双锁（takeLock + putLock）分离读写，吞吐量更高。

• SynchronousQueue：无容量，每个插入必须等待一个提取，常用于线程池（Executors.newCachedThreadPool）。

• PriorityBlockingQueue：无界，基于优先堆，元素需实现 Comparable。

工作原理：利用Condition（notEmpty/notFull）实现阻塞等待。当队列满时，put线程阻塞；当队列空时，take线程阻塞。

2.3 CopyOnWriteArrayList / CopyOnWriteArraySet

• 适用场景：读多写极少（如黑名单、监听器列表）。

• 原理：执行写操作（add/remove/set）时，复制底层数组，修改后重新赋值给 volatile 引用。

• 优点：读操作无需锁，性能极高。

• 缺点：写操作开销大（复制成本），数据强一致性延迟高。

2.4 ConcurrentLinkedQueue

• 无界非阻塞队列，基于Michael-Scott 无锁算法。

• 核心：使用 CAS 更新头尾节点（head和tail不一定指向真正的首尾，存在滞后性）。

• 迭代器：弱一致性，遍历时若节点被删除，跳过即可。

2.5 ConcurrentSkipListMap / SkipListSet

• 有序并发集合，基于跳表（Skip List）结构。

• 优势：提供 O(log n) 的并发访问，避免红黑树复杂的重平衡操作。

• 用途：需要并发有序 Map 时（如替代TreeMap+Collections.synchronized）。

第三部分：实战教程与代码示例

3.1 高性能缓存场景（使用 ConcurrentHashMap）

模拟一个简单的线程安全缓存，支持超时失效（简化版）。

import java.util.concurrent.*; public class ConcurrentCache<K, V> { private final ConcurrentHashMap<K, Future<V>> cache = new ConcurrentHashMap<>(); private final ConcurrentHashMap<K, Long> expireMap = new ConcurrentHashMap<>(); private final long ttlMillis; public ConcurrentCache(long ttlMillis) { this.ttlMillis = ttlMillis; } public V get(K key, Callable<V> loader) throws Exception { long now = System.currentTimeMillis(); // 检查过期 Long expireTime = expireMap.get(key); if (expireTime != null && now > expireTime) { cache.remove(key); expireMap.remove(key); } Future<V> future = cache.get(key); if (future == null) { FutureTask<V> task = new FutureTask<>(loader); Future<V> existing = cache.putIfAbsent(key, task); if (existing == null) { future = task; task.run(); // 执行加载 expireMap.put(key, now + ttlMillis); } else { future = existing; } } return future.get(); } }

3.2 生产者-消费者（使用 LinkedBlockingQueue）

public class ProducerConsumerDemo { public static void main(String[] args) { BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10); Runnable producer = () -> { try { for (int i = 0; i < 20; i++) { queue.put(i); // 阻塞直到有空间 System.out.println("Produced: " + i); } } catch (InterruptedException ignored) {} }; Runnable consumer = () -> { try { while (true) { Integer val = queue.take(); // 阻塞直到有元素 System.out.println("Consumed: " + val); Thread.sleep(100); } } catch (InterruptedException ignored) {} }; new Thread(producer).start(); new Thread(consumer).start(); new Thread(consumer).start(); // 多消费者 } }

3.3 读多写少场景（使用 CopyOnWriteArrayList）

public class BlacklistService { // 黑名单列表，读操作远多于写操作 private final CopyOnWriteArrayList<String> blacklist = new CopyOnWriteArrayList<>(); public boolean isBlacklisted(String ip) { return blacklist.contains(ip); // 无锁，极快 } public void addToBlacklist(String ip) { blacklist.add(ip); // 复制数组，开销大 } public void removeFromBlacklist(String ip) { blacklist.remove(ip); } }

第四部分：常见面试题与调优建议

4.1 常见面试问题

Q1：ConcurrentHashMap 在 JDK 1.8 中为什么放弃分段锁？

• 分段锁占用内存多（每个 Segment 继承 ReentrantLock）。

• 分段锁在扩容时仅锁住当前 Segment，但全局统计 size 需要获取所有锁，效率不高。

• 使用 CAS + synchronized 锁头节点，锁粒度更细（仅锁住单个桶），支持更高并发。

Q2：ConcurrentHashMap 如何保证线程安全？

• 初始化数组：CAS 保证单次初始化。

• 写操作：CAS 插入空桶，非空桶用 synchronized 锁头节点。

• 扩容：多线程协同迁移，每个线程使用 CAS 抢占任务。

Q3：ConcurrentLinkedQueue 的 size() 方法为何不准确？

• 因为在并发环境下，size 需要遍历全队列，且遍历过程中队列可能变化，因此返回的是弱一致性的近似值。

4.2 性能调优建议

4.3 常见陷阱

• 误用ConcurrentHashMap复合操作非原子：if (!map.containsKey(k)) map.put(k,v)不是原子操作，应改用putIfAbsent或computeIfAbsent。

• 阻塞队列的drainTo()在批量取数据时效率远高于循环take()。

• 不要将null值放入ConcurrentHashMap，否则可能 NPE。

第五部分：总结思维导图（文字版）

JDK 并发集合 ├── 设计哲学 │ ├── 细粒度锁 │ ├── CAS无锁 │ ├── Copy-On-Write │ └── 弱一致性迭代器 ├── Map体系 │ ├── ConcurrentHashMap（CAS+synchronized，无序） │ └── ConcurrentSkipListMap（跳表，有序） ├── List/Set体系 │ ├── CopyOnWriteArrayList/Set（读多写少） │ └── 无ConcurrentArrayList（自行Collections.synchronized） ├── Queue/Deque体系 │ ├── ConcurrentLinkedQueue（无界非阻塞） │ ├── LinkedBlockingQueue（有界双锁阻塞） │ ├── ArrayBlockingQueue（有界单锁阻塞） │ └── SynchronousQueue（0容量传递） └── 工具类辅助 ├── LongAdder（高并发计数） └── CompletableFuture（异步编排）