C++11 pthread多线程银行柜员服务模拟与信号量同步互斥实践

简介：围绕操作系统进程间同步与互斥中的银行柜员服务问题，这份PDF面向操作系统课程实验、多线程编程入门及P/V操作学习者，帮助梳理信号量、互斥锁与等待队列在顾客—柜员模型中的配合方式。压缩包仅含1个PDF文件，约370KB，内容覆盖实验题目、实现要求、设计思路、核心伪码算法、程序结构，以及顾客线程和柜员线程的函数分析，并说明测试数据格式与输出要求。文中结合顾客取号、柜员叫号、无顾客时柜员等待、无空闲柜员时顾客等待等约束展开，便于理解号码唯一性和互斥访问的实现思路；对进程创建、信号量初始化、互斥锁保护输出流与等待队列、线程回收等细节亦有涉及。已有1874人学习，适合在Ubuntu/GNU/Linux环境下结合C/C++、pthread与semaphore进行对照阅读，也可为课程实验报告撰写和排错提供参考。

1. 七个顾客同时进门，两个柜员却叫了同一个号

这是我最开始跑银行柜员服务问题这个模拟程序时遇到的真实情况：输入文件里 1、2、3 号顾客都在第 1 秒到达，柜员只有两个，终端滚出来的日志里 3 号顾客被 1 号柜员和 2 号柜员各叫了一次，4 号顾客干脆没人理。问题不在业务逻辑，在于进程间同步互斥没做干净——顾客取号和柜员叫号这两条路径同时摸到了等待队列。这个项目就是把这个场景用 C++11 加 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 共享数据清单与各自的保护方式

把程序里所有跨线程可见的数据列一遍，保护方式就自然清楚了：

这里有个容易被忽略的点：输出流也算共享数据结构。很多人只记得给队列加锁，忘了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 和其他互斥方案的横向对比

实验报告里通常会问「实现互斥的方法有哪些」，把几种方案摆在一起看，为什么最终选信号量就很清楚了：

信号量的短板在报告里也写得很实在：控制逻辑分散在整个程序里，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 LTS（GNU/Linux 4.15 内核），实际上任何带 GCC 和 pthread 的 Linux 发行版都能跑。需要的只有两样：支持 C++11 的 g++，以及libpthread（glibc 自带）。

最省事的是写一个 Makefile，把编译参数固化下来：

CXX = g++ CXXFLAGS = -std=c++11 -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更稳。-std=c++11是因为代码里用了vector的范围 for 和统一初始化。-Wall建议留着，多线程代码里「变量未初始化」这类警告往往是 bug 的前兆。

不想写 Makefile 的话，一条命令也行：

g++ bankcounter.cpp -o bankcounter.exe -std=c++11 -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、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 被某个线程吃满

跑完所有顾客，终端停住不动，Ctrl+C才能结束；或者用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 日志丢失之后养成的，从那以后我每次动同步逻辑，都强制先跑一遍最小规模用例确认配对没丢，再上大规模压测。希望帮到你。

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