简介：《进程间同步互斥问题——银行柜员服务问题1》是一份面向操作系统课程学习者与多线程编程入门者的实验资料，聚焦银行柜员服务场景下的进程同步与互斥建模。压缩包仅含1个PDF文件，大小约370KB，篇幅集中，便于在Ubuntu/GNU Linux环境下对照阅读。资料围绕P、V操作展开，梳理顾客取号、柜员叫号、等待队列、顾客信号量与柜员信号量等核心对象，并给出顾客线程、柜员线程的伪码算法、pthread互斥锁与C++11数据结构的使用思路，以及测试数据格式和输出要求。已有1874人学习，适合需要完成课程实验、理解信号量同步机制或复盘线程协作设计的读者参考，可从中获得问题分解、程序结构与实现要点。

1. 银行柜员服务问题：把 P、V 操作从试卷搬进 pthread 的一课

很多人学操作系统时，信号量就停在 PV 原语那几行伪代码上，考试能画状态图，真让你开个终端敲g++就发怵。这份课程实验的题目叫「进程间同步互斥问题——银行柜员服务问题」，在 Ubuntu 18.04.5 LTS 上，用 C/C++ 加部分 C++11 特性，把 n 个柜员和一群顾客各拆成线程，用sem_t做同步、pthread_mutex_t做互斥，跑出了一个有时序、有输出文件、可复现的完整程序。它适合两类人：一类是刚学完同步互斥、想看看信号量在真实代码里长什么样的同学；另一类是要交同类实验、需要一份能编译能改参数的骨架的开发者。它不讲大道理，就是把「取号—叫号—服务—离开」这条链路用线程和信号量实现了一遍。

读完你能拿到的东西很具体：一份bankcounter.cpp的线程模型拆解，两个信号量各自管什么，为什么等待队列必须和输出流共用一把锁，以及怎么把柜员数从 2 改成 5、把输入文件换成自己的测试集。我最看重的是它把「同步」和「互斥」这两件容易混为一谈的事，在同一份代码里分得清清楚楚——这部分后面会展开。

1.1 先分清哪些是互斥、哪些是同步

题目里给了提示，但真正写起来最容易翻车的就是这一步。互斥的对象是「共享数据结构的访问」：顾客取号要把自己塞进等待队列，柜员叫号要从队列头部弹一个出来，这两个动作都会改queue，还会往cout写日志，所以它们必须串行。同步的对象是「事件发生的先后依赖」：柜员不能在没有顾客的时候叫号，顾客不能在没有空闲柜员的时候开始服务。前者用一把pthread_mutex_t mutex兜住，后者用两个信号量customer_sem和counter_sem表达「现在队列里有几个顾客」「现在有几个柜员腾出手了」。

把这两层分开之后，代码结构就非常清楚：每次碰共享数据前后加解锁，每次等待对方信号量就sem_wait，每次通知对方就sem_post。血泪经验是，别图省事把sem_wait也塞进锁里——一旦锁内阻塞，整个银行瞬间停摆，这个坑我在别的地方踩过不只一次。

1.2 数据结构怎么摆：结构体、vector 和队列

程序定义了Customer和Counter两个结构体。顾客身上挂线程 id、序号、到达时刻、服务时长、开始服务时刻和被分配的柜员号；柜员只挂线程 id 和序号。所有顾客放在vector<Customer>里，所有柜员放在vector<Counter>里，初始化时按输入文件逐行创建。等待中的顾客用一个queue<Customer*>存放指针，这样柜员从队列里取出的就是原始对象，回填counter_id和service_start时直接改在结构体上，不用再拷来拷去。

这里用 C++11 的vector、queue和范围遍历，比手写数组加链表省事得多，也让整个程序能跨平台编译。柜员数量用常量COUNTER_NUM给定，顾客数量则由输入文件的行数决定，两个规模参数是解耦的，改起来很方便。

2. 架构拆解：一个顾客从进门到离开经过了几道关

整个程序的骨架是「主线程造线程、等线程、杀线程」。主线程先读input.txt，把每个顾客结构体建好，再建COUNTER_NUM个柜员结构体，然后为每个顾客pthread_create一个顾客线程，为每个柜员pthread_create一个柜员线程。顾客线程服务完就自然结束，柜员线程是死循环，靠主线程在「所有顾客线程 join 完毕」之后发取消信号收尾。最后销毁互斥锁和信号量，把结果写进output.txt。

