14.2 同步、通信与死锁

完成线程创建与生命周期后，为本篇预留约两小时，并使用 Java 17 平台线程运行三个实验。目标是用一致的锁策略保护不变量，正确等待条件，并让多锁操作遵循稳定顺序。

两把锤子和同一块铁

线程安全也从规则开始。本页用三个可以复现的实验回答：怎样保护计数与可见性，怎样让生产者和消费者等待条件，以及怎样同时锁两个账户而不死锁。

1. 先写不变量，再找临界区

图书馆库存的规则是：

text

借阅前 stock > 0；借阅后 stock 恰好减少 1；stock 永不为负。

银行转账的规则是：

text

扣款与入账不可分割；总余额不变；任一余额不为负。

不变量是每次公开操作完成后都必须成立的条件。临界区是为了维持该条件而必须作为一个整体执行的代码。

2. `count++` 为什么会丢更新

java

final class UnsafeCounter {
    private int count;

    void increment() {
        count++;
    }
}

count++ 包含读取、计算、写回。可能出现：

text

线程 A 读取 100
线程 B 读取 100
线程 A 写回 101
线程 B 写回 101

两次调用只增加了 1。这是竞态条件。

不要把某个具体错误总数写成“预期输出”。未同步程序可能得到 19,973，也可能碰巧得到 20,000；JIT 优化、核心数和调度都会影响观察结果。即使跑一万次都正确，代码仍没有同步保证。

3. `synchronized` 的两项保证

java

final class Counter {
    private int value;

    synchronized void increment() {
        value++;
    }

    synchronized int value() {
        return value;
    }
}

实例同步方法获取 this 的监视器：

互斥：同一时刻只有一个线程能持有同一监视器；
可见性：释放某监视器 happens-before 之后另一个线程成功获取同一监视器。

getter 不是机械地“都要加锁”。准确规则是：所有参与同一不变量的读写都要遵循同一同步协议。若写在锁内，读却绕开锁且没有 join() 等其他发布关系，就缺少读取所需的可见性保证。

锁对象决定是否互斥

java

private final Object lock = new Object();

void increment() {
    synchronized (lock) {
        value++;
    }
}

代码	锁对象
`synchronized void f()`	`this`
`static synchronized void f()`	声明类的 `Class` 对象
`synchronized (lock)`	`lock` 表达式得到的对象

锁住两个不同对象不会互斥。私有锁对象不要泄露给调用者。内置锁可重入，同一线程能再次获得自己已持有的监视器；离开同步区域时会自动释放，包括异常退出。

临界区边界

java

boolean borrow() {
    synchronized (lock) {
        if (stock == 0) {
            return false;
        }
        stock--;
        return true;
    }
}

检查和修改必须在同一临界区。日志格式化等无关计算可以移出锁，但不能为了缩小锁而拆散业务原子性。持锁时不要调用远程服务、磁盘 I/O 或未知回调，否则延迟和死锁风险都会扩大。

实验一：安全计数与状态发布

java

import java.util.concurrent.CountDownLatch;

public class SynchronizationDemo {
    private static final class Counter {
        private int value;

        synchronized void increment() {
            value++;
        }

        synchronized int value() {
            return value;
        }
    }

    private static volatile boolean running = true;

    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter();
        Thread first = incrementer(counter);
        Thread second = incrementer(counter);
        first.start();
        second.start();
        first.join();
        second.join();

        CountDownLatch ready = new CountDownLatch(1);
        Thread watcher = new Thread(() -> {
            ready.countDown();
            while (running) {
                Thread.onSpinWait();
            }
        }, "status-watcher");
        watcher.start();
        ready.await();
        running = false;
        watcher.join();

        if (counter.value() != 20_000 || watcher.isAlive()) {
            throw new AssertionError("同步不变量失败");
        }
        System.out.println("安全计数：" + counter.value());
        System.out.println("停止标志：" + !running);
        System.out.println("同步检查通过");
    }

    private static Thread incrementer(Counter counter) {
        return new Thread(() -> {
            for (int i = 0; i < 10_000; i++) {
                counter.increment();
            }
        });
    }
}

