从仓库、银行到图书馆：操作系统三大经典并发问题，如何走进 Java Web？
为什么电商订单需要消息队列？为什么资源够用也不一定能分配？为什么多个线程能一起读缓存，却不能随意同时修改？从生产者—消费者、银行家算法、读者—写者三个操作系统经典问题出发，以生活类比、PV 操作、伪代码和 C / Java / Python / C++ / JavaScript 五语言实现，理解它们如何迁移到 Spring Boot、任务调度、管理系统与高并发业务，并厘清 JVM 锁、数据库事务和分布式消息之间的边界。
学操作系统时，一个困惑经常出现：我明白了仓库有生产者和消费者，也会手写几个信号量，但这套模型究竟有什么用？ 以后写的是电商、库存管理、药店管理系统，又不是开仓库。
答案是：算法模型不是业务模板，而是一套识别并发矛盾的思维工具。 现实业务的名字会变化，但“任务来得太快”“资源分出去后可能不够用”“多人读写同一份数据”等问题会反复出现。能认出矛盾，才能选择合适的技术。
一、先认出问题：仓库、银行、图书馆
同样有很多人同时做事，却存在三类完全不同的矛盾：
| 生活中的故事 | 操作系统经典问题 | 一句话核心矛盾 | 在软件里首先想到 | | ---------------------------- | ---------------------------------------- | -------------------------------------- | -------------------------------- | | 仓库不断入货，员工不断取货 | 生产者—消费者（Producer–Consumer） | 活儿来得快，处理不过来怎么办？ | 缓冲队列、阻塞、背压、任务异步化 | | 银行同时给多人放贷 | 银行家算法（Banker's Algorithm） | 现在有资源，分出去后还能让大家完成吗？ | 安全状态、资源预留、死锁避免 | | 多人在图书馆读书，有人要改书 | 读者—写者（Readers–Writers） | 谁可以同时读，谁必须独占修改？ | 读写锁、共享状态、数据一致性 |
需要先厘清三个基础概念：
- 互斥（Mutual Exclusion）：同一时刻只让一个执行单元进入特定临界区，解决“不能同时改”的冲突。
- 同步（Synchronization）：某个操作必须等条件满足才能继续，解决“还没轮到你”的先后关系。
- 信号量（Semaphore）：维护许可证的同步工具。经典 P / wait 表示申请，V / signal 表示释放；没有可用许可证时，申请者需要等待。
它们与业务有联系，但不要机械套用：生产者—消费者和读者—写者是经典同步问题；银行家算法主要研究死锁避免，并不是生产消费模型的另一种写法。
阅读下面代码时，请以 Java 为主线，再使用语言标签对照 C、Python、C++、JavaScript：关注不同语言如何表达相同的约束，而不是背诵 15 份语法。
二、生产者—消费者：解决“任务产生和处理速度不一致”
1. 原理：有界仓库与三条约束
把容量为 N=3 的缓冲区看成仓库：生产者不断放入商品，消费者不断取走商品。
Producer A ──┐ ┌── Consumer 1
├─> [ 有界缓冲区 ] ┤
Producer B ──┘ 容量 N └── Consumer 2
如果仓库没有商品，消费者不能凭空取走；如果仓库已经装满，生产者不能强行塞入；多个线程操作同一个数组或队列时，又不能把下标、元素、数量修改到一半就被其他线程打乱。
因此需要三个信号量：
| 信号量 | 初始值 | 约束 |
| ------- | ------ | -------------------------------- |
| empty | N | 剩余多少个可用空位 |
| full | 0 | 已经有多少个可领取商品 |
| mutex | 1 | 同一时刻只能有一个线程修改缓冲区 |
在操作都完成、没有“已申请但尚未入队/出队”的过渡状态时，可以把状态理解为 empty + full = N。实际执行到两个 P/V 之间时，许可证可能被线程临时持有，不能用这条等式直接推断任意瞬间的内部计数。
关键点：同步与互斥缺一不可。 empty/full 负责等待可生产或可消费的条件；mutex 负责保护共享队列自身。
2. PV 伪代码：为什么顺序不能反？
producer():
repeat:
item = produce() // 在临界区外准备商品
P(empty) // 先等待仓库有位置
P(mutex) // 再独占队列
queue.push(item) // 临界区：修改共享数据
V(mutex) // 尽快退出临界区
V(full) // 通知：多了一件商品
consumer():
repeat:
P(full) // 先等待仓库有商品
P(mutex) // 再独占队列
item = queue.pop() // 临界区：修改共享数据
V(mutex) // 释放队列访问权
V(empty) // 通知：多了一个空位
consume(item) // 在临界区外处理商品
反例：先 P(mutex) 再 P(empty)。 假设仓库满了，生产者拿着队列锁等待空位；消费者虽然可以取货，却拿不到队列锁，双方互等，出现死锁。正确顺序是先等业务条件，再进入临界区。
从业务角度看，“生产”可以是 HTTP 接收请求、创建一个导入任务；“消费”则是后台线程发送通知、计算报表或处理文件，并不要求真的存在商品。
3. 五种语言：同一个有限缓冲区
下面都演示容量为 3、生产 8 个整数、消费 8 个整数。C 与 C++ 使用显式同步原语；Java 与 Python 用标准库阻塞队列；JavaScript 用 Promise 模拟有界异步队列。 它们实现同一业务约束，但底层并发机制并不完全相同。每个标签中的代码都可以单独运行。
::: code-group C|Java|Python|C++|JavaScript
// Linux / POSIX：gcc -std=c11 -pthread main.c -o pc
#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#define SIZE 3
#define TOTAL 8
int buffer[SIZE], head = 0, tail = 0, count = 0;
sem_t empty, full, mutex;
void *producer(void *arg) {
(void)arg;
for (int item = 1; item <= TOTAL; ++item) {
sem_wait(&empty); // P(empty)：满了就等待
sem_wait(&mutex); // P(mutex)：独占缓冲区
buffer[tail] = item;
tail = (tail + 1) % SIZE; // 环形队列，尾指针回绕
++count;
printf("生产 %d，库存 %d\n", item, count);
sem_post(&mutex); // V(mutex)：退出临界区
sem_post(&full); // V(full)：通知有商品
}
return NULL;
}
void *consumer(void *arg) {
(void)arg;
for (int i = 0; i < TOTAL; ++i) {
sem_wait(&full); // P(full)：空了就等待
sem_wait(&mutex);
int item = buffer[head];
head = (head + 1) % SIZE;
--count;
printf("消费 %d，库存 %d\n", item, count);
sem_post(&mutex);
sem_post(&empty); // V(empty)：通知有空位
}
return NULL;
}
int main(void) {
pthread_t p, c;
sem_init(&empty, 0, SIZE); // 初始 3 个空位
sem_init(&full, 0, 0); // 初始 0 件商品
sem_init(&mutex, 0, 1); // 二元信号量模拟互斥
pthread_create(&p, NULL, producer, NULL);
pthread_create(&c, NULL, consumer, NULL);
pthread_join(p, NULL);
pthread_join(c, NULL);
sem_destroy(&empty);
sem_destroy(&full);
sem_destroy(&mutex);
return 0;
}
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class ProducerConsumerDemo {
public static void main(String[] args) throws InterruptedException {
// 容量为 3 的线程安全阻塞队列：已封装 PV/锁等同步细节
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(3);
Thread producer = new Thread(() -> {
try {
for (int item = 1; item <= 8; item++) {
queue.put(item); // 满时阻塞，直到消费者取走元素
System.out.println("生产 " + item);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // 保留中断语义
}
}, "Producer");
Thread consumer = new Thread(() -> {
try {
for (int i = 0; i < 8; i++) {
int item = queue.take(); // 空时阻塞
System.out.println("消费 " + item);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "Consumer");
producer.start();
consumer.start();
producer.join(); // 等待两个线程结束
consumer.join();
}
}
# Python 3：标准库 Queue 自带线程安全及阻塞语义
import threading
from queue import Queue
buffer = Queue(maxsize=3) # 固定容量，防止无限积压
def producer():
for item in range(1, 9):
buffer.put(item) # 满时等待，直到有空位置
print(f"生产 {item}")
def consumer():
for _ in range(8):
item = buffer.get() # 空时等待，直到有商品
try:
print(f"消费 {item}")
finally:
buffer.task_done() # 声明该任务已处理完成
p = threading.Thread(target=producer, name="Producer")
c = threading.Thread(target=consumer, name="Consumer")
p.start()
c.start()
p.join()
c.join()
buffer.join() # 等待所有入队任务都 task_done()
// C++20：g++ -std=c++20 -pthread main.cpp -o pc
#include <iostream>
#include <mutex>
#include <queue>
#include <semaphore>
#include <thread>
std::queue<int> buffer;
std::mutex mutex; // 保护队列
std::counting_semaphore<3> emptySlots(3); // 初始 3 个空位
std::counting_semaphore<3> fullSlots(0); // 初始 0 件商品
void produce() {
for (int item = 1; item <= 8; ++item) {
emptySlots.acquire(); // P(empty)
{
std::lock_guard<std::mutex> lock(mutex);
buffer.push(item);
std::cout << "生产 " << item << '\n';
} // RAII 自动解锁
fullSlots.release(); // V(full)
}
}
void consume() {
for (int i = 0; i < 8; ++i) {
fullSlots.acquire(); // P(full)
{
std::lock_guard<std::mutex> lock(mutex);
int item = buffer.front();
buffer.pop();
std::cout << "消费 " << item << '\n';
}
emptySlots.release(); // V(empty)
}
}
int main() {
std::thread p(produce), c(consume);
p.join();
c.join();
}
// Node.js：这里协调异步任务，不是创建 OS 线程
class AsyncBoundedQueue {
constructor(capacity) {
this.capacity = capacity;
this.items = [];
this.waitingProducers = []; // 等空位的 Promise 唤醒函数
this.waitingConsumers = []; // 等商品的 Promise 唤醒函数
}
async put(item) {
while (this.items.length >= this.capacity) {
await new Promise(resolve => this.waitingProducers.push(resolve));
}
this.items.push(item);
this.waitingConsumers.shift()?.(); // 唤醒一个消费者
}
async take() {
while (this.items.length === 0) {
await new Promise(resolve => this.waitingConsumers.push(resolve));
}
const item = this.items.shift();
this.waitingProducers.shift()?.(); // 释放一个空位
return item;
}
}
const queue = new AsyncBoundedQueue(3);
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
async function producer() {
for (let item = 1; item <= 8; item++) {
await queue.put(item);
console.log('生产', item);
await delay(5);
}
}
async function consumer() {
for (let i = 0; i < 8; i++) {
const item = await queue.take();
console.log('消费', item);
await delay(10);
}
}
Promise.all([producer(), consumer()]).catch(console.error);
:::
读代码时锁定这四件事：
- 任务从哪里来？ C/Java/C++ 的生产线程、Python 的生产函数、JS 的异步生产函数。
- 谁保存待办任务？ 一个容量有限的
buffer/queue。 - 什么时候等待？ 满了等待
put，空了等待take，或显式执行信号量的 P 操作。 - 谁保证正确性？
Semaphore、BlockingQueue、Queue、mutex等各语言的同步机制。
Java ArrayBlockingQueue.put()、Python Queue.put() 会在满时等待；take()、get() 会在空时等待。这让业务代码不必再重复造一个三信号量仓库。Java 生产环境还应考虑中断、关闭、重试和线程池的拒绝策略。JavaScript 示例则要特别注意：它演示的是 Node.js 事件循环中的异步等待，不是多个 OS 线程在共享同一块内存；真正的 JS 多线程还涉及 Worker / worker_threads。
一个值得记住的工程原则： 队列缓解短时流量峰值，不会凭空增加处理能力。若每秒产生 1000 件任务而系统只能处理 600 件，长期看积压仍会增长；必须增加处理能力、限流或拒绝新任务。
三、银行家算法：解决“资源够用，却未必分得安全”
1. 原理：银行家为什么要考虑未来？
银行还有钱，不意味着任何人的借款申请都能马上批准。银行更关心：这次借款以后，能否安排一个合理的还款顺序，让所有客户最终满足需求、依次归还资源？
在操作系统中，每个进程事先声明自己所需的各类资源最大数量。系统分配资源前，会假装分配一次，再检查是否存在一个所有进程都能依次完成的顺序（安全序列）。
- 安全状态：至少存在一条安全序列，按该序列有能力满足所有进程的剩余需求。
- 不安全状态：找不到这样的保证；不等于已经死锁，但未来有可能发生死锁。
- 死锁：一组执行单元互相等待对方占有的资源，无法继续推进。
四个核心数据结构（设有 n 个进程、m 类资源）：
| 名称 | 形状 | 表示什么 |
| ------------------ | -------- | ------------------------------ |
| Available[m] | 一维数组 | 系统当前还剩多少可用资源 |
| Max[n][m] | 二维矩阵 | 每个进程事先声明的最大资源需求 |
| Allocation[n][m] | 二维矩阵 | 每个进程已经持有的资源 |
| Need[n][m] | 二维矩阵 | 每个进程以后还需要多少资源 |
公式是：
这是一个含有 5 个进程和 3 类资源 的经典例子（向量的三个分量分别代表资源 A、B、C）：
| 进程 | Allocation | Max | Need | | ---- | ---------- | ------- | ------- | | P0 | (0,1,0) | (7,5,3) | (7,4,3) | | P1 | (2,0,0) | (3,2,2) | (1,2,2) | | P2 | (3,0,2) | (9,0,2) | (6,0,0) | | P3 | (2,1,1) | (2,2,2) | (0,1,1) | | P4 | (0,0,2) | (4,3,3) | (4,3,1) |
当前剩余资源为 Available = [3,3,2]。例如 P1 只需要 [1,2,2]，因此当前就能让它完成。假设它完成后释放原本持有的 [2,0,0]，则模拟可用资源增加至 [5,3,2]。
沿着这条思路，我们能找到以下安全序列：
| 完成进程 | 完成前 Work | 归还 Allocation | 完成后 Work | | -------- | ----------- | --------------- | ----------- | | P1 | [3,3,2] | [2,0,0] | [5,3,2] | | P3 | [5,3,2] | [2,1,1] | [7,4,3] | | P4 | [7,4,3] | [0,0,2] | [7,4,5] | | P0 | [7,4,5] | [0,1,0] | [7,5,5] | | P2 | [7,5,5] | [3,0,2] | [10,5,7] |
所以 P1 → P3 → P4 → P0 → P2 是一条安全序列。这里 Work += Allocation，因为“拿到剩余 Need 再执行完并归还全部资源”的净变化，就是释放它原本持有的 Allocation。
2. 安全性检查与资源请求伪代码
safe_check():
Work = copy(Available)
Finish = [false] * n
sequence = []
while sequence.length < n:
find an unfinished process Pi with Need[i] <= Work
if no such Pi: return UNSAFE
Work = Work + Allocation[i] // 假设 Pi 执行完成
Finish[i] = true
append Pi to sequence
return SAFE(sequence)
request(Pi, Request):
if Request > Need[i]: reject_as_invalid()
if Request > Available: wait_for_resources()
// 仅仅"当前够用"还不够
Available -= Request
Allocation[i] += Request
Need[i] -= Request
if safe_check() == SAFE:
commit_allocation()
else:
undo_all_three_changes() // 回滚
reject_or_wait()
所有 <= 比较都是逐资源维度进行的，A 资源不能凭空代替 B 资源。
再看两笔具体请求：
- P4 请求
[3,3,0]：当前资源确实足够，试分配后Available=[0,0,2]，但没有任何进程的剩余需求能被它满足，因此不安全，回滚。 - P1 请求
[1,0,2]：试分配后Available=[2,3,0]，仍然可以按P1 → P3 → P4 → P0 → P2完成，所以批准。
这就是银行家算法的核心判断：资源数量够不够，与分配结果安不安全，是两个不同的问题。
3. 五种语言：同一资源矩阵、同一次试分配
下面五份独立程序共享同一组测试数据：先尝试 P4 的不安全请求，再尝试 P1 的安全请求，方便比较不同语言的二维数组、循环、状态回滚实现。
::: code-group C|Java|Python|C++|JavaScript
// C11：gcc -std=c11 main.c -o banker
#include <stdio.h>
#define N 5 // 进程数
#define M 3 // 资源类型数
int allocation[N][M] = {
{0,1,0}, {2,0,0}, {3,0,2}, {2,1,1}, {0,0,2}
};
int maximum[N][M] = {
{7,5,3}, {3,2,2}, {9,0,2}, {2,2,2}, {4,3,3}
};
int available[M] = {3,3,2};
int need[N][M];
// 安全性算法：将找到的安全序列写入 sequence
int safe(int sequence[N]) {
int work[M], finish[N] = {0}, done = 0;
for (int j = 0; j < M; j++) work[j] = available[j];
while (done < N) {
int progressed = 0;
for (int i = 0; i < N; i++) {
if (finish[i]) continue;
int feasible = 1;
for (int j = 0; j < M; j++) {
if (need[i][j] > work[j]) feasible = 0;
}
if (!feasible) continue;
// 假设 Pi 执行完毕，归还原来持有的资源
for (int j = 0; j < M; j++) work[j] += allocation[i][j];
finish[i] = 1;
sequence[done++] = i;
progressed = 1;
}
if (!progressed) return 0; // 没有安全序列
}
return 1;
}
int request_resources(int pid, const int req[M]) {
for (int j = 0; j < M; j++) {
if (req[j] < 0 || req[j] > need[pid][j] ||
req[j] > available[j]) return 0;
}
// 试分配：先更新三个矩阵/向量
for (int j = 0; j < M; j++) {
available[j] -= req[j];
allocation[pid][j] += req[j];
need[pid][j] -= req[j];
}
int seq[N];
if (safe(seq)) {
printf("批准 P%d，安全序列：", pid);
for (int i = 0; i < N; i++) printf("P%d ", seq[i]);
putchar('\n');
return 1;
}
// 不安全：执行相反操作，完整回滚
for (int j = 0; j < M; j++) {
available[j] += req[j];
allocation[pid][j] -= req[j];
need[pid][j] += req[j];
}
printf("拒绝 P%d：试分配后不安全，已回滚\n", pid);
return 0;
}
int main(void) {
for (int i = 0; i < N; i++)
for (int j = 0; j < M; j++)
need[i][j] = maximum[i][j] - allocation[i][j];
int seq[N];
printf("初始状态：%s\n", safe(seq) ? "安全" : "不安全");
int unsafe_request[M] = {3,3,0};
int safe_request[M] = {1,0,2};
request_resources(4, unsafe_request); // 资源够，但不安全
request_resources(1, safe_request); // 安全，可以批准
return 0;
}
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
public class BankerDemo {
static class Banker {
private final int[][] allocation = {
{0,1,0}, {2,0,0}, {3,0,2}, {2,1,1}, {0,0,2}
};
private final int[][] max = {
{7,5,3}, {3,2,2}, {9,0,2}, {2,2,2}, {4,3,3}
};
private final int[] available = {3,3,2};
private final int[][] need = new int[5][3];
Banker() {
// Need = Max - Allocation
for (int i = 0; i < 5; i++)
for (int j = 0; j < 3; j++)
need[i][j] = max[i][j] - allocation[i][j];
}
// 返回安全序列；null 代表当前没有找到安全序列
synchronized List<Integer> safeSequence() {
int[] work = available.clone(); // 仅模拟，不改实际资源
boolean[] finish = new boolean[5];
List<Integer> sequence = new ArrayList<>();
while (sequence.size() < 5) {
boolean progress = false;
for (int i = 0; i < 5; i++) {
if (finish[i]) continue;
boolean enough = true;
for (int j = 0; j < 3; j++) {
if (need[i][j] > work[j]) enough = false;
}
if (!enough) continue;
// 假设 Pi 完成，释放 Allocation[i]
for (int j = 0; j < 3; j++)
work[j] += allocation[i][j];
finish[i] = true;
sequence.add(i);
progress = true;
}
if (!progress) return null;
}
return sequence;
}
synchronized boolean request(int pid, int[] req) {
if (pid < 0 || pid >= 5 || req == null || req.length != 3)
throw new IllegalArgumentException("请求参数不合法");
for (int j = 0; j < 3; j++) {
if (req[j] < 0 || req[j] > need[pid][j])
throw new IllegalArgumentException("超过合法需求");
if (req[j] > available[j]) return false; // 当前不足
}
// 试分配
for (int j = 0; j < 3; j++) {
available[j] -= req[j];
allocation[pid][j] += req[j];
need[pid][j] -= req[j];
}
List<Integer> seq = safeSequence();
if (seq != null) {
System.out.println("批准 P" + pid + "，安全序列=" + seq);
return true;
}
// 不安全就撤销试分配
for (int j = 0; j < 3; j++) {
available[j] += req[j];
allocation[pid][j] -= req[j];
need[pid][j] += req[j];
}
System.out.println("拒绝 P" + pid + "：不安全，已回滚");
return false;
}
}
public static void main(String[] args) {
Banker banker = new Banker();
System.out.println("初始安全序列=" + banker.safeSequence());
banker.request(4, new int[]{3,3,0}); // 不安全
banker.request(1, new int[]{1,0,2}); // 安全
}
}
# Python 3：二维列表直接表达银行家算法的资源矩阵
class Banker:
def __init__(self):
self.alloc = [[0, 1, 0], [2, 0, 0], [3, 0, 2],
[2, 1, 1], [0, 0, 2]]
maximum = [[7, 5, 3], [3, 2, 2], [9, 0, 2],
[2, 2, 2], [4, 3, 3]]
self.available = [3, 3, 2]
# Need = Max - Allocation
self.need = [[maximum[i][j] - self.alloc[i][j]
for j in range(3)] for i in range(5)]
def safe_sequence(self):
work = self.available.copy() # 模拟资源，不污染真实状态
finish = [False] * 5
sequence = []
while len(sequence) < 5:
progress = False
for i in range(5):
if finish[i]:
continue
if not all(self.need[i][j] <= work[j] for j in range(3)):
continue
# 假设进程执行完毕，归还先前占有的资源
work = [work[j] + self.alloc[i][j] for j in range(3)]
finish[i] = True
sequence.append(i)
progress = True
if not progress:
return None # 没有安全序列
return sequence
def request(self, pid, req):
if not 0 <= pid < 5 or len(req) != 3:
raise ValueError("请求参数不合法")
if any(req[j] < 0 or req[j] > self.need[pid][j]
for j in range(3)):
raise ValueError("请求超出声明的需求")
if any(req[j] > self.available[j] for j in range(3)):
return False # 当前资源不够
# 试分配；如果最终不安全，必须逐项回滚
for j in range(3):
self.available[j] -= req[j]
self.alloc[pid][j] += req[j]
self.need[pid][j] -= req[j]
seq = self.safe_sequence()
if seq is not None:
print(f"批准 P{pid}，安全序列={seq}")
return True
for j in range(3):
self.available[j] += req[j]
self.alloc[pid][j] -= req[j]
self.need[pid][j] += req[j]
print(f"拒绝 P{pid}：不安全，已回滚")
return False
banker = Banker()
print("初始安全序列=", banker.safe_sequence())
banker.request(4, [3, 3, 0]) # 资源够，但不安全
banker.request(1, [1, 0, 2]) # 安全，可以批准
// C++17：g++ -std=c++17 main.cpp -o banker
#include <array>
#include <iostream>
#include <optional>
#include <vector>
constexpr int N = 5, M = 3;
using Row = std::array<int, M>;
class Banker {
std::array<Row, N> alloc{{
Row{0,1,0}, Row{2,0,0}, Row{3,0,2}, Row{2,1,1}, Row{0,0,2}
}};
std::array<Row, N> maximum{{
Row{7,5,3}, Row{3,2,2}, Row{9,0,2}, Row{2,2,2}, Row{4,3,3}
}};
std::array<Row, N> need{};
Row available{3,3,2};
public:
Banker() {
for (int i = 0; i < N; ++i)
for (int j = 0; j < M; ++j)
need[i][j] = maximum[i][j] - alloc[i][j];
}
// optional为空 => 无安全序列
std::optional<std::vector<int>> safeSequence() const {
Row work = available;
std::array<bool, N> finish{};
std::vector<int> seq;
while (seq.size() < N) {
bool progress = false;
for (int i = 0; i < N; ++i) {
if (finish[i]) continue;
bool enough = true;
for (int j = 0; j < M; ++j)
if (need[i][j] > work[j]) enough = false;
if (!enough) continue;
// 模拟 Pi 完成并归还当前持有的资源
for (int j = 0; j < M; ++j) work[j] += alloc[i][j];
finish[i] = true;
seq.push_back(i);
progress = true;
}
if (!progress) return std::nullopt;
}
return seq;
}
bool request(int pid, Row req) {
if (pid < 0 || pid >= N) return false;
for (int j = 0; j < M; ++j)
if (req[j] < 0 || req[j] > need[pid][j] ||
req[j] > available[j]) return false;
// 先试分配
for (int j = 0; j < M; ++j) {
available[j] -= req[j];
alloc[pid][j] += req[j];
need[pid][j] -= req[j];
}
auto seq = safeSequence();
if (seq) {
std::cout << "批准 P" << pid << "，安全序列：";
for (int p : *seq) std::cout << 'P' << p << ' ';
std::cout << '\n';
return true;
}
// 回滚试分配
for (int j = 0; j < M; ++j) {
available[j] += req[j];
alloc[pid][j] -= req[j];
need[pid][j] += req[j];
}
std::cout << "拒绝 P" << pid << "：不安全，已回滚\n";
return false;
}
};
int main() {
Banker banker;
std::cout << "初始状态："
<< (banker.safeSequence() ? "安全" : "不安全") << '\n';
banker.request(4, Row{3,3,0});
banker.request(1, Row{1,0,2});
}
// Node.js / 浏览器均可运行：银行家算法是纯计算，无需线程
class Banker {
constructor() {
this.alloc = [[0,1,0],[2,0,0],[3,0,2],[2,1,1],[0,0,2]];
const max = [[7,5,3],[3,2,2],[9,0,2],[2,2,2],[4,3,3]];
this.available = [3,3,2];
// Need = Max - Allocation
this.need = max.map((row, i) => row.map((n, j) => n - this.alloc[i][j]));
}
safeSequence() {
const work = [...this.available]; // 复制：不会影响真实可用量
const finished = Array(5).fill(false);
const sequence = [];
while (sequence.length < 5) {
let progress = false;
for (let i = 0; i < 5; i++) {
if (finished[i]) continue;
if (!this.need[i].every((n, j) => n <= work[j])) continue;
// 假设进程完成，回收原来已分配的资源
this.alloc[i].forEach((n, j) => { work[j] += n; });
finished[i] = true;
sequence.push(i);
progress = true;
}
if (!progress) return null;
}
return sequence;
}
request(pid, req) {
if (!Number.isInteger(pid) || pid < 0 || pid >= 5 || req.length !== 3)
throw new Error('请求参数不合法');
if (req.some((n, j) => !Number.isInteger(n) || n < 0 || n > this.need[pid][j]))
throw new Error('请求超出声明的需求');
if (req.some((n, j) => n > this.available[j])) return false;
for (let j = 0; j < 3; j++) {
this.available[j] -= req[j];
this.alloc[pid][j] += req[j];
this.need[pid][j] -= req[j];
}
const sequence = this.safeSequence();
if (sequence !== null) {
console.log(`批准 P${pid}，安全序列=${sequence}`);
return true;
}
// 不安全：撤销试分配
for (let j = 0; j < 3; j++) {
this.available[j] += req[j];
this.alloc[pid][j] -= req[j];
this.need[pid][j] += req[j];
}
console.log(`拒绝 P${pid}：不安全，已回滚`);
return false;
}
}
const banker = new Banker();
console.log('初始安全序列=', banker.safeSequence());
banker.request(4, [3,3,0]); // 不安全
banker.request(1, [1,0,2]); // 安全
:::
真正需要理解的不是语法，而是三个状态变化：
Available[j] -= Request[j]
Allocation[i][j] += Request[j]
Need[i][j] -= Request[j]
这三行是试分配。安全就保留，不安全就做相反操作回滚。每次模拟安全序列都必须复制 Available 到 Work，不能把模拟检查误当成真实资源发放。
安全性算法在这种逐进程扫描的直接实现中，最坏时间复杂度为 O(n²m)。以上代码是教学模型，默认资源需求数据已经合法，并未实现真实资源分配器所需的进程注册、释放、等待队列、异常退出和完整输入校验。
它为什么不常直接出现在电商系统里？因为它要求任务提前声明最大资源需求，而真实 Web 请求的完整资源需求通常难以准确预知。因此电商一般使用线程池上限、连接池、配额、限流和数据库事务等措施管理资源；这些措施与银行家算法共享“不能无限分配”的思想，却不是银行家算法的直接实现。
四、读者—写者：解决“可以多人看，不能一起乱改”
1. 原理：读共享、写独占
想象一份药品分类配置，许多用户同时查询它；管理员偶尔修改它：
| 同时发生的操作 | 允许吗？ | 理由 | | -------------- | --------- | -------------------------------- | | 读者 + 读者 | ✅ 允许 | 都只读，不改变共享状态 | | 读者 + 写者 | ❌ 不允许 | 读取时可能看到修改中的不一致状态 | | 写者 + 写者 | ❌ 不允许 | 两次修改可能互相覆盖、破坏约束 |
若只用一把普通互斥锁包住所有读操作，数据虽安全，但读者也被迫排队。如果读操作很多、写操作较少，这会损失本来可以并发的读取能力。
经典读者优先解法使用 readCount 与两个互斥信号量：
countMutex：保护“正在读的人数”的计数。resource：第一个读者申请，最后一个读者释放；写者必须独占。
2. PV 伪代码：第一个读者进门，最后一个读者关门
reader():
P(countMutex)
readCount++
if readCount == 1:
P(resource) // 第一个读者挡住写者
V(countMutex)
read_shared_data() // 其他读者也可以进入
P(countMutex)
readCount--
if readCount == 0:
V(resource) // 最后一个读者放行写者
V(countMutex)
writer():
P(resource) // 必须单独拿到资源
modify_shared_data()
V(resource)
潜在问题：写者饥饿。 如果读者持续涌入，计数器可能长期不归零，写者一直进不去。所以真实并发工具还有公平锁、写者优先等策略：它们都保证读写互斥，但谁先获得执行权不一定完全相同。
3. 五种语言：读写锁的不同实现
这组示例都演示多个读取任务与一次修改操作，但实现层级有所不同：C 的 pthread_rwlock_t、Java 的公平 ReentrantReadWriteLock、Python 用 Condition 自己实现写者优先、C++ 的 shared_mutex、JavaScript 用 Promise 队列协调异步读写。不同标准库对公平性和饥饿没有统一保证，不要把其中一个语言的唤醒顺序推断为所有语言都相同。
::: code-group C|Java|Python|C++|JavaScript
// Linux / POSIX：gcc -std=c11 -pthread main.c -o rw
#define _POSIX_C_SOURCE 200809L
#include <pthread.h>
#include <stdio.h>
#include <time.h>
pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;
int document = 0; // 共享文档
void pause_briefly(void) {
struct timespec ts = {.tv_sec = 0, .tv_nsec = 30000000};
nanosleep(&ts, NULL);
}
void *reader(void *arg) {
int id = *(int *)arg;
pthread_rwlock_rdlock(&rwlock); // 读锁：可与其他读者共享
printf("R%d 读取到 %d\n", id, document);
pause_briefly(); // 模拟长时间读取
pthread_rwlock_unlock(&rwlock);
return NULL;
}
void *writer(void *arg) {
int value = *(int *)arg;
pthread_rwlock_wrlock(&rwlock); // 写锁：独占，阻止读/写
document = value;
printf("W 写入 %d\n", document);
pause_briefly();
pthread_rwlock_unlock(&rwlock);
return NULL;
}
int main(void) {
pthread_t r1, r2, w, r3;
int ids[] = {1, 2, 3}, value = 100;
pthread_create(&r1, NULL, reader, &ids[0]);
pthread_create(&r2, NULL, reader, &ids[1]);
pthread_create(&w, NULL, writer, &value);
pthread_create(&r3, NULL, reader, &ids[2]);
pthread_join(r1, NULL);
pthread_join(r2, NULL);
pthread_join(w, NULL);
pthread_join(r3, NULL);
pthread_rwlock_destroy(&rwlock);
return 0;
}
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ReadersWritersDemo {
private int document = 0;
// true：启用公平排队策略，减少长期饥饿的风险
private final ReentrantReadWriteLock rw = new ReentrantReadWriteLock(true);
public void read() {
rw.readLock().lock(); // 允许多个线程同时持有读锁
try {
System.out.println(Thread.currentThread().getName()
+ " 读取到 " + document);
Thread.sleep(30); // 演示同时读取
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
rw.readLock().unlock(); // 必须释放，防止写者永远等待
}
}
public void write(int value) {
rw.writeLock().lock(); // 独占：排斥所有读者和其他写者
try {
document = value;
System.out.println("W 写入 " + document);
} finally {
rw.writeLock().unlock();
}
}
public static void main(String[] args) throws InterruptedException {
ReadersWritersDemo demo = new ReadersWritersDemo();
Thread r1 = new Thread(demo::read, "R1");
Thread r2 = new Thread(demo::read, "R2");
Thread w = new Thread(() -> demo.write(100), "W");
Thread r3 = new Thread(demo::read, "R3");
r1.start(); r2.start(); w.start(); r3.start();
r1.join(); r2.join(); w.join(); r3.join();
}
}
# Python 3：Condition 实现“写者等待时阻止新读者进入”
import threading
import time
class ReadWriteLock:
def __init__(self):
self.cond = threading.Condition()
self.readers = 0
self.writer_active = False
self.waiting_writers = 0
def acquire_read(self):
with self.cond:
# 有写者正在执行或已经等待，新读者暂时等待
while self.writer_active or self.waiting_writers > 0:
self.cond.wait()
self.readers += 1
def release_read(self):
with self.cond:
self.readers -= 1
if self.readers == 0:
self.cond.notify_all() # 最后一个读者唤醒等待者
def acquire_write(self):
with self.cond:
self.waiting_writers += 1
try:
while self.writer_active or self.readers > 0:
self.cond.wait()
self.writer_active = True # 写者独占
finally:
self.waiting_writers -= 1
def release_write(self):
with self.cond:
self.writer_active = False
self.cond.notify_all() # 通知等待的读者和写者
rw = ReadWriteLock()
document = 0
def read(name):
rw.acquire_read()
try:
print(f"{name} 读取到 {document}")
time.sleep(0.03)
finally:
rw.release_read()
def write(value):
global document
rw.acquire_write()
try:
document = value
print(f"W 写入 {document}")
finally:
rw.release_write()
threads = [
threading.Thread(target=read, args=("R1",)),
threading.Thread(target=read, args=("R2",)),
threading.Thread(target=write, args=(100,)),
threading.Thread(target=read, args=("R3",)),
]
for t in threads:
t.start()
for t in threads:
t.join()
// C++17：g++ -std=c++17 -pthread main.cpp -o rw
#include <chrono>
#include <iostream>
#include <mutex>
#include <shared_mutex>
#include <thread>
std::shared_mutex rwMutex; // 多读、单写的共享互斥量
std::mutex outputMutex; // 只用于保护控制台输出
int document = 0;
void readDocument(int id) {
std::shared_lock<std::shared_mutex> lock(rwMutex); // 共享读锁
{
std::lock_guard<std::mutex> out(outputMutex);
std::cout << "R" << id << " 读取到 " << document << '\n';
}
std::this_thread::sleep_for(std::chrono::milliseconds(30));
} // RAII：退出作用域时释放读锁
void writeDocument(int value) {
std::unique_lock<std::shared_mutex> lock(rwMutex); // 独占写锁
document = value;
{
std::lock_guard<std::mutex> out(outputMutex);
std::cout << "W 写入 " << document << '\n';
}
} // RAII：释放写锁
int main() {
std::thread r1(readDocument, 1), r2(readDocument, 2);
std::thread w(writeDocument, 100), r3(readDocument, 3);
r1.join(); r2.join(); w.join(); r3.join();
}
// Node.js：串行化异步读写操作；并非共享内存线程锁
class AsyncRWLock {
constructor() {
this.readers = 0;
this.writing = false;
this.waiters = []; // 按请求到达顺序排队
}
acquireRead() { return this.enqueue('read'); }
acquireWrite() { return this.enqueue('write'); }
enqueue(kind) {
return new Promise(resolve => {
this.waiters.push({ kind, resolve });
this.drain();
});
}
drain() {
// 写操作尚未结束，不能批准任何新任务
if (this.writing) return;
// 如果已有读者，仍可批准队首连续的读者；
// 但绝不能跨过排队中的写者
if (this.readers > 0) {
while (this.waiters[0]?.kind === 'read') {
const job = this.waiters.shift();
this.readers++;
job.resolve(() => { this.readers--; this.drain(); });
}
return;
}
if (this.waiters[0]?.kind === 'write') {
const job = this.waiters.shift();
this.writing = true;
job.resolve(() => { // 返回一个“释放写锁”的函数
this.writing = false;
this.drain();
});
} else {
// 合并队首连续的读者，让他们并发执行
while (this.waiters[0]?.kind === 'read') {
const job = this.waiters.shift();
this.readers++;
job.resolve(() => { // 返回“释放读锁”的函数
this.readers--;
this.drain();
});
}
}
}
}
const rw = new AsyncRWLock();
let document = 0;
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
async function read(name) {
const unlock = await rw.acquireRead();
try {
console.log(name, '读取到', document);
await delay(30);
} finally {
unlock();
}
}
async function write(value) {
const unlock = await rw.acquireWrite();
try {
document = value;
console.log('W 写入', document);
await delay(10);
} finally {
unlock();
}
}
Promise.all([read('R1'), read('R2'), write(100), read('R3')])
.catch(console.error);
:::
抓住三个核心机制：
- 读共享： 多个读者可以同时拥有访问权限，因此读取密集时比“一把互斥锁串行所有人”更合适。
- 写独占： 写者必须等现有读者退出，并阻止其他写者与自己重叠。
- 公平与饥饿： 读者优先可能饿死写者；写者优先可能使读者长期等待；公平策略力求减少长期饥饿，但通常会增加排队或调度成本。
特别注意 JavaScript 示例：await 可让多个逻辑上的读取任务交错等待，不代表同一 Node.js 事件循环里的 JavaScript 代码正在 CPU 上多线程并行。C、Java、Python、C++ 示例涉及线程；而 Python 的 GIL 也不意味着可以省略所有共享状态同步，尤其在 I/O 等待时。
五、从操作系统抽象，走向 Java Web 真实业务
学习这三个模型，最终需要建立的是业务矛盾 → 识别模型 → 选用工程工具的推理链，而不是把伪代码一字不改地塞进 ServiceImpl。
1. 电商订单：生产者—消费者怎样落地？
用户在 Vue 页面点击“下单”，请求经过 Spring MVC Controller、Service、Mapper，到 MySQL 创建订单。然后可能需要发送通知、计算积分、触发履约任务。哪些事情必须同步完成，哪些可以异步？
Vue / Axios
│ POST /orders
▼
Spring MVC Controller
│
▼
OrderService ───→ MySQL 创建订单（关键事务）
│
├──→ 可靠记录待发布事件（例如事务 Outbox）
│
└──→ 返回下单结果（按业务接口契约）
│
▼
事件发布器 / MQ
│
[消息队列：缓冲]
│
┌────────┴────────┐
▼ ▼
通知服务 积分服务
消费事件 消费事件
角色对应：订单业务产生事件 = 生产者；消息中间件 = 队列；通知/积分服务 = 消费者。 常见工具是 BlockingQueue、ThreadPoolExecutor、Spring 异步执行，以及跨服务使用的 RabbitMQ/Kafka 等。
不过，普通内存队列不等于可靠消息系统：服务重启可能丢任务、消费者可能重复处理。订单与事件要避免“数据库已提交，但消息没发出去”的不一致，通常需要事务 Outbox 等可靠投递设计；消费者也应考虑幂等、重试、失败处理与积压监控。消息队列本身也不能代替数据库的库存一致性控制。
2. 管理系统：批量导入、文件处理与邮件通知
- 药店管理系统：管理员一次上传 5 万条药品数据。HTTP 接口完成鉴权、文件保存和导入任务登记；后台工作线程分批解析、校验、入库，页面查询任务进度。这就是典型的“提交任务—后台消费任务”。
- 知识平台/博客系统：文章发布后生成搜索索引、清理缓存、通知订阅者；正文发布作为核心事务，耗时的派生操作放到后台任务链路。
- 商品图片服务：用户上传原图是生产事件；压缩、生成缩略图、审核是消费任务。工作线程数、队列容量和重试机制需要按磁盘与 CPU 能力设计。
要谨慎区分：异步只改变完成时机，不天然保证事务正确、处理成功或结果顺序。
3. 资源管理：银行家算法带来的“先检查再分配”意识
- AI 推理任务平台：GPU 显存有限，任务请求 GPU 配额。可以先估算需求、排队、做准入控制，并在任务结束后释放配额。如果任务的最大需求已知且条件满足，才可能使用接近银行家算法的安全检查。
- 批处理系统：内存、临时文件空间、外部 API 配额都是有限资源。需要控制同一时刻运行多少任务，避免无上限地接收和执行。
- 数据库连接池：并发请求必须等待可用连接或超时失败；实际通常用连接池和超时机制，而不是银行家安全性矩阵。
这里要分清三个概念：限流（限制到达速率）、并发限制（限制同时运行的数量）和死锁避免（避免进入潜在循环等待的不安全状态），不能互相替代。
4. 缓存与配置：读者—写者怎样落地？
- 单个 Spring Boot 实例内的共享配置：多个 HTTP 线程读取同一个可变缓存，后台定期整体更新。读多写少时，可考虑
ReentrantReadWriteLock、不可变快照加原子引用，甚至使用线程安全缓存组件。 - 本地统计数据：读线程查询共享内存结构，写线程更新结构。先识别是否可能读到中间状态，再决定使用锁、原子操作还是不可变对象。
- 跨服务器共享数据：如果数据在 MySQL 或 Redis，Java JVM 内的读写锁不能跨进程保护它，此时需要数据库事务、行锁、乐观锁（版本号）、原子更新命令等机制。
一个典型高并发反例：最后一件商品被 100 人同时购买。 这主要是数据库并发更新问题，不是给 OrderService 加一把 Java 读写锁就万事大吉。MySQL 中可以用条件更新表达“库存大于 0 才允许扣减”：
UPDATE sku_stock
SET stock = stock - 1
WHERE sku_id = ? AND stock > 0;
之后检查影响行数是否为 1，并把后续订单状态处理纳入正确的事务与业务流程。真实电商还需要处理支付、取消、释放预占、重试和幂等。多台服务器访问同一数据库时，数据库原子更新才能覆盖所有实例；单 JVM 的本地锁无法代替它。
5. 一张表，把抽象模型和技术工具对齐
| 业务现象 | 应首先想到的模型/原则 | Java / Spring 常用工具 | 必须认清的边界 |
| --------------------------------- | -------------------------- | ------------------------------------- | -------------------------------- |
| HTTP 高峰产生大量后台任务 | 生产者—消费者、削峰填谷 | BlockingQueue、线程池、MQ | 队列可能积压、丢消息、重复消费 |
| 任务同时抢占内存、连接或 GPU | 有限资源分配、死锁避免 | Semaphore、连接池、资源配额、调度器 | 并发限制不等于银行家算法 |
| 多线程频繁读取、偶尔修改 JVM 数据 | 读者—写者 | ReentrantReadWriteLock、不可变快照 | JVM 锁不自动跨服务器 |
| 多实例同时修改 MySQL 库存 | 原子性、隔离性、并发控制 | 条件UPDATE、事务、版本号 | 数据库一致性不能只靠 Java 锁 |
| 多个服务需要订阅订单事件 | 生产者—消费者的跨服务扩展 | RabbitMQ、Kafka、Outbox | 需要可靠投递、消费幂等、失败恢复 |
结语：不要背仓库的故事，要学会认出“问题的形状”
三道经典题最终教会我们的，并不是把“仓库”“银行”“图书馆”这三个比喻记住，而是学会对真实业务连续追问：
任务是不是产生得比处理得快？ 如果是，考虑缓冲、背压、异步消费。资源现在能分，未来还安全吗？ 如果资源有严格需求与占有关系，考虑安全性和死锁。数据能否同时被多人访问？ 如果涉及共享可变状态，考虑读共享、写独占，以及事务边界。
真正的迁移能力：从业务问题识别计算机科学模型，再利用语言、框架和中间件写出可靠的软件。
延伸阅读与资料
- 星雨笔录 · Star Rain Notes：本文拟发布站点。
- 星雨笔录 GitHub 仓库：该站在 v1.4.0 开始支持
::: code-group语言切换；本文代码组按仓库约定写法组织。 - Java BlockingQueue API
- Java ReentrantReadWriteLock API
- Python queue 标准库
- C++ std::shared_mutex
代码运行环境提示： C 示例面向 Linux/POSIX（gcc -std=c11 -pthread），C++ 示例需要 C++20（可运行全部示例），Java 代码使用 JDK 21，Python 使用 3.x 标准库，JavaScript 使用 Node.js。每个代码语言 Tab 都是一份独立小程序；并发程序的打印顺序由调度决定，不是固定输出顺序。银行家算法属于确定性计算，其安全性判断不依赖线程调度。