这个「先 join 顾客、再 cancel 柜员」的顺序是有讲究的：顾客全服务完，说明队列一定空了，柜员此时要么在sem_wait上睡着，要么刚服务完一个人准备下一轮，无论哪种状态，取消都是安全的。如果反过来先杀柜员，队列里可能还有没被服务的顾客，那些顾客线程就会永远卡在sem_wait(&counter_sem)上，进程直接挂死。

2.1 顾客线程：取号、等叫、被服务

顾客线程函数是整个模型里最短的一段，但每个动作都有对应的同步语义：

// 顾客线程：模拟到达 -> 取号入队 -> 等柜员叫号 -> 接受服务 void *customer_thread(void *arg) { Customer &customer = *(Customer *)arg; // 用 sleep 模拟顾客在 arrive_time 时刻到达银行 sleep(customer.arrive_time); // ---- 取号：修改共享的等待队列，必须持锁 ---- pthread_mutex_lock(&mutex); cout << "[CUSTOMER] id " << customer.id << " arrive at " << time(NULL) - program_start_time << endl; customer_waiting.push(&customer); // 入队的是指针，回填结果直接改原对象 sem_post(&customer_sem); // 通知柜员：队列里多了一个人 pthread_mutex_unlock(&mutex); // ---- 等叫号：阻塞在柜员信号量上，直到某个柜员腾出手 ---- sem_wait(&counter_sem); pthread_mutex_lock(&mutex); cout << "[CUSTOMER] id " << customer.id << " served by " << customer.counter_id << " at " << customer.service_start << endl; pthread_mutex_unlock(&mutex); // 用 sleep 模拟接受服务的时长 sleep(customer.service_duration); return NULL; }

sem_post(&customer_sem)和sem_wait(&counter_sem)这两句是顾客与柜员之间的全部同步契约。前者告诉系统「我进队列了，谁来接我」，后者表示「我在这里等，直到有人叫我的号」。注意customer.counter_id和customer.service_start这两个字段并不是在顾客线程里赋值的，而是在柜员线程叫号时写进去的——顾客线程sem_wait返回的时候，柜员已经在锁内把这两个字段填好了，所以这里读它们是安全的。这个「生产者写、消费者读、靠信号量定序」的模式是这份代码里最值得抄走的一招。

参数上，arrive_time和service_duration都直接来自输入文件的第二、三字段，单位就是秒，sleep拿到的就是这两个值，所以整个程序的时间轴是可控的、可复现的。

2.2 柜员线程：死循环里等客、叫号、服务

柜员线程比顾客线程多了两处细节：取消状态的管理和「先取人再 post」的顺序：

// 柜员线程：等顾客 -> 叫号出队 -> 服务 -> 循环 void *counter_thread(void *arg) { Counter &counter = *(Counter *)arg; while (true) { // 没有顾客就睡在这里，这一步不能在锁内做 sem_wait(&customer_sem); // 已经领到活，进入不可取消区，避免叫到一半被主线程干掉 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL); pthread_mutex_lock(&mutex); Customer &customer = *customer_waiting.front(); customer.counter_id = counter.id; // 回填柜员号 customer.service_start = time(NULL) - program_start_time; // 回填开始时刻 cout << "[COUNTER] id " << counter.id << " serve customer " << customer.id << " at " << customer.service_start << endl; customer_waiting.pop(); // 出队 sem_post(&counter_sem); // 通知这位顾客：你可以开始被服务了 pthread_mutex_unlock(&mutex); sleep(customer.service_duration); // 模拟服务耗时 pthread_mutex_lock(&mutex); cout << "[COUNTER] id " << counter.id << " finish serving customer " << customer.id << endl; pthread_mutex_unlock(&mutex); pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); pthread_testcancel(); // 取消点：离开工作状态后才会真正响应 cancel } }

pthread_setcancelstate这两行是很多人第一次写会漏掉的。柜员被主线程pthread_cancel的时候，如果它正好处在「已经出队、还没 post」的中间态，顾客就会永远等不到叫号。把这段工作区标成不可取消，再在循环末尾用pthread_testcancel集中响应取消，是一种「延迟取消」的写法。sleep本身也是一个取消点，所以如果不禁用，柜员很可能睡在服务过程中间就被干掉，日志对不上、顾客悬空，排错时非常难看。