运行：

bash

chapter14_sync_root=$(mktemp -d)
mkdir -p "$chapter14_sync_root/out"
# 将上面的代码保存为 "$chapter14_sync_root/SynchronizationDemo.java"
javac --release 17 -Xlint:all -Werror \
  -d "$chapter14_sync_root/out" \
  "$chapter14_sync_root/SynchronizationDemo.java"
java -ea -cp "$chapter14_sync_root/out" SynchronizationDemo
rm -rf -- "$chapter14_sync_root"

输出：

text

安全计数：20000
停止标志：true
同步检查通过

计数由监视器保护，两个 join() 又保证主线程在读取前等待工作完成。running 是独立的状态标志，写入 false happens-before 工作线程随后对同一 volatile 字段的读取。

示例中的忙等只为了隔离展示可见性；真实服务应避免空转占用核心，优先使用阻塞队列、闩锁或可中断阻塞操作。

4. `volatile` 能做什么

java

private volatile boolean running = true;

对 volatile 字段的写 happens-before 后续对同一字段的读。它常用于：

取消或状态标志；
发布一个构造完成、之后不再原地修改的快照引用；
某些经过严格设计的单变量状态机。

它不能保护：

java

volatile int count;
count++;                    // 复合读-改-写

if (!initialized) {         // 检查后执行
    initialize();
}

left -= amount;             // 多字段不变量
right += amount;

把 volatile 解释成“立刻刷新主内存”既不精确，也容易让人误以为它提供事务性。判断正确性时，应列出需要的 happens-before 边，而不是猜缓存何时刷新。

AtomicInteger.incrementAndGet() 能原子更新单个计数器，但“库存与借阅记录必须一致”仍跨越多个值，需要更高层的锁或事务。原子类也不保证在所有争用模式下都比 synchronized 快，选择应基于语义和测量。

5. Java 17 初学阶段常用的 happens-before 边

前一动作	后一动作
解锁某监视器	之后成功锁住同一监视器
写某 volatile 字段	之后读同一 volatile 字段
调用 `thread.start()` 前的动作	新线程中的动作
工作线程终止前的动作	另一线程从 `join()` 返回后的动作
向执行器提交任务前的动作	任务中的动作
任务中的动作	成功从对应 `Future.get()` 返回后的动作

这张表比“变量最终总会被看见”更有用：没有明确边，就不要依赖可见性。

6. 条件等待：队列满了怎么办

互斥只能让线程排队进入临界区，不能表达“现在即使拿到锁也没法继续”。生产者遇到满队列时，要等待“队列不满”；消费者遇到空队列时，要等待“队列非空”。

`wait()` 的准确语义

在 synchronized (monitor) 内调用 monitor.wait() 时：

调用者必须持有 monitor，否则抛 IllegalMonitorStateException；
它原子地释放 monitor 并进入等待，但不会释放当前线程持有的其他锁；
被通知、超时、中断或伪唤醒后，线程先重新竞争 monitor；
重新获得监视器后，wait() 才返回或抛出异常。

Thread.sleep() 不释放已持有的锁，两者不能互换。

条件必须用 while：

java

synchronized T take() throws InterruptedException {
    while (elements.isEmpty()) {
        wait();
    }
    T value = elements.removeFirst();
    notifyAll();
    return value;
}

原因有两个：允许伪唤醒；即使通知真实发生，另一个消费者也可能先抢到锁并取走元素。

notify() 选择任意等待者，不保证随机或公平。notifyAll() 唤醒所有等待者去竞争监视器，也不保证公平，但在多个条件共用一个监视器时更不容易叫醒“错误类别”后无人推进。

实验二：手写有界阻塞队列

这个实验用于理解监视器协议。实际应用应优先使用经过充分测试的 ArrayBlockingQueue 或 LinkedBlockingQueue。

java

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
import java.util.Objects;

public class BlockingQueueDemo {
    private static final class MyBlockingQueue<T> {
        private final Deque<T> elements = new ArrayDeque<>();
        private final int capacity;

