《C并发编程实战》第6章的题目是设计基于锁的并发数据结构。这一章我第一次读的时候没什么感觉以为是讲怎么给已有的容器加个锁后来上手写过几个并发组件才发现这章是全书写得最容易被低估的部分。它会把你从会用锁保护一个操作拽到设计一套线程安全的数据结构接口这个层次上两者之间的差距比想象中大得多。适合基础学过线程和互斥量、想独立设计并发容器或者维护某个多线程模块的开发者读也适合那些已经把并发布进生产环境、但偶尔被诡异崩溃和数据竞争折磨的人翻一翻。1. 这一章到底在解决什么问题1.1 从给代码加锁到给数据设计接口很多并发教程教的都是这样多线程要同时操作某个共享变量那就加一把std::mutex每个线程在读写前先lock()。这个思路没错但它天然假设了代码是你自己写的你知道哪里要保护。可真实项目里业务代码和数据结构往往是分开的。调用方拿着一个普通的std::queuestd::string两个线程一个往里面塞任务一个往外取。你说加锁加在哪里是调用方在 push 前后自己加还是在 pop 前后自己加如果只包住每个操作问题不大但现实中很少只调用一次就完事。最常见的模式是if (!queue.empty()) { auto value queue.front(); queue.pop(); process(value); }这段代码在单线程下没任何问题一旦双线程执行两个线程都可以同时通过empty()检查然后一个把最后一个元素取走了另一个再去front()读取时容器已经空了。崩溃、空指针、越界都有可能发生。这不是加个锁能解决的。因为锁只能保护单次操作而上面这段代码里的检查、读取、删除是三次独立操作任何一次之间都可能被其他线程插一脚。第6章的核心观点就在这里并发数据结构的设计目标不是给容器的每个方法加锁而是让容器的每一个对外操作都是原子的、自洽的从接口层面就把竞态空间消除掉。1.2 为什么选择基于锁而不是无锁我见过不少开发者一听并发数据结构就想上无锁队列、无锁栈觉得锁是低性能的代名词。第6章不是这么带的。它刻意选择用互斥量、条件变量这些最基础的同步原语来构建设计并且完整地演示了锁方案能做得非常可靠、足够快。无锁数据结构的最大问题是正确性证明难度极高。compare_exchange循环里牵涉到的 ABA 问题、内存序选择、活锁排查每一项都比锁方案复杂一个量级。而基于锁的方案里临界区内部天然有全序的内存语义——你不需要去纠结memory_order_release和memory_order_acquire的配对关系锁本身已经替你完成了所有可见性保证。只要临界区足够短锁竞争足够低性能是完全够用的。所以这一章的一个基本原则是优先用锁只有当锁竞争已经变成明确瓶颈、且你有能力写出正确的无锁实现时才去考虑无锁方案。2. 基于锁设计前必须先想清楚的三件事2.1 选什么锁很多新手以为std::mutex就是唯一答案其实 STL 里还有好几个变体用对了场景能省不少事。第6章没有专门做对比但设计的时候绕不开这个选择。锁类型语义典型使用场景std::mutex独占锁默认选择绝大多数数据结构用它就够std::shared_mutex共享/独占锁读多写少比如配置表、查找表std::recursive_mutex同一线程可重复加锁极少使用通常是设计有问题的信号std::timed_mutex带超时的独占锁需要避免长时间阻塞的业务路径我的建议是默认std::mutex不要多想。只有在明确统计过读操作远多于写操作而且读临界区耗时不可忽略的情况下才考虑把std::shared_mutex引进来。至于recursive_mutex一旦出现同一个线程在持有锁的情况下又要调用自己加锁的函数这通常意味着锁的边界没有划清楚递归锁只是把问题藏起来了不是解决了。另外多说一个 C17 之后的小工具——std::scoped_lock。当你要同时获取多把锁的时候用它一次性传入多个 mutex底层会自动按照统一顺序加锁并做死锁避免处理。这是我在多锁结构里目前遇到的最稳妥的写法。2.2 锁的粒度怎么定锁的粒度是整个设计里最需要权衡的地方。粗粒度简单整个数据结构就是一个std::mutex所有操作串行化。好处是正确性非常好推理坏处是并发能力低。细粒度的好处是不同线程可以同时操作不同部分坏处是锁多了死锁风险和实现复杂度和它带来的性能收益经常不成正比。用一个生活化的类比来说粗锁相当于整栋办公楼只有出入口一道门禁所有人都要从那里过高峰期排队细锁相当于每个房间都有自己的门禁相邻房间的人互不干扰但万一有人要从房间A穿到房间B可能会同时依赖两道门这时候就要小心别和对面的人互相锁住。第6章里那个双锁队列就是典型的分裂锁例子push 只操作尾节点pop 只操作头节点两个方向天然不冲突所以拆成 head_mutex 和 tail_mutex 两把锁。但你要注意这种拆分不是凭空想出来的它的前提是队列的头尾操作确实不共享同一个数据节点。所以我的经验是先把正确版本用单锁写出来跑通、压测发现确实是锁竞争瓶颈了再考虑拆锁。一上来就追求细粒度很容易写出一个既复杂又微妙地错的东西。2.3 等待与通知的正确姿势基于锁的结构经常要和std::condition_variable配合用于实现阻塞语义。不过条件变量这个东西用起来有一些看起来不起眼、但错一个就全错的细节。第一它只能跟std::unique_lock配合不能是lock_guard。因为wait()内部要原子地完成解锁挂起被唤醒后重新加锁这三个动作只有unique_lock这种灵活管理锁所有权的对象才做得到。第二永远要用带谓词的重载也就是cv.wait(lock, [] { return !queue.empty(); });而不是裸写cv.wait(lock);。原因有两个一是条件变量存在虚假唤醒被唤醒不代表条件真的成立二是存在通知丢失问题——如果线程在调用 wait 之前另一个线程已经把数据 push 进去并notify_one()了而这个线程还没来得及进入等待状态通知就先发生了等它真开始 wait 时就再也没人通知它。谓词循环的作用就是每次醒来都重新检查条件不成立就继续睡这才能保证语义正确。第三通知的时机。notify_one()在锁内或锁外调用标准上都是合法的但实践中我习惯把锁先释放再通知{ std::lock_guardstd::mutex lock(mtx); data.push(std::move(value)); } cnd.notify_one();这样做的原因是如果在锁内通知被唤醒的线程会立刻尝试获取同一把锁结果发现锁还没释放又要等一轮调度白白多一次上下文切换。3. 原书三个经典结构逐个拆解3.1 线程安全栈先想清楚 pop 返回什么线程安全栈看起来简单push加锁压入pop加锁弹出。但真正设计的时候第一关就会被pop 到底怎么返回元素卡住。最原始的写法是T pop() { lock_guard lock(mtx); auto value data.top(); data.pop(); return value; }问题在于栈为空时怎么办如果抛异常调用方每次都要包 try-catch体验很差如果返回布尔值那就把检查栈状态和取出数据拆成了两次操作接口又回到了第一节说的竞态陷阱。第6章的思路是让 pop 自己报告是否成功并且尽量不在临界区内做深拷贝。于是就有了这种经典形态std::shared_ptrT pop() { lock_guard lock(mtx); if (data.empty()) return nullptr; auto value data.top(); data.pop(); return value; }返回std::shared_ptrT有两个好处。第一nullptr统一表达空栈这个状态调用方直接用返回值判断是否成功不需要再单独调用 empty()第二shared_ptr的拷贝只是原子地增加引用计数不会复制栈里的真实对象锁内耗时从深拷贝一个未知大小的 T变成拷贝一个指针速度和安全都提升了。即使你业务上非要返回裸 T也建议先std::move移到局部变量再解锁后返回别在锁里做拷贝。如果你确实要向调用方暴露empty()也必须说明它的语义是某一刻的状态快照不能拿来做先查询后操作的前置判断。这一点在文档里写清楚能避免很多使用方的误解。3.2 线程安全队列阻塞、非阻塞与双锁队列比栈多一个特性支持一端进、一端出天然更适合同步任务分发。第6章里提供了几档设计从最简单到最复杂正好对应不同性能需求。最基础的版本就是一个std::queue加一把std::mutex加一个条件变量。push加锁写入然后通知pop等数据try_pop不阻塞直接尝试。这个版本正确、直观、够用是 90% 业务场景的首选。再进一步如果想要push 和 pop 真正并发执行就需要拆锁。实现思路是引入头尾两把锁并且用链表节点代替底层容器template typename T class concurrent_queue { struct node { std::shared_ptrT data; std::unique_ptrnode next; }; std::mutex head_mutex; std::mutex tail_mutex; std::unique_ptrnode head; node* tail; public: concurrent_queue() : head(new node), tail(head.get()) {} void push(T new_value) { auto new_data std::make_sharedT(std::move(new_value)); auto new_node std::make_uniquenode(); std::lock_guard tail_lock(tail_mutex); tail-data new_data; auto old_tail std::move(tail-next); tail-next std::move(new_node); tail tail-next.get(); } std::shared_ptrT try_pop() { std::lock_guard head_lock(head_mutex); if (head.get() get_tail()) return nullptr; auto old_head std::move(head); head std::move(old_head-next); return old_head-data; } };关键点在两个地方。一个是head永远指向一个 dummy 节点实际数据从 dummy 之后的节点开始这样空队列的判断就统一成了head.get() get_tail()避免在头指针为空的情况下去碰尾指针。另一个是锁顺序push 只拿tail_mutexpop 先拿head_mutex需要比较尾节点时再短时获取tail_mutex。没有持有 tail 再去拿 head的路径锁依赖是单向的所以不会死锁——这比一般的双锁结构要安全得多。阻塞版的 pop 其实就是给try_pop外面套一个等待循环std::shared_ptrT pop() { std::unique_lock lock(head_mutex); cnd.wait(lock, [] { return head.get() ! get_tail(); }); auto old_head std::move(head); head std::move(old_head-next); lock.unlock(); return old_head-data; }如果是单消费者notify_one()就够了会唤醒一个等待线程多消费者需要广播或者后续继续调用的场景用notify_all()更稳但代价是唤醒所有线程让它们互相竞争。选择前先明确你的消费模型不要盲选。3.3 查找表读多写少用 shared_mutex第三种结构是并发查找表原书里最常用std::map搭配std::shared_mutex来做。查表这种场景典型特征是读极多、写极少比如配置、路由表、缓存索引。如果还用独占锁所有读线程都会互相阻塞吞吐很难看。std::shared_mutex允许多个读线程同时持有共享锁写线程拿到独占锁时等待所有读者退出。C17 之后标准库直接提供之前只能借助第三方库。读操作这样写std::shared_lockstd::shared_mutex lock(mtx); auto it table.find(key); if (it ! table.end()) return it-second; return default_value;写操作这样写std::unique_lockstd::shared_mutex lock(mtx); table[key] value;这里有个容易犯错的点查找的结果不能返回引用或迭代器。原因是你拿到引用后锁就释放了或者还持着锁但调用方可以保存引用逃逸另一个线程可能立刻做了一次写操作使引用失效。正确做法是在锁内把查到的 value 拷贝一份锁外返回拷贝。另一个边界是如果写操作非常频繁shared_mutex不一定比普通std::mutex更好。因为写锁每次都要等待所有读者释放锁读锁的获取又需要原子地修改同一个共享计数核间缓存一致性开销可能吃掉并行收益。所以引入shared_mutex之前最好先用数据说话。4. 设计过程中最容易踩的坑4.1 接口组合竞态压测现场常见的一种诡异崩溃调用方写的是 check then act 模式数据结构本身每个方法都安全但组合起来却不安全。栈的empty()后pop()、队列的size()后dequeue()、查找表的has(key)后get(key)都是这种模式。解决这种竞态的唯一办法是把检查和操作合并成一个原子接口让用户无从拆分。比如try_pop返回nullptr表示没拿到而不是让用户先查empty。如果有些场景真的需要查询接口那就明确文档说明此方法只能用于监控、日志、统计不能作为并发操作的前置条件。4.2 异常安全锁本身有 RAII 兜底lock_guard在异常时能保证解锁。但比这更难的是数据结构自己的不变量不能被异常破坏。最典型的例子是 pop 返回T时要先在临界区里做一次拷贝构造如果这个拷贝抛出异常节点应该仍然留在容器里而不是先弹出再拷贝导致数据丢失。这也是shared_ptr方案在这里特别值钱的原因把用户对象的生命周期交给shared_ptr管理锁内只做节点指针的方向操作几乎不会抛异常。如果你确实必须返回T正确顺序是锁内把节点摘除到局部变量解锁再从局部变量里拷贝返回值。这样即使拷贝失败最多损失这个已经安全摘除的元素不会破坏容器整体状态也不会让锁死掉。4.3 死锁与持锁调用死锁在双锁结构里最常见典型原因是加锁顺序不一致。比如线程A先锁 head 再锁 tail线程B先锁 tail 再锁 head两个线程就会互相等对方释放。解决的办法就是全局统一加锁顺序或者直接用std::scoped_lock一次性拿所有锁。更隐蔽的死锁来自持锁调用未知代码。比如在锁内调用某个用户提供的函数对象或者直接对 T 做拷贝/析构而这段用户代码内部又尝试获取同一把锁程序就会立刻进入死锁地狱。我在实际工程里的原则是锁内只做最小必要操作凡是可能回调用户代码、可能做耗时 IO、可能触发新的加锁逻辑的地方一律移出临界区。把节点从容器里摘下来以后就解锁剩下的事情让调用方自己处理。4.4 条件变量相关坑围绕条件变量容易出问题的地方比较固定汇总成一张表记起来很方便。典型问题表现对策通知丢失线程 wait 之前 notify 已发生线程永远睡下去用带谓词的 wait把检查条件和睡眠绑定在一起虚假唤醒条件未满足就被唤醒读到错误状态循环检查谓词而不是只检查一次未持锁调用 notify虽然是合法的但可能让被唤醒线程立刻抢锁失败先解锁再 notify减少无谓争抢析构时还有线程在 wait析构先发生等待线程访问已释放对象先停止生产者notify_all 唤醒所有人join 所有线程后再析构还有个经验如果你发现在某个并发结构里频繁使用notify_all来保证所有等待者都被唤醒先想想是不是消费模型本身就设计成了广播而不是单派发。前者用 all后者用 one按模型选不要按习惯选。5. 怎么验证你的设计是对的5.1 用 ThreadSanitizer 抓数据竞争写完一个并发数据结构第一件事不是上压力测试而是先用数据竞争检测器扫一遍。GCC 和 Clang 都支持线程消毒器g -stdc17 -fsanitizethread -g -O1 main.cpp -o test编译后在测试环境里跑如果代码里有未加保护的数据竞争运行时会输出明确的数据访问冲突报告精确到源文件行号。我遇到的很多并发问题靠人眼 review 根本看不出来交给 TSan 一遍就能定位。需要注意两点一是 TSan 会大幅拖慢运行速度只适合在调试阶段开二是它不负责检测死锁死锁需要通过超时机制或者逻辑审查来发现。TSan 报告干净不一定代表完全没问题但它能排除最大的那一类风险。5.2 压测关注吞吐和延迟分位而不是只看平均值数据竞争检测通过以后还要做并发压测。压测的坑在于只看平均耗时会骗人。并发质量真正要关注的是高百分位延迟比如 p99、p999因为在并发场景下偶尔一次的长时间等待可能就是锁竞争失控的信号。我自己的做法是固定一个操作集合分别用 4、8、16 个线程去跑记录每秒操作数和 p99 延迟。对比单锁队列和双锁队列时如果 4 线程下两者差距不大16 线程下单锁延迟急剧攀升而双锁平缓那就说明拆锁确实值。如果两者在 16 线程下一样拉垮那问题就不在锁粒度可能在系统调度、内存分配或者缓存命中上别急着继续拆锁。测试时还要注意构造贴近真实的数据如果生产环境里队列元素是上百字节的结构体测试就别只用 int如果查找表 90% 是读操作测试请按这个比例配置读写。压测失真比不压测的危害更大因为它会给你一个看起来很对的错误方向。我自己写并发容器的体会是锁的实现细节反而是最不费劲的部分mutex一围、condition_variable一配合正确性很容易验证。真正决定一个数据结构好用不好用的是接口怎么设计、返回什么类型、空状态如何表达、能不能让调用方自然地写出无竞态的代码。第6章的可贵之处就是它没有停留在讲锁的 API而是把这些接口层面的决策一个个摆出来告诉你为什么要做成这样。先把这些决策做对再谈性能和优化顺序反了的话后面每一步都要为早期的设计失误买单。