调用顺序也值得盯一眼：一定是先pop再sem_post(&counter_sem)，而且都在同一把锁里。如果先 post 再解锁出队，理论上存在另一个柜员看到信号量但队列已被改动的时间窗，虽然这份代码里柜员不会互相抢同一个人，但养成「改完共享状态、锁内发信号」的习惯，换到多消费者场景就不会翻车。

2.3 主线程：读文件、造线程、收尾的顺序

主线程的责任是把上面两个角色编排起来，关键在收尾阶段的顺序：

# 编译与运行：Makefile 里的核心两条 g++ bankcounter.cpp -o bankcounter.exe -std=c++11 -lpthread ./bankcounter.exe # 输入从 input.txt 读，结果写 output.txt

编译时必须带-lpthread，这是 pthread 系列的链接开关，少了它pthread_create、sem_init一类的符号全部报 undefined reference。-std=c++11是给vector、queue和部分初始化写法用的，用不着新特性的话也可以去掉，但既然题目要求了就别省。

主线程的收尾逻辑是：对每个顾客线程pthread_join，全部返回后说明所有顾客都已完成「到达—服务—离开」；这时再对每个柜员线程pthread_cancel并pthread_join；然后pthread_mutex_destroy、sem_destroy；最后把vector<Customer>里的结果按「到达时间、开始服务时间、离开时间、柜员号」写进output.txt。柜员号从 1 开始计，这一点在输出格式要求里写死了，改代码时别顺手写成从 0 开始。

3. 编译运行与结果验证：拿一组数据把时间轴对一遍

把代码跑起来只要两行命令，难的是跑完之后怎么判断它对不对。程序自带的测试样例是 7 个顾客、2 个柜员，输入长这样：

1 1 11 2 1 4 3 1 7 4 5 2 5 6 3 6 7 5 7 7 2

三个字段依次是顾客序号、进入银行时刻、需要服务时长。时刻 1 同时来了 1、2、3 三位顾客，但只有两个柜员，所以必然有一位要排队；时刻 5 和 6 来的 4、5 号大概率要等到 1 号或 3 号服务完才有柜员空出来。

3.1 怎么核对输出对不对

output.txt每行五个字段：顾客进入时间、开始服务时间、离开时间、柜员号，再加上顾客序号。核对的方法不是逐行跟屏幕日志比对，那样太累，而是验三条不变量：

这四条能兜住绝大部分逻辑错误。比如你把sem_post(&counter_sem)写到了pop之前，或者忘了在叫号时持锁，柜员占用区间就可能重叠；比如队列里存的是值而不是指针，回填的柜员号写不回原对象，输出里就会出现柜员号全 0 的情况。

3.2 结果为什么每次跑不完全一样

同一份输入连跑两次，你会发现柜员分配可能不同——1 号和 2 号谁先抢到 1 号顾客，取决于调度。这不是 bug，恰恰是多线程不可预期性的正常体现。题目本身也承认了这一点：同时到达的顾客多于柜员时，分配的柜员在不同次运行中可能不同。所以验收时别拿某一次的柜员号做唯一标准，只验上面那四条不变量。反倒是「时间轴上事件先后」应该稳定：12 时刻 1 号和 3 号同时完成服务、两个柜员同时空出来，这个节奏在 1 秒粒度的sleep下是能对上的。

提示：time(NULL)精度是秒，两个事件落在同一秒内时日志顺序有随机性。想看得更清楚，把时间源换成std::chrono::steady_clock并输出毫秒，逻辑不用动。

4. 避坑清单：这份代码最容易翻车的几个点

4.1 柜员线程被取消在错误的位置

现象：程序偶发卡死，某个顾客始终没有被服务，进程不退出。原因：主线程pthread_cancel打在柜员「已从队列取出顾客、尚未sem_post(&counter_sem)」的窗口里，柜员线程直接终止，顾客永远等不到信号量。解决：把叫号到 post 这一段用pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL)包起来，循环末尾用pthread_testcancel()集中响应，这份代码就是这么写的，别删。

4.2 在锁内调用 sem_wait 导致全员阻塞

现象：程序一启动就卡住，日志只打印了一两行。原因：把sem_wait(&customer_sem)放进了pthread_mutex_lock和unlock之间。柜员等不到顾客时会一直持有锁，其他柜员和顾客谁都进不来，整个银行死锁。解决：同步信号量的wait永远放在锁外，锁内只做共享数据的读写和post通知，这是铁律。

4.3 队列存值不存指针，回填结果丢失

