并发工具

引言：为什么需要并发工具？

Java并发工具是java.util.concurrent包下的一系列高级工具，用于简化多线程编程中的线程协作、资源控制和同步操作。

想象一个建筑工地：

CountDownLatch（倒计时锁）：
- 像施工队长说"等3个队伍都到齐了才开工"
- 3个队伍陆续到达，每到达一个就倒计时减1
- 倒计时到0时，开工

CyclicBarrier（循环栅栏）：
- 像接力赛，每4个运动员到齐后一起起跑
- 第一轮结束后，可以继续用于第二轮

Semaphore（信号量）：
- 像停车场的闸门，只有2个许可
- 每进去一辆车，许可减1
- 许可用完时，其他车必须等待

CountDownLatch - 倒计时锁

CountDownLatch是一个一次性的同步工具，允许一个或多个线程等待其他线程完成操作。

核心概念

• 计数器：初始化时设置一个值
• 等待：调用await()的线程会阻塞，直到计数器为0
• 倒计时：其他线程调用countDown()，计数器减1
• 一次性：计数器到0后，不能重置

工作原理图

初始状态（count = 3）：
线程A ─┐
      ├─ await() → 阻塞
线程B ─┤
      │
线程C ─┘

线程D ─ countDown() → count = 2
线程E ─ countDown() → count = 1
线程F ─ countDown() → count = 0 → 解除阻塞，A、B、C继续执行

基本用法

import java.util.concurrent.CountDownLatch;

CountDownLatch latch = new CountDownLatch(3);

// 工作线程
for (int i = 1; i <= 3; i++) {
    final int taskId = i;
    new Thread(() -> {
        System.out.println("任务" + taskId + "开始执行...");
        try { Thread.sleep(100 * taskId); } catch (InterruptedException e) {}
        System.out.println("任务" + taskId + "执行完成！");
        latch.countDown();  // 计数减1
    }).start();
}

// 等待所有任务完成
latch.await();  // 阻塞，直到计数器为0
System.out.println("所有任务已完成！");

使用场景

超时等待

// await(timeout, unit)：等待指定时间
boolean success = latch.await(5, TimeUnit.SECONDS);
if (success) {
    System.out.println("所有任务完成");
} else {
    System.out.println("超时，部分任务可能未完成");
}

CyclicBarrier - 循环栅栏

CyclicBarrier是一个可循环使用的同步工具，让一组线程相互等待，直到所有线程都到达某个点后一起继续执行。

核心概念

• parties：参与的线程数量
• await()：线程到达时调用，等待其他线程
• cyclic：计数器可以重置，可以循环使用
• barrierAction：可选的屏障解除时的回调

工作原理图

初始状态（parties = 3）：
  线程A ─ await() → 到达，等待
  线程B ─ await() → 到达，等待
  线程C ─ await() → 到达，等待 → 3个都到达，屏障解除
      ↓
  所有线程继续执行
      ↓
  可重置，继续下一轮

基本用法

import java.util.concurrent.CyclicBarrier;

CyclicBarrier barrier = new CyclicBarrier(3, () -> {
    System.out.println("所有选手就位，发令枪响！");
});

for (int i = 1; i <= 3; i++) {
    final int playerId = i;
    new Thread(() -> {
        System.out.println("选手" + playerId + "准备中...");
        try { Thread.sleep(100 * playerId); } catch (InterruptedException e) {}
        System.out.println("选手" + playerId + "就位");
        barrier.await();  // 等待其他选手
        System.out.println("选手" + playerId + "开始跑！");
    }).start();
}

CyclicBarrier vs CountDownLatch

使用场景

Semaphore - 信号量

Semaphore用于控制同时访问资源的线程数量，常用于限流。

核心概念

• 许可（permits）：初始化时设置许可数量
• acquire()：获取一个许可，没有则等待
• release()：释放一个许可
• 公平性：可以设置为公平或非公平

工作原理图

初始状态（permits = 2）：
┌─────────────────────────────┐
│         停车场（2个车位）    │
└─────────────────────────────┘

车辆A ─ acquire() → 进入，permits = 1
车辆B ─ acquire() → 进入，permits = 0
车辆C ─ acquire() → 等待（没车位了）
车辆D ─ acquire() → 等待

车辆A ─ release() → 离开，permits = 1
车辆C ─ acquire() → 进入，permits = 0

基本用法

import java.util.concurrent.Semaphore;

Semaphore semaphore = new Semaphore(2);  // 只有2个许可

