Java 并发：先保证正确，再讨论吞吐

两个线程同时执行 count++，最后 count 可能比预期少。结论是：先保证正确，再讨论吞吐。

先正确、再吞吐：这篇把任务与完成信号 + 同步与不变量 + 虚拟线程与任务链梳理一遍，先明确共享状态和“完成”的定义，再选择线程与同步工具。

一、任务、线程和完成结果

任务、线程与完成信号的分工是：Runnable 描述做什么，Thread 承载执行，Future 表示是否完成。

调用 start() 会创建真正的并发执行；直接调用 run() 只是让当前线程同步执行方法，效果等同于普通方法调用。创建线程有两种基本写法：继承 Thread 类并实现 run 方法，或者实现 Runnable 接口后再传入 Thread。下面的 Task1 继承 Thread，Task2 实现 Runnable，两者都从 101 / 201 开始计数：

java29 行

public class Main {
    static class Task1 extends Thread {
        @Override
        public void run() {
            for (int i = 101; i <= 103; i++) {
                System.out.println("task1:" + i);
            }
        }
    }

    static class Task2 implements Runnable {
        @Override
        public void run() {
            for (int i = 201; i <= 203; i++) {
                System.out.println("task2:" + i);
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Task1 task1 = new Task1();
        Thread task2Thread = new Thread(new Task2());
        task1.start();
        task2Thread.start();
        task1.join();
        task2Thread.join();
        System.out.println("joined"); // joined
    }
}

运行后 task1 与 task2 的行交错出现，join 保证两个任务完成后才打印 joined。join 的含义是等待该线程执行完成，调用之后下一行代码只有在目标线程结束后才会继续。

生产代码通常不直接操作 Thread，而是通过 Executor（执行器） 提交任务，由它管理线程的生命周期。newFixedThreadPool(2) 表示执行器同时最多执行两个线程。Callable 与 Runnable 的区别是前者可以返回值，submit 返回的 Future 是异步结果与完成信号的容器：

java23 行

import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Main {
    static class Task implements Callable<String> {
        private final String name;
        Task(String name) { this.name = name; }
        @Override
        public String call() throws Exception {
            Thread.sleep(10);
            return "hello-" + name;
        }
    }

    public static void main(String[] args) throws Exception {
        try (var executor = Executors.newFixedThreadPool(2)) {
            Future<String> future = executor.submit(new Task("ada"));
            System.out.println(future.get()); // hello-ada
            System.out.println("done"); // done
        }
    }
}

get() 会阻塞到结果就绪，任务抛出的异常也会从 get() 冒出来，而不是悄悄地消失。Future 本身不是结果，而是“稍后会有结果”的承诺。用完还必须关闭执行器：ExecutorService 在 Java 19 起支持 try-with-resources，Java 17 需要显式 shutdown 并等待终止。这里有个容易漏掉的细节——关闭执行器只表示不再接受新任务，已经提交的任务是否成功，仍然要靠 Future 逐个确认。

一次提交多个 Callable 时，可以用 invokeAll 等待全部完成，用 invokeAny 只取最先完成的那个结果：

java25 行

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

public class Main {
    static Callable<String> task(String name, long sleepMs) {
        return () -> {
            Thread.sleep(sleepMs);
            return name;
        };
    }