现象：output.txt里柜员号全是 0，开始服务时间也全是 0。原因：queue<Customer>存的是拷贝，柜员从队列取出的副本上的counter_id、service_start改完就随副本一起丢了，原vector里的对象没动过。解决：队列声明成queue<Customer*>，push 时传&customer，取出后解引用再改，这份代码用的就是这个写法。

4.4 编译漏掉 -lpthread 或忘了 -std=c++11

现象：链接期一堆undefined reference to pthread_create、sem_init，或者vector、queue相关语法报错。原因：pthread 需要显式链接，C++11 的标准库特性需要语言标准开关。解决：命令固定成g++ bankcounter.cpp -o bankcounter.exe -std=c++11 -lpthread，或者直接make走 Makefile，别手敲半截。

4.5 输出流没加锁，日志交错乱序

现象：屏幕上的[CUSTOMER]和[COUNTER]两行字粘在一起，读不出谁先谁后。原因：cout的多次<<不是原子操作，两个线程同时往标准输出写就会插花。解决：所有cout都包在mutex的 lock/unlock 里，这份代码中每一处打印都在锁内，代价是日志成了串行的，但可读性值得。

5. 二次开发：把柜员数改成 5、把输入换成自己的测试集

这份代码最好的地方是规模参数和业务逻辑解耦，改起来不需要动线程模型。柜员数量直接改常量COUNTER_NUM，从 2 改成 5，主线程会自动多造三次线程、多循环几次，其他一行不用动。输入文件换掉就行，每行三字段的格式没变，顾客数量随行数自动增长。

想做更进一步的改造，我一般按这个顺序走：先把sleep换成std::this_thread::sleep_for(std::chrono::milliseconds(x))，把时间粒度从秒降到毫秒，测试跑得快，时间轴也更细；再把time(NULL) - program_start_time换成steady_clock的差值，避免系统时间跳变干扰；然后加一个「VIP 顾客」字段，让等待队列按优先级而不是严格先进先出出队，这就从单一队列变成了优先队列，锁的作用范围不变，但front()要换成取最高优先级的元素。

验证方法建议写一个独立的校验脚本，读output.txt逐条检查第 3 章那四条不变量，而不是靠肉眼看日志。下面这段用 Python 做抽查的思路可以直接用：

# 校验 output.txt：时间单调、时长一致、柜员区间不重叠 rows = [] with open("output.txt") as f: for line in f: cid, arrive, start, leave, counter = map(int, line.split()) rows.append((cid, arrive, start, leave, counter)) # 每个顾客的开始时间不能早于到达时间 for cid, arrive, start, leave, counter in rows: assert start >= arrive, f"顾客 {cid} 未到先服务" # 服务时长应与输入文件第三字段一致（这里把输入也读进来对一遍） inp = {} with open("input.txt") as f: for line in f: cid, arrive, dur = map(int, line.split()) inp[cid] = dur for cid, arrive, start, leave, counter in rows: assert leave - start == inp[cid], f"顾客 {cid} 服务时长对不上" # 同一柜员的任意两段服务区间不能交叉 by_counter = {} for cid, arrive, start, leave, counter in rows: by_counter.setdefault(counter, []).append((start, leave)) for counter, spans in by_counter.items(): spans.sort() for (s1, e1), (s2, e2) in zip(spans, spans[1:]): assert e1 <= s2, f"柜员 {counter} 服务区间重叠：{s1}-{e1} 与 {s2}-{e2}" print("全部通过")

这段脚本把「人工看日志」变成了「机器验不变量」，改完代码跑一遍就知道有没有把同步逻辑改坏。好处是无论你把柜员数改成几、输入换成多大规模的测试集，校验都通用；坏处是它只能证明「没违反不变量」，不能证明「调度最优」，后者本来就超出这道题的范畴。

注意：改造时如果加了超时机制（比如顾客等超过 N 秒就走人），一定要在sem_wait上换成sem_timedwait，并处理ETIMEDOUT返回值，否则超时只是日志上的一句话，线程该卡还是卡。

这套流程跑下来，我对这类同步题有了个固定习惯：每改一次信号量的位置或锁的范围，哪怕只挪了一行，都强制跑一遍不变量校验脚本再提交。多线程的 bug 不会当场报错，它只会在某次调度运气不好时给你一记闷棍，后悔药是没有的，只能靠校验提前把黑匣子拆开看。希望帮到你。

本文还有配套的精品资源，点击获取