简介围绕操作系统进程间同步与互斥中的银行柜员服务问题这份PDF面向操作系统课程实验、多线程编程入门及P/V操作学习者帮助梳理信号量、互斥锁与等待队列在顾客—柜员模型中的配合方式。压缩包仅含1个PDF文件约370KB内容覆盖实验题目、实现要求、设计思路、核心伪码算法、程序结构以及顾客线程和柜员线程的函数分析并说明测试数据格式与输出要求。文中结合顾客取号、柜员叫号、无顾客时柜员等待、无空闲柜员时顾客等待等约束展开便于理解号码唯一性和互斥访问的实现思路对进程创建、信号量初始化、互斥锁保护输出流与等待队列、线程回收等细节亦有涉及。已有1874人学习适合在Ubuntu/GNU/Linux环境下结合C/C、pthread与semaphore进行对照阅读也可为课程实验报告撰写和排错提供参考。1. 七个顾客同时进门两个柜员却叫了同一个号这是我最开始跑银行柜员服务问题这个模拟程序时遇到的真实情况输入文件里 1、2、3 号顾客都在第 1 秒到达柜员只有两个终端滚出来的日志里 3 号顾客被 1 号柜员和 2 号柜员各叫了一次4 号顾客干脆没人理。问题不在业务逻辑在于进程间同步互斥没做干净——顾客取号和柜员叫号这两条路径同时摸到了等待队列。这个项目就是把这个场景用 C11 加 pthread 写成了一个能跑、能改、能验证的实验程序n 个柜员线程反复从队列里取号m 个顾客线程按到达时间睡眠后入队两把信号量管同步一把互斥锁管临界区。适合正在做操作系统课程实验、或者第一次写 Linux 多线程程序、想搞清楚 P/V 操作到底怎么落到代码里的开发者。2. 架构拆解线程模型、共享数据与两把信号量的三角关系先把整体骨架摆出来。这个程序没有用事件驱动状态机也没有用线程池而是最朴素的「一个顾客一个线程、一个柜员一个线程」主线程只负责读输入、拉起线程、等结束、收尾。这种结构在实验场景里是最合适的因为它的时间轴是真实的sleep(arrive_time)让顾客在第几秒进店这件事由内核睡眠队列推着走你不需要自己维护一个虚拟时钟去 dispatch 事件。2.1 为什么坚持一人一线程如果用单线程状态机模拟你得自己写一个最小堆存所有待发生的事件每次取最早的那个推进虚拟时间再判断这一时刻该唤醒谁。这套东西写起来比同步互斥本身还长反而把实验主题盖住了。用线程的好处是「等待」这件事交给内核顾客线程阻塞在sem_wait(counter_sem)上柜员线程阻塞在sem_wait(customer_sem)上谁该跑谁不该跑由调度器决定代码里只剩下「什么时候 post 一下」这一件事要想清楚。代价也很明确有两个。第一是输出顺序不确定同一秒内多个线程抢着cout不加锁就是满屏乱码第二是线程生命周期不受控主线程如果粗暴地exit柜员线程可能正睡在服务中途日志就断在半截。这两点后面都会单独处理。常见的做法是给每个实体配一个pthread_t字段存在结构体里创建完线程后主线程还能拿到句柄去join或cancel。这个项目就是这么做的Customer和Counter各有一个pID。2.2 共享数据清单与各自的保护方式把程序里所有跨线程可见的数据列一遍保护方式就自然清楚了数据结构类型谁写谁读保护方式customer_listvectorCustomer主线程初始化全部初始化后再不改动无需加锁counter_listvectorCounter主线程初始化全部同上customer_waitingqueueCustomer*顾客线程 push柜员线程 pop双方pthread_mutex_t mutexmutexpthread_mutex_t——本身就是锁customer_semsem_t顾客 post柜员 wait双方信号量自带原子性counter_semsem_t柜员 post顾客 wait双方同上cout/ 输出文件全局流所有线程—复用mutex这里有个容易被忽略的点输出流也算共享数据结构。很多人只记得给队列加锁忘了cout a b c这条链式调用中间是可以被抢占的两个线程的输出会交错成半句话。这份代码把所有cout都包在pthread_mutex_lock(mutex)和unlock之间算是把「所有数据结构在访问时也需要互斥」这条要求落实到位了。另一个隐含约束是customer_waiting里存的是Customer*指针指向customer_list中的元素。vector一旦扩容就会重新分配内存之前的指针全部悬空。所以初始化阶段必须把所有顾客push_back完之后再加入队操作。稳妥一点的写法是初始化前先customer_list.reserve(n)把容量一次要够。2.3 两把信号量的配对逻辑整个程序的同步就靠两条规则customer_sem的语义是「队列里有几个待服务的顾客」。初值 0顾客入队后sem_post加一柜员取号前sem_wait减一。柜员拿不到就阻塞实现了「无顾客时柜员等待」。counter_sem的语义是「当前有几个柜员刚叫完号、正等着顾客过来」。初值 0柜员叫号后sem_post加一顾客被叫到后sem_wait减一。顾客拿不到就阻塞实现了「无空闲柜员时顾客等待」。这两把信号量是严格 1:1 配对的一个顾客入队贡献一次customer_sem恰好被一个柜员消费一个柜员叫号贡献一次counter_sem恰好被一个顾客消费。配对关系成立才不会出现「两个柜员叫同一个号」——因为第二个柜员要拿customer_sem而队列里已经空了它只能阻塞。用伪码写出来只有十几行比任何文字描述都直白顾客: lock; queue.push(c); post(customer_sem); unlock; wait(counter_sem); 接受服务; 柜员: while (true) { wait(customer_sem); lock; c queue.front(); queue.pop(); post(counter_sem); unlock; 服务(c); }关键在post(counter_sem)的位置必须在pop之后、unlock之前或紧随其后。如果先 post 再 pop理论上顾客可能先醒过来读到还没更新的counter_id。虽然本例里counter_id是柜员在自己那一侧赋的值问题不大但这个顺序习惯要养成。2.4 和其他互斥方案的横向对比实验报告里通常会问「实现互斥的方法有哪些」把几种方案摆在一起看为什么最终选信号量就很清楚了方案是否忙等多处理器可用编程复杂度适用场景关中断否否极低单核内核临界区轮转法是是低几乎不用Peterson 算法是是需内存序保证中教学演示硬件指令TSL/XCHG是是低自旋锁底层信号量否是中通用同步互斥管程否是低需要语言支持信号量的短板在报告里也写得很实在控制逻辑分散在整个程序里post少写一次就死锁多写一次就溢出正确性靠人盯。管程把这些封装进语言Java 的synchronized、C 的std::condition_variable都属于这个思路后面二次开发那章会提到怎么往这个方向改。3. 核心链路走读一次取号到离店的完整代码路径架构清楚了剩下就是把两个线程函数拆开看。这两段代码加起来不到八十行但每一行的位置都有讲究挪一格行为就变。3.1 两个结构体的字段含义先看数据结构字段命名很直白但每个字段的生命周期不同这点值得说清struct Customer { pthread_t pID; // 顾客线程 id主线程 join 时用 int id; // 顾客序号来自输入文件第一列 int arrive_time; // 相对程序启动的到达秒数 int service_duration; // 需要被服务的秒数 int service_start; // 开始服务的时刻柜员线程写入 int counter_id; // 服务柜员号柜员线程写入 Customer(int cus_id, int a_time, int s_duration) { pID 0; id cus_id; arrive_time a_time; service_duration s_duration; service_start -1; // 未服务时先置无效值方便排查 counter_id -1; } }; struct Counter { pthread_t pID; // 柜员线程 id主线程 cancel 时用 int id; // 柜员序号从 1 开始 Counter(int count_id) { pID 0; id count_id; } };service_start和counter_id初始化为 -1 是个小习惯比留 0 好输出时如果看到 -1说明这个顾客压根没被服务过比看到 0 号柜员更容易定位问题。arrive_time和service_duration都是秒和sleep()的单位一致不需要换算。3.2 顾客线程睡眠进店、锁内入队、阻塞等叫号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; }逐段看。sleep(arrive_time)是相对程序启动计时所以输入文件里写 5 就真的是第 5 秒到。pthread_mutex_lock到unlock之间做了三件事打印到达日志、入队、sem_post。三件事必须在同一个临界区里否则会出现「日志打了但还没入队柜员这时候抢到锁发现队列是空的」这种错位——虽然柜员有信号量挡着不会真的取空但日志和实际状态对不上排查时会误导你。sem_post(customer_sem)放在锁内不是必须的放锁外能缩短临界区、减少争用。我一般放锁内理由是入队和通知构成一个原子语义读代码的人一眼能看出「入队即通知」少一次上下文切换的心智负担。sem_wait(counter_sem)在锁外这是重点。如果把它写在锁内柜员线程就永远拿不到锁去post直接死锁。信号量的 wait 必须放在临界区外面让线程真正挂起。最后那个sleep(service_duration)是顾客线程自己睡的和柜员线程里的sleep(service_duration)是并行的两段。这一点后面对时间复核时要注意实际「离开时间」以柜员线程的输出为准。3.3 柜员线程循环取号、工作期禁止取消void *counter_thread(void *arg) { Counter counter *(Counter *)arg; while (true) { // 等顾客没有顾客时阻塞在这里 sem_wait(customer_sem); // 进入工作状态此期间不允许被主线程 cancel 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 at time(NULL) - program_start_time endl; pthread_mutex_unlock(mutex); // 恢复可取消状态并检查是否有待处理的取消请求 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); pthread_testcancel(); } }sem_wait(customer_sem)在循环体最上面柜员没事干的时候就停在这儿不占 CPU。pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL)这一行是这份代码里最容易被删掉、删掉就出事的一行。主线程在所有顾客服务完后会pthread_cancel所有柜员线程而pthread_cancel默认是延迟取消只在取消点生效sleep恰好是取消点。如果柜员正睡在服务中途被取消就会出现「开始服务日志打了、结束服务日志没了、顾客记录里counter_id有值但时间轴断裂」的情况。禁止取消把整个「取号到服务完成」包成一段不可中断的业务退出检查统一放到循环末尾。pthread_testcancel()是配套的因为取消了ENABLE状态后不是立刻生效得有一个取消点去检查。循环末尾放一个正好在两次服务之间的空闲位置响应退出请求。sem_post(counter_sem)在pop()之后保证顾客被唤醒时队列已经更新、counter_id和service_start已经写好。这个顺序如果反了顾客醒来读到的可能是上一轮的陈旧值。3.4 主线程拉起、join、cancel、销毁主线程的逻辑不长但顺序不能错int main() { // 1. 读输入文件构造 customer_list / counter_list read_input(input.txt); // 2. 初始化同步原语 pthread_mutex_init(mutex, NULL); sem_init(customer_sem, 0, 0); // 初值 0一开始没有待服务顾客 sem_init(counter_sem, 0, 0); // 初值 0一开始没有空闲柜员叫号 program_start_time time(NULL); // 3. 先起柜员线程再起顾客线程 for (auto c : counter_list) pthread_create(c.pID, NULL, counter_thread, c); for (auto c : customer_list) pthread_create(c.pID, NULL, customer_thread, c); // 4. 等所有顾客结束 for (auto c : customer_list) pthread_join(c.pID, NULL); // 5. 顾客全走了通知柜员下班 for (auto c : counter_list) pthread_cancel(c.pID); for (auto c : counter_list) pthread_join(c.pID, NULL); // 6. 销毁原语、写结果 sem_destroy(customer_sem); sem_destroy(counter_sem); pthread_mutex_destroy(mutex); write_output(output.txt); return 0; }先起柜员再起顾客是为了避免极端情况下第一个顾客已经入队并post了却还没有任何柜员线程存在去消费——信号量是计数的值会攒着其实不会丢但日志顺序会好看很多。第 5 步的pthread_cancel后面必须跟pthread_join否则线程资源不回收而且主线程可能在柜员还在写cout的时候就destroy了mutex那是未定义行为。第 6 步把输出统一写到output.txt字段顺序是「进入时间 开始服务时间 离开时间 柜员号」和题目的输出要求一一对应。4. 跑起来Ubuntu 下的编译、输入输出格式与首个用例复现代码看懂了接下来是让它真的在机器上跑出结果。这一节按「装环境 → 编译 → 造数据 → 核对输出」的顺序来每一步都给可复制的命令。4.1 环境与编译命令目标平台是 Ubuntu 18.04.5 LTSGNU/Linux 4.15 内核实际上任何带 GCC 和 pthread 的 Linux 发行版都能跑。需要的只有两样支持 C11 的 g以及libpthreadglibc 自带。最省事的是写一个 Makefile把编译参数固化下来CXX g CXXFLAGS -stdc11 -Wall -O2 -pthread TARGET bankcounter.exe SRCS bankcounter.cpp $(TARGET): $(SRCS) $(CXX) $(CXXFLAGS) -o $ $^ run: $(TARGET) ./$(TARGET) clean: rm -f $(TARGET) output.txt .PHONY: run clean-pthread这一个参数同时负责编译期宏定义和链接期-lpthread比手动写-lpthread更稳。-stdc11是因为代码里用了vector的范围 for 和统一初始化。-Wall建议留着多线程代码里「变量未初始化」这类警告往往是 bug 的前兆。不想写 Makefile 的话一条命令也行g bankcounter.cpp -o bankcounter.exe -stdc11 -lpthread ./bankcounter.exe编译期如果报sem_init未声明检查有没有#include semaphore.h报pthread_create未定义引用就是链接参数没带-pthread。这两个是新手最常撞的墙。4.2 输入文件字段与一份可复现的用例输入文件每行一条记录三个字段空格分隔顾客序号、进入银行的时间、需要服务的时长单位都是秒。下面是 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 号服务 11 秒会一直占到第 12 秒中间 4、5、6、7 陆续到达队列会积压这就触发了「队列非空时柜员连续取号」的分支。跑完这一份基本所有路径都覆盖了。柜员数量由源码里的常量控制const int COUNTER_NUM 2; // 柜员人数改这里就能改并发度改成 1 会看到明显的串行化改成 5 会看到大量柜员空转。这个常量后面会讲怎么改成命令行参数。4.3 输出核对用日志反推时序对不对程序跑起来会往终端打实时日志同时把结构化结果写进output.txt每行五个字段顾客序号、进入时间、开始服务时间、离开时间、柜员号。按上面那份输入、2 个柜员理论时序是这样的时刻事件1顾客 1、2、3 同时到达1柜员 1 叫 1 号柜员 2 叫 2 号3 号排队5柜员 2 服务完 2 号14叫 3 号5顾客 4 到达排队6顾客 5 到达排队7顾客 6、7 到达排队8柜员 2 服务完 3 号53 的服务从第 5 秒算叫 4 号12柜员 1 服务完 1 号111叫 5 号12柜员 2 服务完 4 号82 从第 8 秒起算叫 6 号15柜员 1 服务完 5 号123叫 7 号17柜员 2 服务完 6 号125空闲17柜员 1 服务完 7 号152空闲核对时看三个不变量就够了每个顾客的「开始时间 ≥ 进入时间」「离开时间 开始时间 服务时长」同一个柜员的两段服务区间不能重叠。这三个只要有一个破了就说明临界区或者信号量配对出了问题。还有一点要有心理预期柜员号的分配本身是不确定的。1、2、3 号同时到达时哪个柜员先抢到锁就叫 1 号这个顺序每次跑可能不一样。这不是 bug恰恰是多线程调度不可预期的体现。写验收脚本的时候不要断言具体柜员号只断言上面那三个不变量。5. 避坑与排查多线程实验最容易翻车的五个位置同步互斥代码的特点是「写错不报错跑错不固定」。下面这五条是我在这个程序上真实踩过的按现象、原因、解决三段写遇到对应的症状可以直接对号入座。5.1 现象柜员日志里最后一条 finish 不见了具体表现是终端能打出serve customer 7但对应的finish serving customer 7没有output.txt里最后一两个顾客的记录也缺。原因在主线程的退出时机。主线程pthread_join完所有顾客线程后立刻pthread_cancel柜员而顾客线程和服务柜员的sleep是两段并行的睡眠顾客那边醒来时柜员可能还差几毫秒才输出 finish。此时取消请求已经发出柜员在下一次pthread_testcancel()处直接退出。解决有两个方向。保守一点是在 cancel 之前插一个短暂的sleep(1)让最后一轮收尾跑完能解决问题但属于掩盖。正规做法是加一个「全部完成」的计数信号量每个柜员服务完一个顾客sem_post一次主线程在 join 顾客之后改sem_wait若干次再 cancel或者干脆不用 cancel改成全局bool finished顾客全跑完后置真并sem_post若干次customer_sem柜员sem_wait返回后检查标志位自行return退出状态可控。5.2 现象运行几次柜员号乱跳同一份输入结果不一致这是最容易被误判成 bug 的现象。1、2、3 号顾客同时在第 1 秒到达两个柜员谁先抢到锁谁就先叫号线程调度顺序本身不受控所以 1 号顾客这次由柜员 1 服务、下次由柜员 2 服务完全是正常的。要区分「合法的不确定性」和「真的错了」看三个不变量有没有顾客被服务两次counter_id被覆盖说明取号出队有问题、有没有顾客counter_id一直是 -1说明信号量丢了配对、同一柜员的两段服务区间有没有重叠说明pop和post顺序反了。这三个都没问题柜员号怎么跳都对。如果确实需要结果可复现比如写报告要贴图可以在柜员取号前加一个基于counter.id的微秒级错峰或者把并发度临时改成 1别去动同步逻辑。5.3 现象段错误或者顾客信息变成一串垃圾值程序跑着跑着Segmentation fault (core dumped)core 文件里能看到崩在customer_waiting.front()或者customer.counter_id ...这一行。八成是vector悬空指针。customer_waiting里存的是customer指向customer_list中的元素如果后续还有customer_list.push_back导致扩容整块内存搬家队列里所有指针全部失效。这个坑在有动态添加顾客需求的时候特别容易触发。解决办法有两个选一个就行。最直接的是读完全部输入后立刻customer_list.reserve(n)并且保证初始化完成前不启动任何线程更稳妥的是把队列元素从Customer*改成int类型的顾客下标柜员用customer_list[idx]去取指针问题从根上消失。5.4 现象输出时间戳全一样或者明显对不上 sleep日志里好几个事件的秒数都是同一个数或者顾客 1 服务了 11 秒、离店时间却只比开始时间大 1。原因是用time(NULL)取时间精度只有秒。好几个事件落在同一秒内时间戳自然重复更麻烦的是time(NULL)返回的是墙上时钟如果机器时间被调整过差值可能是负的。解决办法是换成clock_gettime(CLOCK_MONOTONIC, ts)拿tv_sec * 1000 tv_nsec / 1000000得到毫秒级的单调时间减掉程序启动时的基准值。改成毫秒之后sleep也得相应换成usleep或std::this_thread::sleep_for否则事件密度还是上不去。做实验报告的话毫秒精度足够看清明细了。5.5 现象程序不退出或者 CPU 被某个线程吃满跑完所有顾客终端停住不动CtrlC才能结束或者用top看进程占着几十甚至上百的 CPU。不退出通常是死锁sem_wait卡在一个永远不会被post的信号量上。最常见的是把sem_wait(counter_sem)写进了mutex的临界区里柜员要拿锁才能post顾客拿着锁等柜员两边互锁。排查手段是把mutex换成带错误检查的PTHREAD_MUTEX_ERRORCHECK类型或者跑gdb -p pid然后thread apply all bt看每个线程卡在哪一行。CPU 打满则是另一类问题有人把信号量换成了忙等待的while (!flag) ;或者sem_wait的返回值没检查就当成拿到了资源。信号量本身不忙等出现高 CPU 一定是代码里混进了轮询。这一条改起来简单把轮询换成sem_wait或者条件变量即可。6. 二次开发把柜员数改成命令行参数加一份等待时长统计原程序的柜员人数是编译期常量每次想试不同并发度都得改源码重编做对照实验很别扭。改造的第一步就是把它变成运行时参数。#include cstdlib int main(int argc, char *argv[]) { // 用法./bankcounter.exe 3 input.txt int counter_num (argc 1) ? atoi(argv[1]) : 2; const char *infile (argc 2) ? argv[2] : input.txt; if (counter_num 0) { cerr counter number must be positive endl; return 1; } read_input(infile); // ……后续初始化与线程创建COUNTER_NUM 全部换成 counter_num }atoi不校验非法输入生产代码里用strtol加errno检查更稳实验场景够用。改完之后./bankcounter.exe 5 input.txt就能直接跑到 5 个柜员做「柜员数 vs 总耗时」的曲线不用再重编译。第二步是加统计这个改动信息量最大。在Customer里加一个wait_time字段柜员取号时算// 柜员线程内取号成功后 customer.service_start time(NULL) - program_start_time; customer.wait_time customer.service_start - customer.arrive_time; // 核心指标然后在write_output时汇总平均等待时长和最大等待时长。这份数据直接回答实验报告里那个思考题柜员数增加时总耗时先降后平曲线拐点在哪看平均等待时长什么时候趋近 0 就知道了。第三步可以考虑把 pthread 换成std::thread加std::condition_variable这是更现代也更接近「管程」思路的写法。用condition_variable之后customer_sem和counter_sem就不需要了改成wait(lock, predicate)谓词里判断队列非空还是柜员空闲逻辑集中在一个函数里比散落各处的sem_post好维护。迁移时最容易翻车的是忘记在notify_one之前释放锁或者谓词写成捕获引用的临时变量。我一般会先拿单柜员、单顾客的最小用例跑通再扩展别一上来就全量替换。最容易忽略的其实是最后一步验证改完之后拿原来的 7 人 2 柜员用例跑一遍把输出和改造前的结果按「三个不变量」逐条比对而不是比对具体柜员号。这个习惯是那次最后一轮 finish 日志丢失之后养成的从那以后我每次动同步逻辑都强制先跑一遍最小规模用例确认配对没丢再上大规模压测。希望帮到你。本文还有配套的精品资源点击获取