从仓库、银行到图书馆：操作系统三大经典并发问题，如何走进 Java Web？ | 星雨笔录

学操作系统时，一个困惑经常出现：我明白了仓库有生产者和消费者，也会手写几个信号量，但这套模型究竟有什么用？ 以后写的是电商、库存管理、药店管理系统，又不是开仓库。 答案是：算法模型不是业务模板，而是一套识别并发矛盾的思维工具。 现实业务的名字会变化，但“任务来得太快”“资源分出去后可能不够用”“多人读写同一份数据”等问题会反复出现。能认出矛盾，才能选择合适的技术。 一、先认出问题：仓库、银行、图书馆 同样有很多人同时做事，却存在三类完全不同的矛盾： | 生活中的故事 | 操作系统经典问题 | 一句话核心矛盾 | 在软件里首先想到 | | ---------------------------- | ---------------------------------------- | -------------------------------------- | -------------------------------- | | 仓库不断入货，员工不断取货 | 生产者—消费者（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 #include #include #define SIZE 3 #define TOTAL 8 int buffer[SIZE], head = 0, tail = 0, count = 0; semt empty, full, mutex; void producer(void arg) { (void)arg; for (int item = 1; item <= TOTAL; ++item) { semwait(&empty); // P(empty)：满了就等待 semwait(&mutex); // P(mutex)：独占缓冲区 buffer[tail] = item; tail = (tail + 1) % SIZE; // 环形队列，尾指针回绕 ++count; printf("生产 %d，库存 %d\n", item, count); sempost(&mutex); // V(mutex)：退出临界区 sempost(&full); // V(full)：通知有商品 } return NULL; } void consumer(void arg) { (void)arg; for (int i = 0; i < TOTAL; ++i) { semwait(&full); // P(full)：空了就等待 semwait(&mutex); int item = buffer[head]; head = (head + 1) % SIZE; --count; printf("消费 %d，库存 %d\n", item, count); sempost(&mutex); sempost(&empty); // V(empty)：通知有空位 } return NULL; } int main(void) { pthreadt p, c; seminit(&empty, 0, SIZE); // 初始 3 个空位 seminit(&full, 0, 0); // 初始 0 件商品 seminit(&mutex, 0, 1); // 二元信号量模拟互斥 pthreadcreate(&p, NULL, producer, NULL); pthreadcreate(&c, NULL, consumer, NULL); pthreadjoin(p, NULL); pthreadjoin(c, NULL); semdestroy(&empty); semdestroy(&full); semdestroy(&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 queue = new ArrayBlockingQueue (3); Thread producer = new Thread(() -> { try { for (int item = 1; item { 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.taskdone() # 声明该任务已处理完成 p = threading.Thread(target=producer, name="Producer") c = threading.Thread(target=consumer, name="Consumer") p.start() c.start() p.join() c.join() buffer.join() # 等待所有入队任务都 taskdone() // C++20：g++ -std=c++20 -pthread main.cpp -o pc #include #include #include #include #include std::queue buffer; std::mutex mutex; // 保护队列 std::countingsemaphore emptySlots(3); // 初始 3 个空位 std::countingsemaphore fullSlots(0); // 初始 0 件商品 void produce() { for (int item = 1; item lock(mutex); buffer.push(item); std::cout lock(mutex); int item = buffer.front(); buffer.pop(); std::cout = 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 / workerthreads。 一个值得记住的工程原则： 队列缓解短时流量峰值，不会凭空增加处理能力。若每秒产生 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. 安全性检查与资源请求伪代码 safecheck(): Work = copy(Available) Finish = [false] n sequence = [] while sequence.length Need[i]: rejectasinvalid() if Request > Available: waitforresources() // 仅仅"当前够用"还不够 Available -= Request Allocation[i] += Request Need[i] -= Request if safecheck() == SAFE: commitallocation() else: undoallthreechanges() // 回滚 rejectorwait() 所有 #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 work[j]) feasible = 0; } if (!feasible) continue; // 假设 Pi 执行完毕，归还原来持有的资源 for (int j = 0; 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 unsaferequest[M] = {3,3,0}; int saferequest[M] = {1,0,2}; requestresources(4, unsaferequest); // 资源够，但不安全 requestresources(1, saferequest); // 安全，可以批准 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 safeSequence() { int[] work = available.clone(); // 仅模拟，不改实际资源 boolean[] finish = new boolean[5]; List sequence = new ArrayList (); while (sequence.size() work[j]) enough = false; } if (!enough) continue; // 假设 Pi 完成，释放 Allocation[i] for (int j = 0; j = 5 || req == null || req.length != 3) throw new IllegalArgumentException("请求参数不合法"); for (int j = 0; j need[pid][j]) throw new IllegalArgumentException("超过合法需求"); if (req[j] > available[j]) return false; // 当前不足 } // 试分配 for (int j = 0; j 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 safesequence(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 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.safesequence() 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.safesequence()) banker.request(4, [3, 3, 0]) # 资源够，但不安全 banker.request(1, [1, 0, 2]) # 安全，可以批准 // C++17：g++ -std=c++17 main.cpp -o banker #include #include #include #include constexpr int N = 5, M = 3; using Row = std::array ; class Banker { std::array alloc{{ Row{0,1,0}, Row{2,0,0}, Row{3,0,2}, Row{2,1,1}, Row{0,0,2} }}; std::array maximum{{ Row{7,5,3}, Row{3,2,2}, Row{9,0,2}, Row{2,2,2}, Row{4,3,3} }}; std::array need{}; Row available{3,3,2}; public: Banker() { for (int i = 0; i 无安全序列 std::optional > safeSequence() const { Row work = available; std::array finish{}; std::vector seq; while (seq.size() work[j]) enough = false; if (!enough) continue; // 模拟 Pi 完成并归还当前持有的资源 for (int j = 0; j = N) return false; for (int j = 0; 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 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 n { 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 = 5 || req.length !== 3) throw new Error('请求参数不合法'); if (req.some((n, j) => !Number.isInteger(n) || 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) readshareddata() // 其他读者也可以进入 P(countMutex) readCount-- if readCount == 0: V(resource) // 最后一个读者放行写者 V(countMutex) writer(): P(resource) // 必须单独拿到资源 modifyshareddata() V(resource) 潜在问题：写者饥饿。 如果读者持续涌入，计数器可能长期不归零，写者一直进不去。所以真实并发工具还有公平锁、写者优先等策略：它们都保证读写互斥，但谁先获得执行权不一定完全相同。 3. 五种语言：读写锁的不同实现 这组示例都演示多个读取任务与一次修改操作，但实现层级有所不同：C 的 pthreadrwlockt、Java 的公平 ReentrantReadWriteLock、Python 用 Condition 自己实现写者优先、C++ 的 sharedmutex、JavaScript 用 Promise 队列协调异步读写。不同标准库对公平性和饥饿没有统一保证，不要把其中一个语言的唤醒顺序推断为所有语言都相同。 ::: code-group C|Java|Python|C++|JavaScript // Linux / POSIX：gcc -std=c11 -pthread main.c -o rw #define POSIXCSOURCE 200809L #include #include #include pthreadrwlockt rwlock = PTHREADRWLOCKINITIALIZER; int document = 0; // 共享文档 void pausebriefly(void) { struct timespec ts = {.tvsec = 0, .tvnsec = 30000000}; nanosleep(&ts, NULL); } void reader(void arg) { int id = (int )arg; pthreadrwlockrdlock(&rwlock); // 读锁：可与其他读者共享 printf("R%d 读取到 %d\n", id, document); pausebriefly(); // 模拟长时间读取 pthreadrwlockunlock(&rwlock); return NULL; } void writer(void arg) { int value = (int )arg; pthreadrwlockwrlock(&rwlock); // 写锁：独占，阻止读/写 document = value; printf("W 写入 %d\n", document); pausebriefly(); pthreadrwlockunlock(&rwlock); return NULL; } int main(void) { pthreadt r1, r2, w, r3; int ids[] = {1, 2, 3}, value = 100; pthreadcreate(&r1, NULL, reader, &ids[0]); pthreadcreate(&r2, NULL, reader, &ids[1]); pthreadcreate(&w, NULL, writer, &value); pthreadcreate(&r3, NULL, reader, &ids[2]); pthreadjoin(r1, NULL); pthreadjoin(r2, NULL); pthreadjoin(w, NULL); pthreadjoin(r3, NULL); pthreadrwlockdestroy(&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.writeractive = False self.waitingwriters = 0 def acquireread(self): with self.cond: # 有写者正在执行或已经等待，新读者暂时等待 while self.writeractive or self.waitingwriters > 0: self.cond.wait() self.readers += 1 def releaseread(self): with self.cond: self.readers -= 1 if self.readers == 0: self.cond.notifyall() # 最后一个读者唤醒等待者 def acquirewrite(self): with self.cond: self.waitingwriters += 1 try: while self.writeractive or self.readers > 0: self.cond.wait() self.writeractive = True # 写者独占 finally: self.waitingwriters -= 1 def releasewrite(self): with self.cond: self.writeractive = False self.cond.notifyall() # 通知等待的读者和写者 rw = ReadWriteLock() document = 0 def read(name): rw.acquireread() try: print(f"{name} 读取到 {document}") time.sleep(0.03) finally: rw.releaseread() def write(value): global document rw.acquirewrite() try: document = value print(f"W 写入 {document}") finally: rw.releasewrite() 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 #include #include #include #include std::sharedmutex rwMutex; // 多读、单写的共享互斥量 std::mutex outputMutex; // 只用于保护控制台输出 int document = 0; void readDocument(int id) { std::sharedlock lock(rwMutex); // 共享读锁 { std::lockguard out(outputMutex); std::cout lock(rwMutex); // 独占写锁 document = value; { std::lockguard 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 skustock SET stock = stock - 1 WHERE skuid = ? 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 都是一份独立小程序；并发程序的打印顺序由调度决定，不是固定输出顺序。银行家算法属于确定性计算，其安全性判断不依赖线程调度。