        MyBlockingQueue(int capacity) {
            if (capacity <= 0) {
                throw new IllegalArgumentException("capacity 必须大于 0");
            }
            this.capacity = capacity;
        }

        synchronized void put(T value) throws InterruptedException {
            Objects.requireNonNull(value, "队列不接受 null");
            while (elements.size() == capacity) {
                wait();
            }
            elements.addLast(value);
            notifyAll();
        }

        synchronized T take() throws InterruptedException {
            while (elements.isEmpty()) {
                wait();
            }
            T value = elements.removeFirst();
            notifyAll();
            return value;
        }

        synchronized int size() {
            return elements.size();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        MyBlockingQueue<Integer> queue = new MyBlockingQueue<>(2);
        List<Integer> consumed = new ArrayList<>();

        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 5; i++) {
                    consumed.add(queue.take());
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "consumer");
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    queue.put(i);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "producer");

        consumer.start();
        producer.start();
        producer.join();
        consumer.join();

        if (!consumed.equals(List.of(1, 2, 3, 4, 5)) || queue.size() != 0) {
            throw new AssertionError("阻塞队列没有保持 FIFO 不变量");
        }
        System.out.println("消费顺序：" + consumed);
        System.out.println("最终大小：" + queue.size());
        System.out.println("阻塞队列检查通过");
    }
}

运行：

bash

chapter14_queue_root=$(mktemp -d)
mkdir -p "$chapter14_queue_root/out"
# 将上面的代码保存为 "$chapter14_queue_root/BlockingQueueDemo.java"
javac --release 17 -Xlint:all -Werror \
  -d "$chapter14_queue_root/out" \
  "$chapter14_queue_root/BlockingQueueDemo.java"
java -ea -cp "$chapter14_queue_root/out" BlockingQueueDemo
rm -rf -- "$chapter14_queue_root"

输出：

text

消费顺序：[1, 2, 3, 4, 5]
最终大小：0
阻塞队列检查通过

队列拒绝 null，否则 null 既可能表示元素，也容易与某些队列 API 的“没有结果”混淆。测试不靠休眠：两个 join() 建立完成关系，断言检查 FIFO 和最终为空；容量上限由 put 的条件循环保证。这些断言不能证明每次运行都曾发生阻塞。consumed 只有消费者写，主线程在 join 后才读，因此不需要并发 List。

这个教学队列只在 wait() 中响应中断；等待进入 synchronized 本身不可中断。捕获中断的任务恢复了状态，但本例没有额外取消协议。如果生产者提前退出，消费者可能永远等待。生产系统还要设计关闭方式，例如毒丸、关闭标志、取消全部任务，或使用支持超时的标准队列操作。

7. 死锁与稳定锁顺序

两个线程可能各持一把锁，并等待对方：

text

线程 A：持有账户 1，等待账户 2
线程 B：持有账户 2，等待账户 1

“同时锁两个账户”还不够。所有线程必须遵守相同的全局顺序。稳定唯一 id 是一种简单顺序；不要用可能相同且不稳定的业务余额排序。

实验三：按稳定顺序转账

java

import java.util.ArrayList;
import java.util.List;
import java.util.SplittableRandom;

public class BankTransferDemo {
    private static final class Account {
        private final int id;
        private int balance;

        Account(int id, int balance) {
            if (balance < 0) throw new IllegalArgumentException("Negative balance");
            this.id = id;
            this.balance = balance;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        List<Account> accounts = new ArrayList<>();
        for (int id = 0; id < 10; id++) {
            accounts.add(new Account(id, 1_000));
        }

        List<Thread> workers = new ArrayList<>();
        for (int workerId = 0; workerId < 5; workerId++) {
            int seed = workerId;
            Thread worker = new Thread(() -> transferBatch(accounts, seed), "transfer-" + workerId);
            workers.add(worker);
            worker.start();
        }
        for (Thread worker : workers) {
            worker.join();
        }

        int total = accounts.stream().mapToInt(account -> account.balance).sum();
        long negative = accounts.stream().filter(account -> account.balance < 0).count();
        if (total != 10_000 || negative != 0) {
            throw new AssertionError("转账不变量失败");
        }
        System.out.println("余额守恒：" + total);
        System.out.println("负余额账户：" + negative);
        System.out.println("转账检查通过");
    }

