基于锁的并发数据结构
读完能设计线程安全的栈和队列(pop 返回 optional、condvar 阻塞队列),能用 hand-over-hand 交替加锁的细粒度链表和按 key 分桶的查找表提高并发,并知道为什么绝不能把指向内部受保护数据的指针或引用泄露到锁的作用域之外。
学习目标
- 能写出没有 check-then-act 竞态的线程安全栈,以及支持关闭协议的阻塞队列,并解释双锁队列为何能让 push 与 pop 重叠
- 能设计细粒度锁链表和分桶查找表,画出每把锁保护的不变量,并把跨分区操作改成固定锁序
- 能回答:一个「每个成员函数都正确加了锁」的线程安全栈,为什么调用方写
if(!s.empty()) v=s.top(); s.pop();仍然是错的?怎么改?
机制总览
锁粒度、接口安全与并发数据结构
- 1
定义接口
pop 返回值与状态变化一次完成,不暴露锁外失效的内部引用。
- 2
拆分锁域
分桶或逐节点锁让独立键和节点并行,但每个不变量有明确边界。
- 3
移动锁
hand-over-hand 先取得下一节点再释放当前节点。
章级决策实验
锁粒度、接口安全与并发数据结构
选择数据结构层级,比较粗粒度锁、分桶锁和逐节点锁的正确性边界。
选择推理阶段
当前阶段 · 定义接口
pop 返回值与状态变化一次完成,不暴露锁外失效的内部引用。
可核验证据
接口竞态测试与异常路径。
细粒度锁只有在不变量可以局部分解时才提高并发;接口、异常和节点寿命仍必须整体证明。
失效—证据矩阵
锁粒度、接口安全与并发数据结构
定义接口
典型失效
线程安全成员函数组合成非线程安全调用序列。
核验证据
接口竞态测试与异常路径。
拆分锁域
典型失效
锁拆得比不变量更细,跨桶操作观察到半更新。
核验证据
锁域表、并发度与状态一致性。
移动锁
典型失效
释放当前锁后再找下一节点,节点可能被并发删除。
核验证据
节点寿命策略、锁序列与 sanitizer。
从一把总钥匙开始拆分
前几章你学会了给一块共享数据配「唯一的厨刀」(锁),让线程有序地碰它。可一旦要给整个栈、队列、链表、查找表这种数据结构做到多线程能安全共用,新问题来了:这把锁该锁多大?
最省事的办法是给整个后厨配一把总钥匙:谁要操作就锁住全场,别人全在门口等。安全是安全,可三个厨师里始终只有一个能干活,另两个干等——人多了也快不起来。
更聪明的办法是每个操作台各配一把小锁:动不同台子的厨师互不干扰、可以同时开工,只有两人挤到同一张台子才需要排队。代价是钥匙多了、规矩也多了,搞不好两人各攥一把对方要的钥匙就僵住。这一章就是教你怎么给常见数据结构选锁、配锁,既安全又快。
设计原则:怎样才算「线程安全」,以及四条红线
我们说一个数据结构是 ↡多个线程可以同时调用它的成员函数,而不会破坏数据的不变量、不会产生数据竞争、也不会读到「改了一半」的中间状态——使用者不必在外面再额外加锁。 的,意思是:多个线程同时调用它的成员函数,既不破坏数据的不变量,也不会有数据竞争,使用者不必在外面再加锁。要做到这点,设计时有四条红线(后面每一节都在贯彻它们):
- 每一处读写共享数据都得在同一把锁的保护下——漏一处,竞态就回来了。
- 别让接口竞态钻空子:把「必须一气呵成」的复合操作(如「判空 + 取顶 + 弹出」)合并成一个加锁的成员,而不是让调用方分几步组合(这正是上面学习目标里那个自测题)。
- 异常安全:操作中途抛异常时,锁要正确释放、数据的不变量不能被留在破损状态(用
lock_guard这类 RAII 锁,再小心安排「会抛的那步」的位置)。 - 绝不把指向内部受保护数据的指针或引用泄露到锁外:返回
T&或内部容器的引用,调用方就能在锁之外访问它——保护被「漏」出了临界区,等于没锁。要么返回值拷贝,要么把要做的事以回调传进锁内执行。
这就是并发数据结构设计的起点:先定义线程安全接口提供哪些原子语义,再决定所有权、锁分区和组合顺序。不能先把 mutex 撒进成员函数,再期待接口自然变安全。
还有一条贯穿全章的取舍:锁粒度。
锁粒度:粗与细的取舍
↡指「一把锁保护多大范围的数据」。粗粒度=一把大锁锁住整个结构(实现简单、并发度低);细粒度=把数据拆成多块、每块各一把小锁(并发度高、实现复杂)。 指一把锁保护多大范围的数据。粗粒度就是「一把总钥匙锁全场」:实现最简单,但任一时刻只有一个线程能碰这个结构,多线程退化成串行。细粒度则是「每台一把小锁」:把数据拆成多块、每块一把锁,动不同块的线程可以并行,并发度高得多——代价是钥匙多了,死锁、不变量跨锁被破坏、异常安全都更难处理。
下面这张对照动画把同样三个操作分别用粗粒度和细粒度跑一遍,你会一眼看出吞吐差距。先猜一猜,再玩:
猜一猜:把一把大锁拆成多把小锁,三个线程的总耗时会变短吗?什么情况下细粒度也帮不上忙?
第 1 / 6 步 · 粗粒度:A 拿到唯一的总钥匙,锁住整个结构开始操作
同样三个操作:粗粒度(一把大锁)铺满整条时间轴串行做完;细粒度(多把小锁)不冲突的并行、只有撞同一块才等——挤在前段就干完。可暂停、单步、拖进度。
看清楚了:粗粒度下三个操作铺满整条时间轴(串行);细粒度下,动不同块的 A、B 在第一拍就并行开工,只有也要动块1 的 C 才需要排队——同样的活,挤在前段就干完。本章接下来讲的栈/队列多用粗粒度(简单够用),链表和查找表则用细粒度榨并发。
线程安全栈与队列:把接口竞态焊死
栈和队列结构简单,通常一把锁就够(粗粒度,但访问本身很快,够用)。关键不在粒度,而在接口:必须把会被组合使用的操作合并成原子的成员。线程安全栈的核心是把 top + pop 合并成一个返回 std::optional<T> 的 pop()——空栈返回 nullopt,绝不让调用方「先 empty() 再 top()」留缝。
队列还多一个需求:消费者常常希望「队列空了就睡、有了立刻醒」。这就是 ↡队列为空时,取元素的线程不是立刻返回失败,而是挂起等待,直到有元素被放进来才被唤醒——通常用 std::condition_variable 实现 wait_and_pop。与之相对的 try_pop 队列空时立刻返回,不等待。:用 std::condition_variable 实现 wait_and_pop(阻塞等到有元素),同时也提供 try_pop(空就立刻返回)。具体代码见第六节。
细粒度链表:hand-over-hand 交替加锁
链表想要高并发,就不能一把锁锁全表(那样多线程退化成串行)。细粒度方案给每个节点各配一把锁,遍历时用 ↡细粒度链表的遍历手法:沿链表逐节点走,握住下一个节点的锁才松开当前节点的锁(像沿一长条料理台逐格挪,抓住下一格才放手当前格)。任一时刻最多只锁相邻两个节点,从不锁整张表。:握住下一个节点的锁,才松开当前节点的锁——像沿一条长料理台逐格往前挪手,任一时刻最多只锁住相邻两个节点。下面这张动画把这套「攀爬」演给你看:
第 1 / 6 步 · ① 先锁住 node1:开始遍历,手里只握着第一格的锁
先锁 node1 → 锁 node2 再放 node1 → 锁 node3 再放 node2 → ……握住下一格才松开当前格,逐节点挪到尾,从不锁全表。可暂停、单步、拖进度。
盯住每一拍「同时握着哪两格」:因为从不锁住整张表,别的线程能同时在你没握到的节点上操作——这就是细粒度链表比粗粒度并发度高的原因。但它也最容易写出死锁:所有线程必须沿同一方向逐节点锁(永远「先持当前、再获取后继,拿到才放当前」),否则两个线程反向相遇就会各持一把互等(详见第七节)。代码见第六节。
线程安全查找表:分桶 + 读写锁
查找表(哈希表)天然适合细粒度:把数据按 key 的哈希分到若干 ↡线程安全查找表(哈希表)的细粒度方案:把数据按 key 的哈希分到若干个桶,每个桶各配一把独立的锁。访问某 key 时只锁它所在的那个桶——落不同桶的访问用不同的锁,因此并行;只有落同一桶才争同一把锁。,每个桶各配一把独立的锁。访问某 key 时只锁它所在的桶——落不同桶的访问用的是不同的锁,因此能并行;只有落到同一个桶才需要排队。下面这张动画演示三个线程落桶后谁并行、谁排队:
第 1 / 5 步 · 三个线程各拿一个 key,先哈希算出落到哪个桶(apple→桶0、melon→桶2、acorn→桶0)
A 落桶0、B 落桶2——不同锁,并行;C 也落桶0——和 A 争同一把锁,排队。分桶把一把大锁拆成多把,只有撞同一桶才串行。可暂停、单步、拖进度。
更进一步,查找表通常读多写少,每个桶用 std::shared_mutex(读写锁)而非普通 mutex:查找用共享锁(多读并行)、修改用独占锁。于是「分桶」和「读写锁」两层并发叠加——不同桶完全并行,同一桶的并发读也不互斥。代码见第六节。
上手玩一玩这三张图
这三张图都是可单步的教具,配合代码一起玩比读十遍管用:
- 粗 vs 细对照
<CoarseVsFineLockDiagram />:单步对比上区(一把大锁,三操作串行铺满时间轴)和下区(多把小锁,不冲突的并行、只有撞同一块才等)。重点体会「拆锁 ≠ 一定更快」——撞同一块时细粒度也得排队。 - hand-over-hand
<HandOverHandDiagram />:单步看「握住下一格才松开当前格」,盯住从不出现「四格全亮」——那才是锁全表。理解为什么这样能让别的线程同时动没被握住的节点。 - 分桶锁
<BucketLockDiagram />:单步看 A 落桶0、B 落桶2 并行,C 也落桶0 与 A 排队。体会「落不同桶 = 不同锁 = 并行,落同一桶 = 同一把锁 = 排队」。
玩熟这三张图,下一节每一段代码里的 lock_guard / unique_lock / shared_lock / hand-over-hand 移交,你都能对上图里的某个节点。
代码:四种线程安全数据结构
线程安全栈:pop 返回 optional,焊死接口竞态
承接第三章的 SafeStack,这里给一个更完整的版本。关键是把 top + pop 合并成一个加锁的 pop(),空栈返回 std::nullopt:
#include <mutex>
#include <optional>
#include <stack>
#include <utility>
template <typename T>
class ThreadSafeStack {
std::stack<T> data;
mutable std::mutex m;
public:
void push(T value) {
std::lock_guard<std::mutex> g(m);
data.push(std::move(value));
}
// 判空 + 取顶 + 弹出,一把锁内一气呵成;空栈返回 nullopt
std::optional<T> pop() {
std::lock_guard<std::mutex> g(m);
if (data.empty()) return std::nullopt;
T value = std::move(data.top());
data.pop();
return value;
}
};调用方只用一句 if (auto v = s.pop()) use(*v);——「查空 + 取顶 + 弹出」在同一把锁里完成,中间没有缝可钻。注意 m 声明为 mutable,这样 const 成员函数(如一个 empty() const)里也能锁它。
泛型版本还要声明异常保证:若 T 的移动构造会抛出并修改源对象,data.top() 可能留下被部分移动的值。生产实现可要求 nothrow move,在可用时先复制,或返回预先构造好的 std::shared_ptr<T>,确保修改容器前返回对象已经安全建立。
线程安全阻塞队列:condition_variable 的 wait_and_pop
队列要支持「空了就睡、有了就醒」。用一个 std::condition_variable 配 std::mutex:push 后 notify_one,wait_and_pop 在条件不满足时挂起:
#include <condition_variable>
#include <mutex>
#include <optional>
#include <queue>
#include <stdexcept>
#include <utility>
template <typename T>
class ThreadSafeQueue {
mutable std::mutex m;
std::condition_variable cv;
std::queue<T> data;
bool closed = false;
public:
void push(T value) {
{
std::lock_guard<std::mutex> g(m);
if (closed) throw std::logic_error("queue is closed");
data.push(std::move(value));
}
cv.notify_one(); // 在锁外通知,唤醒一个等待者
}
void close() {
{
std::lock_guard<std::mutex> g(m);
closed = true;
}
cv.notify_all();
}
std::optional<T> wait_and_pop() {
std::unique_lock<std::mutex> lk(m);
cv.wait(lk, [this] { return closed || !data.empty(); });
if (data.empty()) return std::nullopt;
T value = std::move(data.front());
data.pop();
return value;
}
};close() 让等待者有可证明的退出路径:队列关闭且排空后返回 nullopt。没有关闭状态,析构前仍在 wait 的消费者无法有序离开。wait_and_pop 使用 unique_lock,因为 wait 要原子地释放锁并阻塞,醒来后再锁回并检查谓词。
同一个队列再加一个非阻塞的 try_pop:空就立刻返回 nullopt,不等待。关闭状态与“暂时为空”若需要由调用者区分,应返回枚举或结果类型,而不是让一个 nullopt 承担两种含义。
#include <mutex>
#include <optional>
#include <utility>
std::optional<T> try_pop() {
std::lock_guard<std::mutex> g(m);
if (data.empty()) return std::nullopt;
T value = std::move(data.front());
data.pop();
return value;
}双锁队列:让生产端和消费端真正并行
上面的单 mutex 队列适合先保证正确。原书进一步用哨兵节点把 head 和 tail 分离:最后一个空节点永远由 tail 指向,push 只修改尾端,pop 只移走头端。队列为空当且仅当 head.get() == tail_snapshot()。
#include <condition_variable>
#include <memory>
#include <mutex>
#include <utility>
template <typename T>
class TwoLockQueue {
struct Node {
std::shared_ptr<T> data;
std::unique_ptr<Node> next;
};
std::mutex head_mutex;
std::unique_ptr<Node> head;
std::mutex tail_mutex;
Node* tail;
std::condition_variable data_cv;
Node* tail_snapshot() {
std::lock_guard<std::mutex> lk(tail_mutex);
return tail;
}
public:
TwoLockQueue() : head(std::make_unique<Node>()), tail(head.get()) {}push 在尾哨兵写入数据并追加一个新哨兵;wait_and_pop 持有 head 锁,在谓词中短暂读取 tail。全类锁序只有 head_mutex -> tail_mutex,push 从不反向获取 head 锁,因此不会形成等待环。
void push(T value) {
auto item = std::make_shared<T>(std::move(value));
auto dummy = std::make_unique<Node>();
Node* new_tail = dummy.get();
{
std::lock_guard<std::mutex> lk(tail_mutex);
tail->data = std::move(item);
tail->next = std::move(dummy);
tail = new_tail;
}
{
std::lock_guard<std::mutex> handshake(head_mutex);
}
data_cv.notify_one();
}
std::shared_ptr<T> wait_and_pop() {
std::unique_lock<std::mutex> lk(head_mutex);
data_cv.wait(lk, [this] { return head.get() != tail_snapshot(); });
auto result = head->data;
head = std::move(head->next);
return result;
}
};这是基于锁的队列细化:一个生产者持 tail 锁时,消费者仍可处理已有 head。push 更新尾端后短暂获取并释放 head mutex,确保若消费者刚把谓词判断为假,生产者的 notify 会发生在消费者进入 wait 之后,避免错过通知。所有嵌套获取仍只有 head -> tail;push 在获取 head 前已释放 tail。生产版本还应把关闭状态纳入同一协议,并规定析构只能发生在所有调用者停止之后。
细粒度锁链表:hand-over-hand 的遍历
链表每个节点带一把自己的锁。push_front 只需锁住头节点;遍历则用 hand-over-hand——先看 push_front 与「查找」:
#include <memory>
#include <mutex>
#include <utility>
template <typename T>
class ThreadSafeList {
struct Node {
std::mutex m;
std::shared_ptr<T> data; // head 哨兵节点 data 为空
std::unique_ptr<Node> next;
Node() : next(nullptr) {}
explicit Node(T const& value)
: data(std::make_shared<T>(value)) {}
};
Node head; // 哨兵头节点
public:
void push_front(T const& value) {
std::unique_ptr<Node> n(new Node(value));
std::lock_guard<std::mutex> lk(head.m); // 只锁头
n->next = std::move(head.next);
head.next = std::move(n);
}
// ... find_first_if 见下 ...
};push_front 只锁头节点,不打扰链表其余部分——别的线程仍能在后面遍历。真正用到 hand-over-hand 的是遍历:先持当前节点的锁,再获取下一个节点的锁,拿到后才释放当前(对照上面那张攀爬动画):
#include <memory>
#include <mutex>
#include <utility>
template <typename Predicate>
std::shared_ptr<const T> find_first_if(Predicate p) {
Node* current = &head;
std::unique_lock<std::mutex> lk(head.m); // 握住当前(头)
while (Node* const next = current->next.get()) {
std::unique_lock<std::mutex> next_lk(next->m); // 先握住下一个
lk.unlock(); // 再放开当前
if (p(*next->data)) return next->data; // 命中:返回 shared_ptr 值
current = next;
lk = std::move(next_lk); // 下一个变「当前」,锁所有权移交
}
return std::shared_ptr<const T>();
}这里用 unique_lock 是因为它能提前解锁和移动转移。返回 shared_ptr<const T>,让对象生存期独立于节点且不通过该接口暴露可变访问。谓词在节点锁内执行,因此契约必须要求它短小、不得重入本链表;否则会延长锁持有时间或自锁。若可接受复制 T,可在锁内复制快照后解锁再调用未知代码。
线程安全查找表:分桶 + shared_mutex
查找表按 key 哈希分桶,每桶一把 std::shared_mutex。查找用共享锁(多读并行),增改用独占锁:
#include <functional>
#include <list>
#include <memory>
#include <shared_mutex>
#include <stdexcept>
#include <utility>
#include <vector>
template <typename Key, typename Value, typename Hash = std::hash<Key>>
class ThreadSafeLookupTable {
struct Bucket {
std::list<std::pair<Key, Value>> data;
mutable std::shared_mutex mutex; // 每桶一把读写锁
};
std::vector<std::unique_ptr<Bucket>> buckets;
Hash hasher;
Bucket& get_bucket(Key const& k) const {
return *buckets[hasher(k) % buckets.size()];
}
public:
explicit ThreadSafeLookupTable(unsigned n = 19) : buckets(n) {
if (n == 0) throw std::invalid_argument("bucket count must be positive");
for (auto& b : buckets) b.reset(new Bucket);
}
// ... value_for / add_or_update 见下 ...
};桶的数量固定(构造时定好、不再变),所以桶数组本身不需要加锁——只需锁单个桶。这是分桶的关键简化。读写两个操作分别用两种锁:
#include <mutex>
#include <shared_mutex>
// 读:共享锁,多个读线程可同时进同一个桶
Value value_for(Key const& key, Value const& def = Value()) const {
Bucket& b = get_bucket(key);
std::shared_lock<std::shared_mutex> lk(b.mutex);
for (auto const& e : b.data)
if (e.first == key) return e.second; // 返回值拷贝
return def;
}
// 写:独占锁,写时连读也挡
void add_or_update(Key const& key, Value const& value) {
Bucket& b = get_bucket(key);
std::unique_lock<std::shared_mutex> lk(b.mutex);
for (auto& e : b.data)
if (e.first == key) { e.second = value; return; }
b.data.push_back({key, value});
}读用 std::shared_lock、写用 std::unique_lock。桶数固定,所以没有并发 rehash;若要扩容,必须另行设计覆盖所有桶的全局协议。快照等跨桶操作要按桶索引固定顺序持有全部共享锁。shared_mutex 是否更快以及读写公平性由实现和负载决定,必须测量。
容易踩的坑
小结
- 线程安全数据结构的四条红线:每处读写同一把锁保护、把复合操作合并成原子成员避免接口竞态、异常安全、绝不把内部受保护数据的指针/引用泄到锁外
- 线程安全栈/队列:把
top+pop合并成返回std::optional的pop()焊死接口竞态;队列用std::condition_variable做空了就阻塞的wait_and_pop(必带谓词防虚假唤醒)+ 不阻塞的try_pop - 锁粒度取舍:粗粒度(一把大锁)简单但串行、并发为零;细粒度(多把小锁)并发高但更难写对——粒度太粗等于没并发
- 细粒度链表用 hand-over-hand 交替加锁(握住下一个才松开当前),任一刻只锁相邻两节点、从不锁全表;必须单向逐节点锁,否则死锁
- 线程安全查找表按 key 分桶、每桶一把锁让落不同桶的访问并行;读多写少时每桶用
shared_mutex,读共享、写独占,再叠一层并发
练习
问题 1(改代码型) 下面这个「线程安全栈」用了 std::stack 的经典接口,把 top 和 pop 分开了,存在隐患。请把它改成没有接口竞态的版本(提示:合并 top+pop,空栈返回 std::optional)。
#include <mutex>
#include <stack>
#include <utility>
template <typename T>
class BadStack {
std::stack<T> data;
mutable std::mutex m;
public:
void push(T v) {
std::lock_guard<std::mutex> g(m);
data.push(std::move(v));
}
bool empty() const {
std::lock_guard<std::mutex> g(m);
return data.empty();
}
T& top() { // ← 隐患:返回内部引用
std::lock_guard<std::mutex> g(m);
return data.top();
}
void pop() { // ← 隐患:与 top/empty 组合用有竞态
std::lock_guard<std::mutex> g(m);
data.pop();
}
};问题 2(问答型) 同事写细粒度锁链表的 find 时,为了「快点拿到下一个节点」,先锁住 current->next、再锁 current。他说反正都锁上了肯定安全。这样写有什么风险?正确的加锁顺序是什么?
问题 3(独立实现型) 实现一个线程安全的有界缓冲区 BoundedBuffer<T>:put(T) 在满时等待,take() 在空时等待,构造时拒绝零容量。要求用一把 mutex 和两个 condition variable,并说明生产关闭时还需增加什么协议。
名词解释
本章出现的专业名词,用大白话再讲一遍。
- 线程安全(thread-safe)
说一个数据结构「线程安全」,是指多个线程可以同时调用它的成员函数,而不会破坏数据的不变量、不会有数据竞争、也不会读到「改了一半」的中间状态——使用者不必在外面再额外加锁。通常靠在内部给每个会读写共享数据的操作配上互斥量来实现。详见本章「设计原则」一节。
- 锁粒度(lock granularity)
指「一把锁保护多大范围的数据」。粗粒度=一把大锁锁住整个结构,像后厨只有一把总钥匙、谁进谁锁全场,实现简单但任一刻只有一个线程能干活(并发低);细粒度=把数据拆成多块、每块各一把小锁,像每个操作台一把锁、动不同台子的厨师可同时开工(并发高),代价是更难写对(死锁、异常安全等)。详见本章「锁粒度」一节。
- hand-over-hand(交替加锁)
细粒度链表的遍历手法:沿链表逐节点往前走,握住下一个节点的锁,才松开当前节点的锁——像沿一条长料理台逐格挪手,抓住下一格才放手当前格,故名「手递手」。任一时刻最多只锁住相邻两个节点,从不锁住整张表,所以别的线程能同时操作你没握到的节点,并发度比「一把锁锁全表」高得多。注意必须单向逐节点锁,否则会死锁。详见本章「细粒度链表」一节。
- 分桶锁(bucket lock)
线程安全查找表(哈希表)的细粒度方案:把数据按 key 的哈希分到若干个桶,每个桶各配一把独立的锁;访问某个 key 时只锁它所在的那个桶。像按食材种类把后厨分成几个独立料理区、各区各一把小锁——落到不同桶的访问用的是不同的锁,因此能并行,只有落到同一个桶才需要争同一把锁排队。这就把「一把大锁」拆成了「每桶一把锁」,降低争用。详见本章「线程安全查找表」一节。
- 阻塞队列(blocking queue)
一种线程安全队列:当队列为空时,取元素的线程不会立刻返回失败,而是挂起等待,直到有元素被放进来才被唤醒——通常用
std::condition_variable实现wait_and_pop。与之相对的是try_pop:队列空时立刻返回(返回空optional或 false),不等待。前者适合「消费者等着干活」,后者适合「拿不到就走、不想阻塞」。详见本章「线程安全栈与队列」一节。