Java 并发工具类全景指南
本文系统介绍
java.util.concurrent包中最核心的并发工具类,涵盖同步协调、锁机制、原子类、并发容器与异步编排五大板块。每个工具类都配有完整代码示例,帮助你快速上手。
目录
- 开篇:为什么需要 JUC?
- CountDownLatch — 倒计时门闩
- CyclicBarrier — 循环栅栏
- Semaphore — 信号量
- ReentrantLock — 可重入锁
- ReentrantReadWriteLock — 读写锁
- StampedLock — 邮戳锁
- 原子类 Atomic 家族
- 并发容器
- CompletableFuture — 异步编排
- 横向对比与选型指南
- 总结
1. 开篇:为什么需要 JUC?
在 JUC 出现之前,Java 只有 synchronized 和 wait/notify 来处理并发。它们存在几个明显的局限性:
| 痛点 | 说明 |
|---|---|
| 不可中断 | synchronized 拿不到锁就一直阻塞,无法设置超时 |
| 不够灵活 | 无法实现"读共享、写独占"等细粒度控制 |
| 协调困难 | wait/notify 容易出错,且无法实现"等待 N 个线程完成"这样的语义 |
| 无高级抽象 | 没有限流、屏障、异步编排等开箱即用的工具 |
JDK 5 引入了 java.util.concurrent(JUC)包,由并发大师 Doug Lea 设计,提供了一整套高级并发工具。本文将逐一讲透其中最核心的成员。
2. CountDownLatch — 倒计时门闩
是什么?
CountDownLatch 是一个一次性的倒计数器。你设置一个初始计数值 N,每当一个任务完成时调用 countDown() 将计数减 1,而等待的线程调用 await() 阻塞直到计数归零。
一句话定位:让一个或多个线程等待其他 N 个操作完成后再继续。
核心方法
| 方法 | 说明 |
|---|---|
CountDownLatch(int count) | 构造器,设置倒计数初始值 |
countDown() | 计数减 1,当计数到 0 时唤醒所有等待线程 |
await() | 阻塞当前线程,直到计数归零 |
await(long timeout, TimeUnit unit) | 带超时的等待,超时返回 false |
getCount() | 获取当前剩余计数 |
场景一:主线程等待多个子任务完成
最经典的用法——主线程发起 N 个并行任务,等它们全部完成后汇总结果。
public class HealthCheckDemo {
public static void main(String[] args) throws InterruptedException {
// 3 个服务需要健康检查
CountDownLatch latch = new CountDownLatch(3);
String[] services = {"数据库", "Redis", "消息队列"};
for (String service : services) {
new Thread(() -> {
try {
// 模拟检查耗时
Thread.sleep((long) (Math.random() * 2000));
System.out.println(service + " 健康检查通过 ✅");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
latch.countDown(); // 无论成功失败都要 countDown
}
}).start();
}
System.out.println("等待所有服务健康检查...");
boolean ready = latch.await(5, TimeUnit.SECONDS); // 最多等 5 秒
if (ready) {
System.out.println("所有服务就绪,应用启动!🚀");
} else {
System.out.println("部分服务超时,启动失败 ❌");
}
}
}
场景二:模拟并发压测(发令枪)
反过来用——让 N 个线程等待一个信号,同时起跑。
public class ConcurrentTestDemo {
public static void main(String[] args) throws InterruptedException {
int threadCount = 10;
CountDownLatch startSignal = new CountDownLatch(1); // 发令枪
CountDownLatch doneSignal = new CountDownLatch(threadCount); // 完成信号
for (int i = 0; i < threadCount; i++) {
final int id = i;
new Thread(() -> {
try {
startSignal.await(); // 所有线程在此等待
System.out.println("线程" + id + " 开始执行,时间:" + System.currentTimeMillis());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
doneSignal.countDown();
}
}).start();
}
Thread.sleep(1000); // 确保所有线程都到达 await
System.out.println("3...2...1...Go! 🏁");
startSignal.countDown(); // 发令!所有线程几乎同时开始
doneSignal.await(); // 等待所有线程完成
System.out.println("所有线程执行完毕");
}
}
⚠️ 注意事项
- 一次性不可重置:计数归零后无法重新设置,想复用请用
CyclicBarrier - countDown 必须在 finally 中:如果线程异常退出忘记
countDown(),await()将永久阻塞 - 务必使用带超时的 await:生产环境中避免无限等待
- count 只能减不能加:初始值设好后就固定了
3. CyclicBarrier — 循环栅栏
是什么?
CyclicBarrier 让一组线程互相等待,直到所有线程都到达屏障点后才一起继续执行。与 CountDownLatch 最大的区别:它可以重复使用(Cyclic = 循环)。
一句话定位:N 个线程互相等待,到齐后一起出发,可以反复使用。
核心方法
| 方法 | 说明 |
|---|---|
CyclicBarrier(int parties) | 构造器,指定参与线程数 |
CyclicBarrier(int parties, Runnable barrierAction) | 到齐后先执行 barrierAction,再唤醒所有线程 |
await() | 到达屏障点并等待其他线程 |
await(long timeout, TimeUnit unit) | 带超时的等待 |
reset() | 重置屏障(会使正在等待的线程抛出 BrokenBarrierException) |
getNumberWaiting() | 当前已到达屏障点的线程数 |
代码示例:多线程分片处理后汇总
public class CyclicBarrierDemo {
public static void main(String[] args) {
int workerCount = 3;
// 所有线程到齐后执行汇总操作(由最后到达的线程执行)
CyclicBarrier barrier = new CyclicBarrier(workerCount, () -> {
System.out.println(">>> 所有分片处理完毕,开始汇总!(执行线程:"
+ Thread.currentThread().getName() + ")");
});
for (int i = 0; i < workerCount; i++) {
final int shardId = i;
new Thread(() -> {
try {
System.out.println("线程 " + shardId + " 正在处理第一批数据...");
Thread.sleep((long) (Math.random() * 2000));
System.out.println("线程 " + shardId + " 第一批完成,等待其他线程 ⏳");
barrier.await(); // 第一轮
System.out.println("线程 " + shardId + " 正在处理第二批数据...");
Thread.sleep((long) (Math.random() * 2000));
System.out.println("线程 " + shardId + " 第二批完成,等待其他线程 ⏳");
barrier.await(); // 第二轮 —— 自动复用!
System.out.println("线程 " + shardId + " 全部完成 ✅");
} catch (InterruptedException | BrokenBarrierException e) {
Thread.currentThread().interrupt();
}
}, "Worker-" + shardId).start();
}
}
}
CountDownLatch vs CyclicBarrier 对比
| 维度 | CountDownLatch | CyclicBarrier |
|---|---|---|
| 可重用 | ❌ 一次性 | ✅ 自动重置,可循环使用 |
| 等待模式 | 线程 A 等待线程 B、C、D 完成 | 线程 A、B、C 互相等待,到齐再走 |
| 计数方式 | 调用 countDown() 主动减计数 | 调用 await() 时自动计数 |
| 额外操作 | 无 | 到齐后可执行 barrierAction |
| 典型场景 | 主线程等子任务完成 | 多线程分阶段协同 |
💡 选择建议:如果是"一个等多个"→ CountDownLatch;如果是"多个互相等"→ CyclicBarrier。
⚠️ 注意事项
- BrokenBarrierException:如果某个线程在
await()期间被中断或超时,屏障会被"打破",其他等待线程都会收到此异常 - reset() 慎用:会使所有正在等待的线程抛异常
- parties 数量必须精确:多了会导致永久等待,少了会逻辑错误
4. Semaphore — 信号量
是什么?
Semaphore(信号量)维护了一组许可证(permits)。线程通过 acquire() 获取许可证,通过 release() 归还许可证。当许可证耗尽时,后续线程将阻塞等待。
一句话定位:控制同时访问某个资源的最大线程数。
核心方法
| 方法 | 说明 |
|---|---|
Semaphore(int permits) | 构造器,设置许可证数量(默认非公平) |
Semaphore(int permits, boolean fair) | 可选公平模式 |
acquire() | 获取一个许可证,无可用则阻塞 |
acquire(int n) | 一次获取 n 个许可证 |
tryAcquire() | 非阻塞地尝试获取,返回 boolean |
tryAcquire(long timeout, TimeUnit unit) | 带超时的尝试获取 |
release() | 归还一个许可证 |
availablePermits() | 当前可用许可证数量 |
代码示例:停车场限流
public class ParkingLotDemo {
public static void main(String[] args) {
// 停车场只有 3 个车位
Semaphore parkingLot = new Semaphore(3, true); // 公平模式
for (int i = 1; i <= 8; i++) {
final int carId = i;
new Thread(() -> {
try {
System.out.println("🚗 车辆 " + carId + " 想进入停车场...");
parkingLot.acquire(); // 获取车位
System.out.println("🅿️ 车辆 " + carId + " 已停入,剩余车位:"
+ parkingLot.availablePermits());
Thread.sleep((long) (Math.random() * 5000)); // 模拟停车时间
System.out.println("🚙 车辆 " + carId + " 驶出停车场");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
parkingLot.release(); // 必须在 finally 中释放!
}
}).start();
}
}
}
⚠️ 注意事项
- release 可以超出初始值:如果你多调了一次
release(),许可证总数会增加!这不是 Bug,是设计——但通常不是你想要的 - acquire/release 必须配对:放在
try/finally中确保释放 - Semaphore(1) ≈ 互斥锁:但与
ReentrantLock不同,Semaphore 没有"持有者"概念,任何线程都可以 release - 公平模式有性能开销:按 FIFO 排队会降低吞吐量,非必要不开启
5. ReentrantLock — 可重入锁
是什么?
ReentrantLock 是 synchronized 的增强替代品。它提供了与 synchronized 相同的互斥性和可重入性,但额外支持可中断等待、超时获取、公平锁、多条件变量等高级特性。
一句话定位:功能更强大、更灵活的 synchronized。
核心方法
| 方法 | 说明 |
|---|---|
lock() | 获取锁,获取不到就阻塞 |
unlock() | 释放锁,必须手动调用 |
tryLock() | 非阻塞尝试获取锁 |
tryLock(long timeout, TimeUnit unit) | 带超时的尝试获取 |
lockInterruptibly() | 可响应中断的获取锁 |
newCondition() | 创建条件变量(替代 wait/notify) |
isHeldByCurrentThread() | 当前线程是否持有锁 |
与 synchronized 的对比
| 维度 | synchronized | ReentrantLock |
|---|---|---|
| 获取锁 | 自动获取 | 手动 lock() |
| 释放锁 | 自动释放 | 手动 unlock()(必须 finally) |
| 可中断 | ❌ | ✅ lockInterruptibly() |
| 可超时 | ❌ | ✅ tryLock(timeout) |
| 公平锁 | ❌ | ✅ 构造时指定 |
| 条件变量 | 一个(wait/notify) | 多个(newCondition()) |
| 性能 | JDK 6 后优化很好 | 相当,差异很小 |
💡 选择建议:简单同步用
synchronized(简洁安全);需要上述高级特性时用ReentrantLock。
代码示例:生产者-消费者(Condition 版)
public class ProducerConsumerDemo {
private static final ReentrantLock lock = new ReentrantLock();
private static final Condition notFull = lock.newCondition(); // 队列未满
private static final Condition notEmpty = lock.newCondition(); // 队列非空
private static final Queue<Integer> queue = new LinkedList<>();
private static final int MAX_SIZE = 5;
public static void main(String[] args) {
// 生产者
new Thread(() -> {
for (int i = 0; i < 20; i++) {
lock.lock();
try {
while (queue.size() == MAX_SIZE) {
notFull.await(); // 队列满了,等待消费者消费
}
queue.offer(i);
System.out.println("生产: " + i + ",队列大小: " + queue.size());
notEmpty.signal(); // 通知消费者
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock(); // 绝对不能忘!
}
}
}, "Producer").start();
// 消费者
new Thread(() -> {
for (int i = 0; i < 20; i++) {
lock.lock();
try {
while (queue.isEmpty()) {
notEmpty.await(); // 队列空了,等待生产者生产
}
int value = queue.poll();
System.out.println("消费: " + value + ",队列大小: " + queue.size());
notFull.signal(); // 通知生产者
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
}
}
}, "Consumer").start();
}
}
注意 Condition 的精妙之处:生产者只唤醒消费者(notEmpty.signal()),消费者只唤醒生产者(notFull.signal()),避免了 synchronized 中 notifyAll() 的惊群效应。
⚠️ 注意事项
- unlock 必须在 finally 中:这是最容易踩的坑,忘记释放锁就是死锁
- lock() 要写在 try 外面:如果写在 try 里面,lock() 抛异常时 finally 中的 unlock() 会释放一个未持有的锁
- 公平锁有代价:
new ReentrantLock(true)保证 FIFO,但吞吐量下降明显 - await 要用 while 循环:防止虚假唤醒(spurious wakeup)
// ✅ 正确写法
lock.lock();
try {
// 业务代码
} finally {
lock.unlock();
}
// ❌ 错误写法
try {
lock.lock(); // 如果这里抛异常,finally 会执行 unlock
// 业务代码
} finally {
lock.unlock();
}
6. ReentrantReadWriteLock — 读写锁
是什么?
ReentrantReadWriteLock 将锁分为读锁(共享锁)和写锁(独占锁)。多个线程可以同时持有读锁,但写锁是互斥的。
一句话定位:读多写少场景的性能优化锁。
核心规则
| 操作 | 读锁 | 写锁 |
|---|---|---|
| 读锁已持有 | ✅ 可以再获取读锁 | ❌ 写锁阻塞 |
| 写锁已持有 | ❌ 读锁阻塞 | ❌ 写锁阻塞(可重入除外) |
简单记:读读共享,读写互斥,写写互斥。
代码示例:缓存系统
public class CacheDemo {
private static final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
private static final Map<String, String> cache = new HashMap<>();
// 读操作:多线程可并发读
public static String get(String key) {
rwLock.readLock().lock();
try {
System.out.println(Thread.currentThread().getName() + " 正在读取...");
Thread.sleep(100); // 模拟耗时
return cache.get(key);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
} finally {
rwLock.readLock().unlock();
}
}
// 写操作:独占,其他读写都阻塞
public static void put(String key, String value) {
rwLock.writeLock().lock();
try {
System.out.println(Thread.currentThread().getName() + " 正在写入...");
Thread.sleep(200); // 模拟耗时
cache.put(key, value);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
rwLock.writeLock().unlock();
}
}
}
锁降级(重要概念)
ReentrantReadWriteLock 支持锁降级(写锁 → 读锁),但不支持锁升级(读锁 → 写锁)。
// ✅ 锁降级:先获取写锁,再获取读锁,然后释放写锁
rwLock.writeLock().lock();
try {
// 修改数据
cache.put("key", "newValue");
// 降级:持有写锁时获取读锁
rwLock.readLock().lock();
} finally {
rwLock.writeLock().unlock(); // 释放写锁,保留读锁
}
try {
// 此时持有读锁,可以继续读取,且不怕其他写线程修改
String value = cache.get("key");
} finally {
rwLock.readLock().unlock();
}
// ❌ 锁升级:持有读锁时获取写锁 → 死锁!
rwLock.readLock().lock();
try {
rwLock.writeLock().lock(); // 💀 永久阻塞!因为写锁需要等所有读锁释放
} finally {
rwLock.readLock().unlock();
}
⚠️ 注意事项
- 不支持锁升级:在持有读锁时尝试获取写锁会导致死锁
- 写饥饿问题:如果读操作持续不断,写线程可能长时间拿不到锁。可以用公平模式缓解
- 适用场景:读写比至少 10:1 以上才有明显收益,否则不如直接用
ReentrantLock
7. StampedLock — 邮戳锁(Java 8+)
是什么?
StampedLock 是 JDK 8 引入的读写锁升级版,新增了乐观读模式。在没有写操作的情况下,乐观读完全不加锁,性能极高。
一句话定位:在读写锁基础上增加乐观读,极致读性能。
三种模式
| 模式 | 说明 | 性能 |
|---|---|---|
写锁 writeLock() | 独占锁,与 ReentrantReadWriteLock 类似 | - |
悲观读锁 readLock() | 共享锁,与 ReentrantReadWriteLock 类似 | 中 |
乐观读 tryOptimisticRead() | 不加锁! 只获取一个标记(stamp),事后验证 | 极高 |
代码示例:坐标点的读写(JDK 官方经典示例)
public class Point {
private double x, y;
private final StampedLock sl = new StampedLock();
// 写操作
public void move(double deltaX, double deltaY) {
long stamp = sl.writeLock(); // 获取写锁
try {
x += deltaX;
y += deltaY;
} finally {
sl.unlockWrite(stamp); // 释放写锁
}
}
// 乐观读(推荐的读方式)
public double distanceFromOrigin() {
long stamp = sl.tryOptimisticRead(); // 乐观读,不加锁!
double currentX = x, currentY = y; // 读取数据到局部变量
if (!sl.validate(stamp)) {
// 在读取期间有写操作发生,乐观读失败,升级为悲观读锁
stamp = sl.readLock();
try {
currentX = x;
currentY = y;
} finally {
sl.unlockRead(stamp);
}
}
return Math.sqrt(currentX * currentX + currentY * currentY);
}
}
乐观读的执行流程
1. tryOptimisticRead() → 获得 stamp(不加锁,超快)
2. 读取数据到局部变量(注意:必须是局部变量!)
3. validate(stamp) → 检查期间是否有写操作
├── true → 数据有效,直接使用 ✅
└── false → 数据可能被修改,升级为悲观读锁重新读取 🔄
StampedLock vs ReentrantReadWriteLock
| 维度 | ReentrantReadWriteLock | StampedLock |
|---|---|---|
| 可重入 | ✅ | ❌ |
| 乐观读 | ❌ | ✅ |
| Condition | ✅ | ❌ |
| 读性能 | 好 | 极好(乐观读无锁) |
| 使用难度 | 简单 | 较复杂(stamp 管理) |
⚠️ 注意事项
- 不可重入! 在持有 StampedLock 的锁时再次获取同类型的锁会导致死锁
- 不支持 Condition
- 乐观读的模式必须严格遵循:先读到局部变量,再 validate,失败后降级
- stamp 不能传 0:
validate(0)永远返回 false - 适合读密集场景:如果写操作频繁导致乐观读一直失败,性能反而不如 ReentrantReadWriteLock
8. 原子类 Atomic 家族
是什么?
原子类利用 CAS(Compare-And-Swap) 操作实现无锁的线程安全。CAS 是一种硬件级别的原子指令:比较内存中的值和预期值是否相同,如果相同就更新为新值,否则不做任何操作。
一句话定位:不用锁也能实现线程安全的变量操作,性能优于 synchronized。
分类一览
| 分类 | 包含的类 | 适用场景 |
|---|---|---|
| 基本原子类 | AtomicInteger, AtomicLong, AtomicBoolean | 计数器、标志位 |
| 引用原子类 | AtomicReference, AtomicStampedReference, AtomicMarkableReference | 原子更新对象引用 |
| 数组原子类 | AtomicIntegerArray, AtomicLongArray, AtomicReferenceArray | 原子更新数组中某个元素 |
| 累加器(Java 8+) | LongAdder, LongAccumulator, DoubleAdder, DoubleAccumulator | 高并发计数/统计 |
8.1 基本原子类
public class AtomicCounterDemo {
// 线程不安全的计数器
private static int unsafeCount = 0;
// 线程安全的计数器
private static final AtomicInteger safeCount = new AtomicInteger(0);
public static void main(String[] args) throws InterruptedException {
int threadCount = 100;
CountDownLatch latch = new CountDownLatch(threadCount);
for (int i = 0; i < threadCount; i++) {
new Thread(() -> {
for (int j = 0; j < 1000; j++) {
unsafeCount++; // 非原子操作,结果不确定
safeCount.incrementAndGet(); // 原子操作,结果一定是 100000
}
latch.countDown();
}).start();
}
latch.await();
System.out.println("unsafeCount = " + unsafeCount); // 大概率 < 100000
System.out.println("safeCount = " + safeCount.get()); // 一定是 100000
}
}
常用方法:
| 方法 | 说明 |
|---|---|
get() | 获取当前值 |
set(int newValue) | 设置新值 |
getAndIncrement() | 先获取再自增(i++) |
incrementAndGet() | 先自增再获取(++i) |
getAndAdd(int delta) | 先获取再加上 delta |
compareAndSet(expect, update) | CAS 操作:期望值等于 expect 则更新为 update |
updateAndGet(UnaryOperator) | 原子地使用函数更新值(Java 8+) |
8.2 引用原子类与 ABA 问题
AtomicReference 可以原子更新对象引用,但会遇到 ABA 问题:
时刻1:线程1读取值 A
时刻2:线程2将 A 改为 B
时刻3:线程2将 B 改回 A
时刻4:线程1执行 CAS,发现值还是 A,认为没变过 → 但实际上变过了!
解决方案:使用 AtomicStampedReference,每次修改都更新版本号(stamp)。
public class AbaDemo {
public static void main(String[] args) {
// 初始值 "A",版本号 1
AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 1);
// 线程1:制造 ABA
new Thread(() -> {
int stamp = ref.getStamp();
ref.compareAndSet("A", "B", stamp, stamp + 1);
ref.compareAndSet("B", "A", stamp + 1, stamp + 2);
System.out.println("线程1 完成 A→B→A,当前版本号:" + ref.getStamp());
}).start();
// 线程2:尝试 CAS
new Thread(() -> {
int stamp = ref.getStamp(); // 拿到版本号 1
try { Thread.sleep(500); } catch (InterruptedException e) {}
boolean success = ref.compareAndSet("A", "C", stamp, stamp + 1);
System.out.println("线程2 CAS 结果:" + success); // false!版本号对不上
System.out.println("当前值:" + ref.getReference() + ",版本号:" + ref.getStamp());
}).start();
}
}
8.3 累加器 LongAdder(重点推荐)
LongAdder 是 Java 8 引入的高性能计数器。在高并发场景下,它的性能远优于 AtomicLong。
原理简述:AtomicLong 只有一个 value,所有线程 CAS 竞争同一个变量。LongAdder 将热点分散到多个 Cell 中,每个线程在自己的 Cell 上操作,最后汇总求和。
AtomicLong: 所有线程 → [value] ← CAS 冲突严重
LongAdder: 线程1 → [Cell0]
线程2 → [Cell1] sum() 时汇总
线程3 → [Cell2]
public class LongAdderDemo {
public static void main(String[] args) throws InterruptedException {
LongAdder adder = new LongAdder();
AtomicLong atomicLong = new AtomicLong(0);
int threadCount = 100;
CountDownLatch latch = new CountDownLatch(threadCount * 2);
long start1 = System.nanoTime();
for (int i = 0; i < threadCount; i++) {
new Thread(() -> {
for (int j = 0; j < 100000; j++) {
adder.increment();
}
latch.countDown();
}).start();
}
long start2 = System.nanoTime();
for (int i = 0; i < threadCount; i++) {
new Thread(() -> {
for (int j = 0; j < 100000; j++) {
atomicLong.incrementAndGet();
}
latch.countDown();
}).start();
}
latch.await();
System.out.println("LongAdder 结果: " + adder.sum());
System.out.println("AtomicLong 结果: " + atomicLong.get());
// 在高并发下,LongAdder 通常快数倍
}
}
选型建议
| 场景 | 推荐 |
|---|---|
| 简单计数、状态标志 | AtomicInteger / AtomicLong |
| 高并发计数(如:统计 QPS) | LongAdder ✅ |
| 需要获取精确实时值 | AtomicLong(LongAdder 的 sum() 非精确快照) |
| 原子更新对象引用 | AtomicReference |
| 需要防止 ABA 问题 | AtomicStampedReference |
9. 并发容器
传统的 ArrayList、HashMap 在多线程环境下不安全。Collections.synchronizedXxx() 虽然安全但性能差(整个方法加锁)。JUC 提供了专门为并发设计的高性能容器。
9.1 ConcurrentHashMap
一句话定位:线程安全的 HashMap,高并发读写性能优秀。
实现原理简述:
- JDK 7:分段锁(Segment),将 Map 分为 16 段,每段一把锁
- JDK 8+:CAS + synchronized(锁单个 Node),粒度更细,性能更好
public class ConcurrentHashMapDemo {
public static void main(String[] args) throws InterruptedException {
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
// ❌ 错误用法:先 get 再 put 不是原子操作
// if (map.get("count") == null) { map.put("count", 1); }
// ✅ 正确用法:使用原子复合操作
map.putIfAbsent("count", 0); // 不存在时才放入
map.compute("count", (key, val) -> val == null ? 1 : val + 1); // 原子计算
map.merge("count", 1, Integer::sum); // 原子合并
// 并发统计词频
ConcurrentHashMap<String, LongAdder> wordCount = new ConcurrentHashMap<>();
String[] words = {"java", "go", "java", "rust", "java", "go"};
for (String word : words) {
wordCount.computeIfAbsent(word, k -> new LongAdder()).increment();
}
wordCount.forEach((word, count) ->
System.out.println(word + ": " + count.sum()));
}
}
与其他线程安全 Map 对比:
| 维度 | Hashtable | synchronizedMap | ConcurrentHashMap |
|---|---|---|---|
| 锁粒度 | 整个 Map | 整个 Map | 单个 Node(JDK 8+) |
| 允许 null key/value | ❌ | ✅ | ❌ |
| 性能 | 差 | 差 | 优秀 |
| 原子复合操作 | ❌ | ❌ | ✅ compute/merge 等 |
| 推荐程度 | ❌ 已过时 | ⚠️ 临时用 | ✅ 首选 |
9.2 CopyOnWriteArrayList / CopyOnWriteArraySet
一句话定位:写时复制,读不加锁,适合读远多于写的场景。
工作原理:
读操作 → 直接读底层数组,不加锁 ✅
写操作 → 复制整个数组 → 在副本上修改 → 用副本替换原数组(加锁)
public class CopyOnWriteDemo {
public static void main(String[] args) {
// 适合场景:事件监听器列表、配置项列表
CopyOnWriteArrayList<String> listeners = new CopyOnWriteArrayList<>();
listeners.add("监听器A");
listeners.add("监听器B");
listeners.add("监听器C");
// 读操作:无锁,可在遍历时安全添加/删除(不会 ConcurrentModificationException)
for (String listener : listeners) {
System.out.println("通知: " + listener);
if ("监听器B".equals(listener)) {
listeners.remove(listener); // 遍历中删除,不会报错!
}
}
System.out.println("剩余监听器: " + listeners);
}
}
⚠️ 注意事项:
- 写操作很重:每次写都复制整个数组,数据量大时内存开销巨大
- 弱一致性:写入后,正在遍历的线程看到的可能还是旧数据
- 适用条件:读操作 >> 写操作(如 100:1 以上),且数据量不大
9.3 BlockingQueue 系列
一句话定位:带阻塞能力的队列,是生产者-消费者模式的最佳搭档。
四组操作方法对比:
| 操作 | 抛异常 | 返回特殊值 | 阻塞 | 超时 |
|---|---|---|---|---|
| 插入 | add(e) | offer(e) → false | put(e) | offer(e, timeout, unit) |
| 移除 | remove() | poll() → null | take() | poll(timeout, unit) |
| 检查 | element() | peek() → null | - | - |
💡 生产环境推荐使用
put/take(阻塞)或offer/poll(带超时),避免使用抛异常的方法。
常见实现类:
| 实现类 | 底层结构 | 是否有界 | 特点 |
|---|---|---|---|
ArrayBlockingQueue | 数组 | ✅ 有界 | 创建时指定容量,公平/非公平可选 |
LinkedBlockingQueue | 链表 | 可选 | 默认 Integer.MAX_VALUE,吞吐量更高 |
PriorityBlockingQueue | 堆 | ❌ 无界 | 按优先级出队,非 FIFO |
SynchronousQueue | 无 | 无容量 | 不存储元素,直接传递(一手交一手) |
DelayQueue | 堆 | ❌ 无界 | 元素到期后才能取出(定时任务) |
LinkedTransferQueue | 链表 | ❌ 无界 | 支持 transfer() 预占模式 |
代码示例:生产者-消费者
public class BlockingQueueDemo {
public static void main(String[] args) {
// 有界队列,容量 5
BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
// 生产者
new Thread(() -> {
try {
for (int i = 1; i <= 10; i++) {
String item = "商品-" + i;
queue.put(item); // 队列满时自动阻塞
System.out.println("📦 生产: " + item + ",队列大小: " + queue.size());
}
queue.put("END"); // 结束信号
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "Producer").start();
// 消费者
new Thread(() -> {
try {
while (true) {
String item = queue.take(); // 队列空时自动阻塞
if ("END".equals(item)) break;
System.out.println("🛒 消费: " + item);
Thread.sleep(500); // 模拟处理耗时
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "Consumer").start();
}
}
⚠️ BlockingQueue 选型建议:
- 固定大小的缓冲区 →
ArrayBlockingQueue - 高吞吐量的生产者-消费者 →
LinkedBlockingQueue(注意设置容量上限!) - 线程池的任务传递 →
SynchronousQueue(Executors.newCachedThreadPool 就用它) - 定时任务/延迟执行 →
DelayQueue - 按优先级处理 →
PriorityBlockingQueue
10. CompletableFuture — 异步编排(Java 8+)
是什么?
CompletableFuture 是 Java 8 引入的异步编程框架。传统的 Future 只能通过 get() 阻塞等待结果,而 CompletableFuture 支持回调链式调用,可以优雅地编排多个异步任务的执行顺序和结果组合。
一句话定位:Java 版的 Promise,异步任务编排神器。
创建异步任务
// 有返回值的异步任务
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
return "查询结果";
});
// 无返回值的异步任务
CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
System.out.println("执行异步操作");
});
// 指定线程池(推荐!)
ExecutorService pool = Executors.newFixedThreadPool(4);
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "result", pool);
链式操作速查表
| 方法 | 入参 | 返回值 | 说明 |
|---|---|---|---|
thenApply(Function) | 上一步结果 | 有 | 转换结果(map) |
thenAccept(Consumer) | 上一步结果 | 无 | 消费结果 |
thenRun(Runnable) | 无 | 无 | 执行后续操作 |
thenCompose(Function) | 上一步结果 | CompletableFuture | 扁平化(flatMap) |
thenCombine(other, BiFunction) | 两个结果 | 有 | 合并两个独立任务的结果 |
allOf(cf1, cf2, ...) | 多个 CF | Void | 等待所有任务完成 |
anyOf(cf1, cf2, ...) | 多个 CF | Object | 任一任务完成即返回 |
💡 以上方法都有
xxxAsync变体(如thenApplyAsync),表示在异步线程中执行回调。
异常处理
CompletableFuture<String> future = CompletableFuture
.supplyAsync(() -> {
if (true) throw new RuntimeException("出错了!");
return "ok";
})
.exceptionally(ex -> {
// 类似 try-catch,返回兜底值
System.out.println("异常: " + ex.getMessage());
return "默认值";
});
// 或者用 handle —— 同时处理正常结果和异常
CompletableFuture<String> future2 = CompletableFuture
.supplyAsync(() -> "ok")
.handle((result, ex) -> {
if (ex != null) {
return "出错了: " + ex.getMessage();
}
return "成功: " + result;
});
实战示例:电商下单(并行查询 + 汇总)
public class OrderDemo {
public static void main(String[] args) {
ExecutorService pool = Executors.newFixedThreadPool(3);
// 三个独立查询并行执行
CompletableFuture<Integer> stockFuture = CompletableFuture.supplyAsync(() -> {
sleep(200);
System.out.println("查询库存...");
return 100; // 库存数量
}, pool);
CompletableFuture<Double> priceFuture = CompletableFuture.supplyAsync(() -> {
sleep(300);
System.out.println("查询价格...");
return 99.9; // 价格
}, pool);
CompletableFuture<Double> couponFuture = CompletableFuture.supplyAsync(() -> {
sleep(150);
System.out.println("查询优惠券...");
return 10.0; // 优惠金额
}, pool);
// 汇总三个结果,计算最终价格
CompletableFuture<String> orderFuture = stockFuture
.thenCombine(priceFuture, (stock, price) -> {
if (stock <= 0) throw new RuntimeException("库存不足");
return price;
})
.thenCombine(couponFuture, (price, coupon) -> {
double finalPrice = price - coupon;
return "下单成功!最终价格:" + finalPrice;
})
.exceptionally(ex -> "下单失败:" + ex.getMessage());
System.out.println(orderFuture.join()); // join() 不抛 checked exception
pool.shutdown();
}
private static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException e) {}
}
}
⚠️ 注意事项
- 不要使用默认线程池:
supplyAsync()默认使用ForkJoinPool.commonPool(),它是全局共享的,线程数等于 CPU 核心数。IO 密集型任务应指定独立线程池 - 异常会被吞没:如果不处理异常(没有
exceptionally/handle/whenComplete),异常只有在get()/join()时才会抛出 - join() vs get():
join()抛非受检异常,get()抛受检异常(需要 try-catch)。推荐用join() - 避免嵌套 CompletableFuture:用
thenCompose替代thenApply+get() - 记得关闭线程池:自定义线程池用完要
shutdown()
11. 横向对比与选型指南
同步协调工具对比
| 维度 | CountDownLatch | CyclicBarrier | Semaphore |
|---|---|---|---|
| 可重用 | ❌ | ✅ | ✅ |
| 参与者 | 固定 | 固定 | 动态 |
| 等待模式 | 一个等多个 | 互相等待 | 许可证控制 |
| 到达后操作 | 无 | barrierAction | 无 |
| 典型场景 | 启动等待/汇总 | 分段协同计算 | 限流/资源池 |
锁的选型决策树
需要线程同步吗?
│
├── 简单的互斥同步
│ └── synchronized(优先!JVM 优化好,不会忘记释放)
│
├── 需要高级特性(可中断/超时/公平/多条件)
│ └── ReentrantLock
│
├── 读多写少(读写比 > 10:1)
│ ├── 需要可重入、Condition → ReentrantReadWriteLock
│ └── 追求极致读性能 → StampedLock(乐观读)
│
└── 不需要锁,只是简单的变量操作
├── 低并发 → AtomicInteger / AtomicLong
└── 高并发计数 → LongAdder
一句话总结表
| 工具类 | 一句话定位 | 最佳场景 | 踩坑警告 |
|---|---|---|---|
| CountDownLatch | 一次性倒计数 | 等待 N 个任务完成 | 一次性不可重置 |
| CyclicBarrier | 可重用屏障 | 多线程分阶段协同 | parties 必须精确 |
| Semaphore | 许可证限流 | 控制并发数 | release 会膨胀 permits |
| ReentrantLock | 增强版 synchronized | 需要高级锁特性 | unlock 必须 finally |
| ReadWriteLock | 读写分离锁 | 读多写少的缓存 | 不支持锁升级 |
| StampedLock | 乐观读锁 | 极致读性能 | 不可重入! |
| AtomicInteger | 原子计数器 | 简单计数/标志位 | CAS 自旋开销 |
| LongAdder | 高并发计数器 | 统计 QPS/热点计数 | sum() 非精确快照 |
| ConcurrentHashMap | 并发 Map | 多线程共享 Map | 复合操作要用 compute |
| CopyOnWriteArrayList | 写时复制 List | 读多写极少 | 写操作内存开销大 |
| BlockingQueue | 阻塞队列 | 生产者-消费者 | LinkedBQ 要设上限 |
| CompletableFuture | 异步编排 | 并行任务组合 | 别用默认线程池 |
12. 总结
学习路线建议
入门(必须掌握)
├── CountDownLatch + CyclicBarrier + Semaphore
├── ReentrantLock(含 Condition)
├── ConcurrentHashMap
└── CompletableFuture
进阶(按需深入)
├── StampedLock(乐观读)
├── LongAdder(高并发计数)
├── BlockingQueue 全家族
└── CopyOnWriteArrayList
底层原理(另起一篇)
├── AQS(AbstractQueuedSynchronizer)—— 大部分同步工具的基石
├── CAS 底层实现(Unsafe / VarHandle)
└── JMM(Java 内存模型)与 happens-before
推荐阅读
- 📖 《Java 并发编程的艺术》— 方腾飞
- 📖 《Java Concurrency in Practice》— Brian Goetz
- 📖 JDK 源码中每个并发类的 Javadoc(写得非常好,值得细读)
📝 作者注:本文以"会用 + 知道原理概要"为目标,不深入 AQS 源码和 CAS 底层实现。关于 AQS 的详细解析,将在后续文章中单独展开。