    private static void transferBatch(List<Account> accounts, int seed) {
        SplittableRandom random = new SplittableRandom(seed);
        for (int i = 0; i < 100; i++) {
            Account from = accounts.get(random.nextInt(accounts.size()));
            Account to = accounts.get(random.nextInt(accounts.size()));
            transfer(from, to, random.nextInt(1, 26));
        }
    }

    private static void transfer(Account from, Account to, int amount) {
        java.util.Objects.requireNonNull(from, "from");
        java.util.Objects.requireNonNull(to, "to");
        if (amount <= 0) throw new IllegalArgumentException("Non-positive amount");
        if (from == to) {
            return;
        }
        if (from.id == to.id) throw new IllegalArgumentException("Duplicate account id");
        Account first = from.id < to.id ? from : to;
        Account second = from.id < to.id ? to : from;
        synchronized (first) {
            synchronized (second) {
                if (from.balance >= amount) {
                    int targetBalance = Math.addExact(to.balance, amount);
                    from.balance -= amount;
                    to.balance = targetBalance;
                }
            }
        }
    }
}

运行：

bash

chapter14_transfer_root=$(mktemp -d)
mkdir -p "$chapter14_transfer_root/out"
# 将上面的代码保存为 "$chapter14_transfer_root/BankTransferDemo.java"
javac --release 17 -Xlint:all -Werror \
  -d "$chapter14_transfer_root/out" \
  "$chapter14_transfer_root/BankTransferDemo.java"
java -ea -cp "$chapter14_transfer_root/out" BankTransferDemo
rm -rf -- "$chapter14_transfer_root"

输出：

text

余额守恒：10000
负余额账户：0
转账检查通过

账户余额以整数分表示，构造时拒绝负值，转账拒绝非正金额和不同对象的重复 id；目标余额溢出时在扣款前失败。示例工厂只创建 id 为 0 至 9 的十个账户，总额为 10000，因此正常任务不会溢出。金额不足时不修改余额；这里演示同一进程内的互斥，不包含持久化事务。

随机种子让每个工作线程生成稳定的转账序列，但线程交错仍不固定。输出只检查不受调度影响的不变量。

在所有工作线程 join() 后，主线程可以读取余额。若系统要在转账进行中提供余额快照，读取也必须使用同一锁策略，或使用另一种能给出一致快照的设计。

其他死锁防线

不在持锁期间等待网络、文件或用户输入；
不在持锁期间调用未知外部回调；
用短小方法集中多锁顺序，不让调用者自行组合；
需要超时锁获取时使用 ReentrantLock.tryLock(timeout, unit)，需要可中断等待时也可用 lockInterruptibly()；
线上卡死时获取线程转储，寻找 BLOCKED 线程和互相持有的监视器。

8. 优先减少共享状态

加锁不是唯一答案：

问题	更简单的候选方案
数据创建后不再变化	不可变对象
数据只在一个工作线程使用	线程封闭
生产者把工作交给消费者	`BlockingQueue`
线程等待一次性事件	`CountDownLatch`
并发映射	`ConcurrentHashMap` 的原子组合方法
独立单值计数	`AtomicInteger`/`LongAdder`，先确认语义
简单复合不变量	`synchronized`
多条件、超时、可中断获取	`ReentrantLock`/`Condition`

LongAdder.sum() 在并发更新期间不是原子快照，不能直接代替需要精确读改写的余额。使用 ConcurrentHashMap 也不代表跨多个调用的逻辑自动原子化。优先使用 compute、merge、putIfAbsent 等单次原子组合，而不是先 containsKey 再 put。

9. 语言差异窗口

CPython

不能把 GIL 当作业务锁；“检查库存再扣减”仍需要整体同步，也不要把某个版本的 count += 1 调度表现当成保证。启用 GIL 时，I/O 和释放 GIL 的扩展允许并发；CPU 密集型并行常使用多进程或释放 GIL 的库。CPython 从 3.13 起另提供自由线程构建，可以禁用 GIL，但共享不变量仍需锁或队列。

C++

C++ 中，不同线程的冲突访问若至少一个不是原子操作，且没有 happens-before 顺序，就构成数据竞争并导致未定义行为。应使用 std::mutex 或满足所需内存顺序的原子类型。Java 的 volatile 不能直接等同于 C++ 的任意 std::atomic 用法。

10. 常见故障

现象	检查点
偶发少计数	是否所有读写都遵循同一锁协议
换成 `volatile int` 仍少计数	`++` 仍是复合操作
`wait()` 抛监视器异常	是否持有调用对象的监视器
通知后线程仍不动	是否通知了不同对象；条件是否可能仍不成立
用 `if` 等待后偶发越界	改为 `while` 重查条件
`sleep()` 期间别人拿不到锁	`sleep()` 不释放监视器
两笔反向转账卡住	多锁获取顺序是否全局一致
为缩小锁拆开检查与修改	临界区已破坏不变量

11. 练习

给 MyBlockingQueue 增加带超时的 offer 思路说明。先写清超时、伪唤醒和剩余等待时间如何处理，再决定是否实现。
把队列调用端改为 ArrayBlockingQueue<Integer>，比较需要维护的不变量数量。
给转账实验增加一致的余额快照方法。说明快照如何遵守相同锁顺序。
构造“锁 A 后锁 B”与“锁 B 后锁 A”的最小死锁示例，只在隔离进程运行；用线程转储定位循环等待，不把它放进自动化成功测试。

把三条不变量放回压力测试

计数实验要在所有写入和读取上遵守同一个监视器协议；把 int 换成 volatile int 仍不能让 ++ 成为原子操作。不要给不安全程序写一个固定的“错误答案”，它偶尔得到 20000 也没有建立正确性保证。

阻塞队列必须在持有同一监视器时用 while 检查条件。分别让生产者先到、消费者先到，并测试容量 1；输出顺序可以变化，FIFO、容量与最终为空的不变量不能变化。真实项目优先换成标准 BlockingQueue，手写版本只用于理解协议。

转账实验反复执行反向转账，并在所有线程 join 后检查总余额和非负余额。若要在运行中读取一致快照，读取也要取得全部相关锁，并沿相同的稳定顺序。锁顺序只在每条路径都遵守时才有意义。

查阅 Java 17 的正式规则

JLS 第 17 章：Threads and Locks：内存模型、数据竞争、锁和可见性。
JLS 14.19：synchronized 语句。
Object.wait/notifyAll：监视器所有权、唤醒与中断。
BlockingQueue：生产代码应优先采用的队列契约。

下一步：给工位和等候区设上限

同一个锁对象和同一套访问协议，才使并发操作遵守共同规则。监视器保护完整规则，volatile 发布单个状态变化，条件等待表达“现在还不能继续”。下一篇把任务交给长期工作的线程，并为等候区和过载行为写下明确合同。

继续学习线程池与任务结果

第1章逻辑

第2章集合论

第3章证明方法

第4章递归、归纳与递推

第5章关系与函数

第6章组合数学

第7章图论

第8章概率与统计

第9章数论

第10章线性代数

第11章微积分与优化

第12章信息论与数值计算

14.2 同步、通信与死锁

两把锤子和同一块铁

1. 先写不变量，再找临界区

2. `count++` 为什么会丢更新

3. `synchronized` 的两项保证

锁对象决定是否互斥

临界区边界

实验一：安全计数与状态发布

4. `volatile` 能做什么

5. Java 17 初学阶段常用的 happens-before 边

6. 条件等待：队列满了怎么办

`wait()` 的准确语义

实验二：手写有界阻塞队列

7. 死锁与稳定锁顺序

实验三：按稳定顺序转账

其他死锁防线

8. 优先减少共享状态

9. 语言差异窗口

CPython

C++

10. 常见故障

11. 练习

把三条不变量放回压力测试

查阅 Java 17 的正式规则

下一步：给工位和等候区设上限