Skip to main content

Java 并发工具类全景指南

本文系统介绍 java.util.concurrent 包中最核心的并发工具类，涵盖同步协调、锁机制、原子类、并发容器与异步编排五大板块。每个工具类都配有完整代码示例，帮助你快速上手。

目录​

  1. 开篇：为什么需要 JUC？
  2. CountDownLatch — 倒计时门闩
  3. CyclicBarrier — 循环栅栏
  4. Semaphore — 信号量
  5. ReentrantLock — 可重入锁
  6. ReentrantReadWriteLock — 读写锁
  7. StampedLock — 邮戳锁
  8. 原子类 Atomic 家族
  9. 并发容器
  10. CompletableFuture — 异步编排
  11. 横向对比与选型指南
  12. 总结

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("所有线程执行完毕");
    }
}

⚠️ 注意事项​

  1. 一次性不可重置：计数归零后无法重新设置，想复用请用 CyclicBarrier
  2. countDown 必须在 finally 中：如果线程异常退出忘记 countDown()，await() 将永久阻塞
  3. 务必使用带超时的 await：生产环境中避免无限等待
  4. 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 对比​

维度CountDownLatchCyclicBarrier
可重用❌ 一次性✅ 自动重置，可循环使用
等待模式线程 A 等待线程 B、C、D 完成线程 A、B、C 互相等待，到齐再走
计数方式调用 countDown() 主动减计数调用 await() 时自动计数
额外操作无到齐后可执行 barrierAction
典型场景主线程等子任务完成多线程分阶段协同

💡 选择建议：如果是"一个等多个"→ CountDownLatch；如果是"多个互相等"→ CyclicBarrier。

⚠️ 注意事项​

  1. BrokenBarrierException：如果某个线程在 await() 期间被中断或超时，屏障会被"打破"，其他等待线程都会收到此异常
  2. reset() 慎用：会使所有正在等待的线程抛异常
  3. 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();
        }
    }
}

⚠️ 注意事项​

  1. release 可以超出初始值：如果你多调了一次 release()，许可证总数会增加！这不是 Bug，是设计——但通常不是你想要的
  2. acquire/release 必须配对：放在 try/finally 中确保释放
  3. Semaphore(1) ≈ 互斥锁：但与 ReentrantLock 不同，Semaphore 没有"持有者"概念，任何线程都可以 release
  4. 公平模式有性能开销：按 FIFO 排队会降低吞吐量，非必要不开启

5. ReentrantLock — 可重入锁​

是什么？​

ReentrantLock 是 synchronized 的增强替代品。它提供了与 synchronized 相同的互斥性和可重入性，但额外支持可中断等待、超时获取、公平锁、多条件变量等高级特性。

一句话定位：功能更强大、更灵活的 synchronized。

核心方法​

方法说明
lock()获取锁，获取不到就阻塞
unlock()释放锁，必须手动调用
tryLock()非阻塞尝试获取锁
tryLock(long timeout, TimeUnit unit)带超时的尝试获取
lockInterruptibly()可响应中断的获取锁
newCondition()创建条件变量（替代 wait/notify）
isHeldByCurrentThread()当前线程是否持有锁

与 synchronized 的对比​

维度synchronizedReentrantLock
获取锁自动获取手动 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() 的惊群效应。

⚠️ 注意事项​

  1. unlock 必须在 finally 中：这是最容易踩的坑，忘记释放锁就是死锁
  2. lock() 要写在 try 外面：如果写在 try 里面，lock() 抛异常时 finally 中的 unlock() 会释放一个未持有的锁
  3. 公平锁有代价：new ReentrantLock(true) 保证 FIFO，但吞吐量下降明显
  4. 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();
}

⚠️ 注意事项​

  1. 不支持锁升级：在持有读锁时尝试获取写锁会导致死锁
  2. 写饥饿问题：如果读操作持续不断，写线程可能长时间拿不到锁。可以用公平模式缓解
  3. 适用场景：读写比至少 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​

维度ReentrantReadWriteLockStampedLock
可重入✅❌
乐观读❌✅
Condition✅❌
读性能好极好（乐观读无锁）
使用难度简单较复杂（stamp 管理）

⚠️ 注意事项​

  1. 不可重入！ 在持有 StampedLock 的锁时再次获取同类型的锁会导致死锁
  2. 不支持 Condition
  3. 乐观读的模式必须严格遵循：先读到局部变量，再 validate，失败后降级
  4. stamp 不能传 0：validate(0) 永远返回 false
  5. 适合读密集场景：如果写操作频繁导致乐观读一直失败，性能反而不如 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 对比：

维度HashtablesynchronizedMapConcurrentHashMap
锁粒度整个 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);
    }
}

⚠️ 注意事项：

  1. 写操作很重：每次写都复制整个数组，数据量大时内存开销巨大
  2. 弱一致性：写入后，正在遍历的线程看到的可能还是旧数据
  3. 适用条件：读操作 >> 写操作（如 100:1 以上），且数据量不大

9.3 BlockingQueue 系列​

一句话定位：带阻塞能力的队列，是生产者-消费者模式的最佳搭档。

四组操作方法对比：

操作抛异常返回特殊值阻塞超时
插入add(e)offer(e) → falseput(e)offer(e, timeout, unit)
移除remove()poll() → nulltake()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, ...)多个 CFVoid等待所有任务完成
anyOf(cf1, cf2, ...)多个 CFObject任一任务完成即返回

💡 以上方法都有 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) {}
    }
}

⚠️ 注意事项​

  1. 不要使用默认线程池：supplyAsync() 默认使用 ForkJoinPool.commonPool()，它是全局共享的，线程数等于 CPU 核心数。IO 密集型任务应指定独立线程池
  2. 异常会被吞没：如果不处理异常（没有 exceptionally/handle/whenComplete），异常只有在 get()/join() 时才会抛出
  3. join() vs get()：join() 抛非受检异常，get() 抛受检异常（需要 try-catch）。推荐用 join()
  4. 避免嵌套 CompletableFuture：用 thenCompose 替代 thenApply + get()
  5. 记得关闭线程池：自定义线程池用完要 shutdown()

11. 横向对比与选型指南​

同步协调工具对比​

维度CountDownLatchCyclicBarrierSemaphore
可重用❌✅✅
参与者固定固定动态
等待模式一个等多个互相等待许可证控制
到达后操作无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 的详细解析，将在后续文章中单独展开。