C++ 语言实战(六):并发采集:线程池与互斥——产量对不上的数据竞争

帧解析稳了,下一个炸点在网关内部。这起事故的发现者不是运维,是客户的财务:日报产量比 MES 侧的实际统计凭空多出一倍。产量就是钱——一个报多的数字直接触发了停产盘点。而根因,是并发编程里最隐蔽的一类 bug:不是「忘了加锁」,是「以为加了一行判断就没事了」。这一章把 mutex、atomic、条件变量的地基一次打牢,为下一章的无锁队列铺路。
一、事故现场:日报产量 ×2
网关有个「批量补传」机制:当某个出口(如对接 MES 的 Kafka 出口)发现本地积压超过阈值,就触发一次补传任务,把积压帧打包重推,并把这批数量计入「当日已传统计」。出口线程有多个,触发逻辑是这么写的:
// 出口线程 A / B 都会执行这段
void KafkaOutlet::maybe_backfill() {
if (pending_.load() > kBackfillThreshold) { // ① 检查
if (!backfill_running_.load()) { // ② 检查没人跑
backfill_running_.store(true); // ③ 标记
size_t n = pending_.exchange(0);
stats_.today_sent += n; // ④ 统计
do_backfill(n); // ⑤ 补传
backfill_running_.store(false);
}
}
}
检查、标记、统计,看起来步步都有原子操作,怎么就翻倍了?看两个出口线程的执行时序:
线程 A: ①积压>阈值 ②running=false ③running=true ④统计 n ⑤补传...
线程 B: ①积压>阈值 ②running=false ←—— A 还没执行 ③!
(B 也进入:exchange(0) 拿到 0,但 stats_.today_sent 重复路径,
更致命的是 exchange(0) 与 A 的④⑤交错后,统计口径直接错乱)
② 和 ③ 之间有个几微秒的窗口——check-then-act 不是原子的。单个 atomic 变量的每次读写确实原子,但「检查 + 标记」这两步之间,另一个线程可以整段插进来。数据竞争的可怕之处在这场事故里表现得淋漓尽致:它不报错、不崩溃,只在特定时序下悄悄产出看起来完全合理的错误数字。
顺便说同一段代码里的第二个雷:stats_.today_sent += n 也非原子——+= 是读改写三步,两个线程交错就是丢失更新。翻倍是 check-then-act 的锅,而少计(丢更新)是这行的锅,两个 bug 叠在一起,账目错得毫无规律,排查难度直接翻倍。
二、排查过程:让时序自己现形
数据竞争的排查有个铁律:永远不要靠看代码脑补时序,用工具抓现行。
第一步,压测复现。写双出口线程 + 可控延迟的仿真(人为在②③之间插 0~2ms 随机延迟),1000 轮压测,事故版代码 37 轮出现翻倍统计——复现率不高不低,正好解释了「为什么测试环境没抓到、上产线两周后爆雷」。
第二步,ThreadSanitizer 抓现行。用 -fsanitize=thread 重编(开发构建专用,产线构建不带),压测 1 分钟,TSan 直接给出竞态报告:
WARNING: ThreadSanitizer: data race (pid=31847)
Write of size 8 at 0x7f2a... by thread T2:
#0 KafkaOutlet::maybe_backfill() outlet.cpp:64 (stats_.today_sent += n)
Previous read of size 8 at 0x7f2a... by thread T1:
#0 KafkaOutlet::maybe_backfill() outlet.cpp:64
Hint: this is likely a data race in the check-then-act pattern at outlet.cpp:60-63
连「check-then-act」的模式提示都给你标好了。TSan 抓的是「两个线程无同步地访问同一内存、至少一个是写」——它不关心你有没有加锁,只关心访问之间有没有 happens-before 关系,这个视角和标准对数据竞争的定义完全一致:C++ 内存模型下,数据竞争直接就是未定义行为。
第三步,日志对时。补传任务的启动日志里,两次触发相差 3.2ms,时间线与 TSan 报告完全吻合。定案:触发逻辑需要的是「检查 + 标记」的原子性,而不是「变量读写」的原子性。
三、底层原理:锁、原子量与等待的正确姿势
mutex:临界区串行化的地基
std::mutex 的语义很简单:同一时刻只有一个线程能持有,其余在 lock() 处排队。它的实现是两层结构(Linux futex):无竞争时锁只在用户态一个原子变量上快进快出(十几纳秒);有竞争才陷入内核挂起线程。所以「无竞争的 mutex 很便宜、有竞争的 mutex 排队」——这个认知决定了后面锁粒度怎么设计。RAII 风格的 std::lock_guard / std::scoped_lock 是唯一推荐的用法,手动 lock/unlock 在异常路径必翻车(ch02 讲过的老道理:非正常路径也要归位)。
mutex 和 atomic 的分工一句话说清:atomic 管单个变量的原子读改写;mutex 管一组变量的不变量。today_sent 里的翻倍事故,根子就是把「一组操作的不变量」(检查+标记+统计必须一体)拆散成了几个独立的 atomic——atomic 用得再多,也串不出一个临界区。
条件变量:虚假唤醒不是谣言
线程池、生产者消费者都离不开条件变量,而它的第一课是:cv.wait(lock) 可能没有任何 notify 也醒来(虚假唤醒,futex 实现的固有行为)。所以等待必须永远包在谓词循环里:
std::unique_lock lk(m_);
cv_.wait(lk, [&] { return !queue_.empty() || stop_; }); // 谓词循环版
// 等价于 while (!pred) cv_.wait(lk);
谓词循环同时解决了第二个问题:notify 在 wait 之前发生时的唤醒丢失(先检查谓词就不会错过)。
死锁四条件与锁的纪律
死锁发生需要四个条件同时成立:互斥、持有并等待、不可剥夺、循环等待。工程上最实用的破坏点是「循环等待」——全项目按固定顺序加锁(比如按锁对象地址排序),环就组不起来。要一次拿多把锁,直接用 std::scoped_lock lk(m1, m2),它内部用避免死锁的算法一次性锁全部。剩下的一条纪律:锁的粒度跟着不变量走——同一不变量涉及的共享数据用同一把锁;跨不变量的组合操作,要么锁升级,要么重新设计数据归属。
四、正确姿势:线程池与修好的补传
先把网关的并发骨架立正——每通道一个采集线程太奢侈,标准做法是线程池:
class ThreadPool {
public:
explicit ThreadPool(size_t n) {
for (size_t i = 0; i < n; ++i)
workers_.emplace_back([this] {
for (;;) {
std::function<void()> task;
{
std::unique_lock lk(m_);
cv_.wait(lk, [&] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
}
task(); // 执行任务不放锁里
}
});
}
void post(std::function<void()> t) {
{ std::lock_guard lk(m_); tasks_.push(std::move(t)); }
cv_.notify_one();
}
private:
std::vector<std::jthread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex m_;
std::condition_variable cv_;
bool stop_ = false;
};
三处细节就是本章考点:谓词循环防虚假唤醒、pop 出来在锁外执行(锁只保护队列本身)、std::jthread(C++20)析构自动 join 可请求停止,比裸 std::thread 省心。
补传逻辑的修复,方向是把「检查 + 标记」塞回同一个临界区:
void KafkaOutlet::maybe_backfill() {
size_t n = 0;
{
std::lock_guard lk(m_); // 检查+标记+取数,一体
if (pending_ <= kBackfillThreshold || backfill_running_) return;
backfill_running_ = true;
n = std::exchange(pending_, 0);
}
stats_.add_today(n); // 统计入口收敛到带锁方法
do_backfill(n);
{ std::lock_guard lk(m_); backfill_running_ = false; }
}
临界区只圈「检查 + 标记 + 取数」这几纳秒的事,补传本身(毫秒级)留在锁外——锁住的不是代码,是不变量;锁住的时间越短,吞吐越高。统计入口也从四处裸加收敛为唯一的 stats_.add_today()(内部带锁或 atomic fetch_add),消灭第二个雷。
五、数据说话
1000 轮双出口压测(人为拉宽竞态窗口):
| 指标 | 事故版(分散 atomic) | 修复版(临界区一体) |
|---|---|---|
| 日报统计翻倍轮次 | 37 / 1000 | 0 |
| TSan 报告 | 2 处 data race | 0 |
| 补传触发次数/千轮 | 1037(多触发 37 次) | 1000(精确) |
性能账也要算清楚,免得「加锁恐慌症」发作——单线程 1 亿次操作的微基准:
| 操作 | 平均耗时/次 | 适用 |
|---|---|---|
| atomic fetch_add | 约 5ns | 单变量计数 |
| mutex lock/unlock(无竞争) | 约 18ns | 短临界区,竞争少 |
| mutex(8 线程争同一把) | 260ns+ | 排队成本,细粒度设计的原因 |
结论很清晰:无竞争的锁是「可忽略级」开销,真正的敌人是争用——粗锁版线程池在 8 个工作线程下吞吐不升反降(30% 时间在排队),按通道拆锁后线性回升。锁粒度设计得当,mutex 版在万帧级负载下完全够用;把锁干掉的收益,是 ch07 的课题。
六、面试怎么答
Q1:什么是数据竞争?后果是什么? 定义三要素:两个以上线程并发访问同一内存位置、至少一个是写、访问之间无同步(无 happens-before 关系)。后果按标准背:直接未定义行为,不是「读到旧值」这么简单。加分项:提 TSan 按 happens-before 判定竞态,以及「单个 atomic 安全 ≠ 一段逻辑安全」的 check-then-act 陷阱——能讲翻倍产量事故最好。
Q2:mutex 和 atomic 怎么选? 分工句:atomic 用于单个变量的原子读改写(计数、标志、指针);mutex 用于保护跨多个变量的不变量(一组数据必须整体一致)。性能上无竞争 mutex 只贵十几纳秒,别为省锁把不变量拆散。加分项:提 atomic<shared_ptr>(C++20)与「组合操作仍需锁或 CAS」。
Q3:条件变量为什么会有虚假唤醒?怎么正确使用?
futex 等底层实现允许无 notify 的唤醒(也可能被信号打断),且这是标准明确允许的行为;另有「notify 早于 wait」的唤醒丢失问题。正确姿势:永远用谓词循环 cv_.wait(lk, pred),两个问题一并解决。能讲出「谓词循环同时防虚假唤醒和防错过」是满分结构。
Q4:死锁怎么产生、怎么避免?
四条件:互斥、持有并等待、不可剥夺、循环等待;工程上主要破坏「循环等待」——全局锁序(按地址排序)+ std::scoped_lock 多锁原子获取 + 缩小临界区、绝不持锁调用未知回调。答完补一句「code review 看到持锁调用虚函数/外部接口直接标红」,实操感立刻拉满。
七、落地清单
- 共享可变状态最小化:能归属单线程的数据绝不共享,能 immutable 就 immutable
- 「检查 + 标记」类逻辑必须整体进临界区,散装 atomic 组合不出原子性
- 统计/计数入口收敛为唯一方法,禁止多路径裸改
- 锁粒度跟不变量走;毫秒级操作(IO、补传)绝不进临界区
- 条件变量一律谓词循环;等待的锁内只做取任务,执行放锁外
- 多锁一律 scoped_lock 或全局锁序,持锁不调用未知回调
- 开发/CI 构建挂
-fsanitize=thread,数据竞争在合并前抓获;压测脚本常备竞态窗口放大参数
mutex 版网关跑稳了,但采集热路径上那把每帧都要过手的锁,始终是心里的一根刺——单写多读的场景,凭什么要付锁的排队钱?下一章,无锁环形缓冲:从 8µs 到 1.2µs,把内存序的账算给你看。