    public static void main(String[] args) throws Exception {
        try (var executor = Executors.newFixedThreadPool(3)) {
            var tasks = List.of(task("a", 20), task("b", 20), task("c", 20));
            var futures = executor.invokeAll(tasks);
            for (var f : futures) {
                System.out.println(f.get()); // a b c，按任务顺序输出
            }
            String fastest = executor.invokeAny(
                    List.of(task("fast", 10), task("slow", 500)));
            System.out.println(fastest); // fast
        }
    }
}

注意：关闭执行器不等于任务成功：关闭只表示不再接受新任务，已经提交的任务是否成功，仍然要靠 Future 逐个确认。

二、同步保护的是不变量，不只是字段

以 count++ 为例：它是读取、计算、写回三个步骤，另一个线程在中间插入就会丢失更新。

count++ 的读取、计算和写回如果被另一条线程插入，就会丢失更新，这就是典型的竞态条件（race condition，结果依赖线程执行先后顺序）。修复它并不是“加个锁”这么简单，而是先问：哪个业务不变量需要被保护？

单个计数需要准确 → AtomicInteger，它把“读—改—写”变成不可分割的原子操作（atomic operation）。
余额和流水要同时变，任意一个失败都要回滚 → 需要把一组修改放进同一个临界区，用 synchronized 或 ReentrantLock（可重入锁）。
只是发布一个状态标志，不做复合操作 → 用 volatile，它保证特定读写之间的可见性与顺序保证，但不会把复合操作变成原子操作。对应规则见 JLS 17。

synchronized 与 ReentrantLock 的区别在能力而不在新旧：后者支持可中断等待、尝试加锁、超时和公平策略，代价是必须手动在 finally 中释放。选择哪种，由需要的语义决定，不该按“传统”或“现代”来挑。

工具选型可以简化成一张表：

要保护的是什么	常用工具	关键点
单个计数	AtomicInteger	原子自增，锁竞争开销低
一组关联字段同时变更	synchronized / ReentrantLock	临界区覆盖全部相关写操作
可见性、标志位	volatile	不做复合操作
下游并发上限	Semaphore	同时通过的许可数量

三、虚拟线程适合什么任务

虚拟线程适合大量等待 I/O 的任务，它不会增加 CPU 核心数。

Java 21 的**虚拟线程（virtual thread）**由 JVM 调度，适合大量等待 I/O 的任务：它把“阻塞等待”的成本从操作系统线程转移到 JVM。但它不会凭空增加 CPU 核心数，也不会扩大数据库连接池。内存、打开的文件句柄和下游服务能力仍然有限，不能承诺任意数量的线程都安全。Java 21 虚拟线程指南对调度与使用边界有完整说明。

对每个任务创建一个虚拟线程即可，不需要再为它建固定线程池。真正的限制点在下游，用 Semaphore（信号量，限制同时进入某段代码的数量） 控制；同时还要在入口限制待处理任务总量，避免大量等待任务堆在内存里：

java33 行

import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

public class Main {
    public static void main(String[] args) throws Exception {
        var permits = new Semaphore(4);
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var active = new java.util.concurrent.atomic.AtomicInteger();
            var peak = new java.util.concurrent.atomic.AtomicInteger();
            var results = new java.util.ArrayList<java.util.concurrent.Future<String>>();
            for (int i = 0; i < 12; i++) {
                results.add(executor.submit(() -> {
                    permits.acquire();
                    try {
                        int now = active.incrementAndGet();
                        peak.accumulateAndGet(now, Math::max);
                        Thread.sleep(10); // 演示等待逻辑
                        return "done";
                    } finally {
                        active.decrementAndGet();
                        permits.release();
                    }
                }));
            }
            for (var result : results) result.get();
            System.out.println("peak=" + peak.get()); // peak=4
            if (peak.get() > 4 || active.get() != 0) {
                throw new AssertionError("并发限制或资源释放失败");
            }
            System.out.println("ok"); // ok
        }
    }
}

示例核心是两个不变量：Semaphore 保证同时执行不超过 4，finally 保证无论异常与否都归还许可。并发代码最容易犯的错不是正常路径，而是异常路径漏掉 release()——这类泄漏通常在流量上来之后才出现，所以验证要同时覆盖正常与异常两种结束方式。

警告：异常路径最容易漏掉 release：Semaphore 许可必须放在 finally 里归还，这类泄漏通常在流量上来之后才出现，验证要同时覆盖正常与异常两种结束方式。

四、拒绝、超时和取消：一条任务链上的同一个决策

线程数、队列、拒绝策略和超时属于同一次设计，不是四个相互独立的配置。

固定平台线程池也可能挂着无界队列，线程数再大也挡不住积压。生产服务必须把线程数、队列容量、拒绝策略和超时放在一起设计：任务进不来时是排队、丢弃还是立即失败？排队超时后任务是否已经开始？允许取消吗？取消后会不会留下外部副作用（比如已经扣款、已写日志）？这些问题都必须在提交前有明确答案。

mermaid7 行

flowchart LR
A[提交任务] --> B[队列与限流]
B --> C{能执行吗}
C -->|许可不足| D[拒绝或等待超时]
C -->|拿到许可| E[执行任务]
E --> F[回收许可]
F --> G[确认 Future 结果]

捕获 InterruptedException 后，惯例是向上抛出，或恢复**中断标记（interrupt flag）**并结束任务。cancel(true) 发出的是中断请求而不是暴力终止，目标代码和底层的阻塞调用需要配合响应，才能把任务在安全位置停下来。

提示：cancel 发出的是中断请求：cancel(true) 不是暴力终止，目标代码和底层阻塞调用需要配合响应，才能把任务在安全位置停下来。

验证时同时检查最终结果、任务异常、队列积压和资源回收；测性能则要固定任务数量、等待与计算比例、以及下游限额。

五、收口：练习与小结

通过下面的练习检查是否掌握，再阅读小结收口。

练习

补全条件，让并发上限检查通过：

java7 行

public class Main {
    public static void main(String[] args) throws Exception {
        // TODO: 把 ___ 换成正确的比较，让“峰值不超过 4”时通过（提示：peak 是 AtomicInteger 记录的最大同时执行数）
        int peak = 4; // TODO：来自上文示例的 peak.get()
        System.out.println(peak <= 4 ? "测试通过" : "测试失败"); // 测试通过
    }
}

小结

先定共享状态与完成定义，再选线程与同步工具，线程数增加也可能让数据库、内存或 CPU 更早达到限制；
直接调 run() 不是并发，生产代码用 Executor 提交任务，用 Future 逐个确认已提交任务是否成功；
同步保护的是业务不变量，先问保护什么，再选 AtomicInteger、锁、volatile 或 Semaphore；
锁和许可都要在 finally 里释放，ReentrantLock 手动释放，Semaphore 许可同样在 finally 归还；
虚拟线程只适合大量等 I/O 的任务，它不增加 CPU 核心数，也不扩大连接池，下游用 Semaphore 控制并发；
线程数、队列、拒绝策略和超时一起设计，提交前明确排队、丢弃还是失败，以及取消后的外部副作用；
验证同时看结果、异常、积压与回收，正常与异常两种结束方式都要覆盖。

只有结果先正确，吞吐才有讨论价值。