for (int i = 1; i <= 4; i++) {
    final int taskId = i;
    new Thread(() -> {
        try {
            System.out.println("任务" + taskId + "等待获取许可...");
            semaphore.acquire();  // 获取许可
            System.out.println("任务" + taskId + "开始执行！");
            Thread.sleep(500);  // 模拟执行
            System.out.println("任务" + taskId + "执行完成，释放许可");
            semaphore.release();  // 释放许可
        } catch (InterruptedException e) {}
    }).start();
}

使用场景

公平 vs 非公平

// 非公平（默认）：可能导致饥饿
Semaphore unfair = new Semaphore(2);

// 公平：先到先得（FIFO顺序）
Semaphore fair = new Semaphore(2, true);

// 公平semaphore的acquire会等前面排队的线程

获取和释放多个许可

Semaphore semaphore = new Semaphore(3);

// 获取多个许可
semaphore.acquire(2);  // 需要2个许可

// 释放多个许可
semaphore.release(2);  // 释放2个许可

Exchanger - 交换器

Exchanger用于两个线程之间交换数据。

基本用法

import java.util.concurrent.Exchanger;

Exchanger exchanger = new Exchanger<>();

new Thread(() -> {
    try {
        String data = "数据A";
        System.out.println("线程A准备交换: " + data);
        String received = exchanger.exchange(data);
        System.out.println("线程A收到: " + received);
    } catch (InterruptedException e) {}
}).start();

new Thread(() -> {
    try {
        String data = "数据B";
        System.out.println("线程B准备交换: " + data);
        String received = exchanger.exchange(data);
        System.out.println("线程B收到: " + received);
    } catch (InterruptedException e) {}
}).start();

使用场景

Phaser - 多阶段同步

Phaser是Java 7引入的更灵活的同步工具，支持多阶段任务的协调。

基本用法

import java.util.concurrent.Phaser;

Phaser phaser = new Phaser(3);  // 3个参与方

for (int i = 1; i <= 3; i++) {
    final int taskId = i;
    new Thread(() -> {
        System.out.println("任务" + taskId + "阶段1");
        phaser.arriveAndAwaitAdvance();  // 等待其他任务

        System.out.println("任务" + taskId + "阶段2");
        phaser.arriveAndAwaitAdvance();  // 等待其他任务

        System.out.println("任务" + taskId + "完成");
        phaser.arriveAndDeregister();  // 完成任务
    }).start();
}

对比总结

实战：模拟考试系统

import java.util.concurrent.*;

public class ExamSimulation {
    public static void main(String[] args) throws Exception {
        System.out.println("===== 学生考试场景 =====");

        int studentCount = 3;
        CountDownLatch startLatch = new CountDownLatch(1);  // 等发卷
        CountDownLatch submitLatch = new CountDownLatch(studentCount);  // 等交卷

        for (int i = 1; i <= studentCount; i++) {
            final int studentId = i;
            new Thread(() -> {
                try {
                    System.out.println("学生" + studentId + "等待发卷...");
                    startLatch.await();  // 等老师发卷
                    System.out.println("学生" + studentId + "开始答题...");
                    Thread.sleep((long)(Math.random() * 1000));  // 答题时间不同
                    System.out.println("学生" + studentId + "交卷！");
                    submitLatch.countDown();
                } catch (InterruptedException e) {}
            }).start();
        }

        Thread.sleep(500);  // 准备时间
        System.out.println("老师发卷！");
        startLatch.countDown();  // 发出试卷

        submitLatch.await();  // 等待所有学生交卷
        System.out.println("所有学生交卷，老师收卷！");
    }
}

实战：数据库连接池

import java.util.concurrent.*;

public class ConnectionPoolSimulation {
    public static void main(String[] args) {
        System.out.println("===== 数据库连接池场景 =====");

        int poolSize = 2;  // 连接池大小
        Semaphore semaphore = new Semaphore(poolSize);

        for (int i = 1; i <= 4; i++) {
            final int taskId = i;
            new Thread(() -> {
                try {
                    System.out.println("任务" + taskId + "等待获取连接...");
                    semaphore.acquire();
                    System.out.println("任务" + taskId + "获取到连接，开始执行SQL...");

                    Thread.sleep(1000);  // 模拟SQL执行
                    System.out.println("任务" + taskId + "执行完成，释放连接");

                    semaphore.release();
                } catch (InterruptedException e) {}
            }).start();
        }
    }
}

小结