线程安全的单向链表是并发编程中经常被忽略的基础组件。很多人认为只要用一把互斥锁把整个链表操作保护起来就万事大吉,但实际在高并发读多写少的场景下,这种粗粒度锁会成为系统瓶颈。更危险的是,如果直接对链表节点进行无保护的插入和删除,轻则出现野指针,重则导致程序崩溃且难以复现。要解决这个问题,可以从两个方向入手:一是降低锁的粒度,让不同位置的节点操作可以并行;二是使用原子指针,通过CAS(Compare-And-Swap)操作实现无锁化。这两种思路各有侧重,本文将分别实现细粒度锁版本和原子指针版本的单向链表,并给出关键源码和对比分析。

在细粒度锁方案中,常见的做法是为每个节点维护一个独立的锁。插入或删除节点时,只需要锁定目标位置前后的少量节点,而其他线程可以同时操作链表的不同区域。原子指针方案则通过std::atomic<Node*>来存储next指针,插入和删除依靠CAS循环完成,完全避免锁的上下文切换开销。但原子指针方案容易受到ABA问题影响,本文的源码会通过带引用计数的指针或使用内存回收机制来规避。
细粒度锁实现单向链表
细粒度锁的核心是把锁放在节点内部,每个Node包含一个std::mutex成员。需要修改链表结构时,例如在某个节点之后插入新节点,必须同时锁定前驱节点和待插入位置的后继节点。如果只锁一个节点,极端情况下另一个线程可以绕过锁进行并发修改。经典的细粒度链表删除操作需要锁定两个节点:待删除节点的前驱和待删除节点本身,然后调整前驱的next指针指向待删除节点的next,最后释放两个锁。
这里有一个容易忽略的边界问题:当待删除节点是首节点时,前驱节点不存在,此时必须锁定链表头指针本身。因此链表类中需要额外引入一个head_mutex来保护head指针的读写。另一个边界是空链表,插入首节点时也需要锁定head_mutex而非某个节点的锁。下面给出的代码把head定义为一个特殊的哨兵节点,哨兵节点不存储实际数据且永远不会被删除,这样可以统一插入和删除逻辑,避免对头指针的特殊处理。哨兵节点的next指向真正的首节点,所有修改操作都从哨兵节点开始向后锁定。
#include <iostream>
#include <mutex>
#include <memory>
template <typename T>
class FineGrainedList {
private:
struct Node {
T data;
Node* next;
std::mutex mtx;
Node(const T& val) : data(val), next(nullptr) {}
};
Node head; // 哨兵节点
public:
FineGrainedList() : head(T{}) {}
void insert(const T& val) {
Node* newNode = new Node(val);
Node* prev = &head;
prev->mtx.lock();
Node* cur = prev->next;
if (cur) {
cur->mtx.lock();
}
while (cur != nullptr) {
Node* next = cur->next;
if (next) {
next->mtx.lock();
}
prev->mtx.unlock();
prev = cur;
cur = cur->next;
}
// 在尾部插入
newNode->next = nullptr;
prev->next = newNode;
prev->mtx.unlock();
}
bool remove(const T& val) {
Node* prev = &head;
prev->mtx.lock();
Node* cur = prev->next;
if (!cur) {
prev->mtx.unlock();
return false;
}
cur->mtx.lock();
while (cur != nullptr && cur->data != val) {
Node* next = cur->next;
prev->mtx.unlock();
prev = cur;
cur = cur->next;
if (cur) cur->mtx.lock();
}
if (cur) {
prev->next = cur->next;
cur->mtx.unlock();
prev->mtx.unlock();
delete cur;
return true;
} else {
prev->mtx.unlock();
return false;
}
}
void print() {
Node* cur = head.next;
while (cur) {
std::cout << cur->data << " ";
cur = cur->next;
}
std::cout << std::endl;
}
};
上述insert实现采用了一种“锁链式”前进策略,每次循环都预先锁定后一个节点再解锁当前节点,这样在遍历过程中任何时刻至少持有两个锁(prev和cur),保证了指针调整的安全性。remove操作同样遵循这一原则,找到目标节点后同时持有prev和cur的锁,然后修改prev->next即可安全删除。注意删除后要先解锁再delete,避免析构操作在锁内执行导致潜在的死锁或长时间持锁。
细粒度锁版本的缺点是每个节点都携带一个mutex,内存开销较大。在节点数量多但并发竞争不激烈的场景下,频繁的锁获取和释放会成为额外负担。此外,相邻节点操作仍然存在锁冲突,比如连续插入多个节点时,因为都需要锁定尾节点,所以无法并行。如果想进一步降低冲突,可以考虑让遍历过程只锁一个节点,但这样需要更复杂的验证机制,通常不推荐。
原子指针与无锁插入
无锁链表的核心理念是用std::atomic<Node*>存储每个节点的next指针,插入和删除操作都通过compare_exchange_weak循环完成。以插入到头部为例,先创建新节点,让新节点的next指向当前head,然后用CAS尝试把head从旧值改为新节点。如果CAS失败说明有其他线程修改了head,需要重新读取并重试。这种方式完全消除了锁,线程不会被阻塞,但会引入忙等,需要确保CAS循环能较快收敛。
#include <iostream>
#include <atomic>
#include <memory>
template <typename T>
class LockFreeList {
private:
struct Node {
T data;
std::atomic<Node*> next;
Node(const T& val) : data(val), next(nullptr) {}
};
std::atomic<Node*> head;
public:
LockFreeList() : head(nullptr) {}
void push_front(const T& val) {
Node* newNode = new Node(val);
Node* oldHead = head.load(std::memory_order_relaxed);
do {
newNode->next.store(oldHead, std::memory_order_relaxed);
} while (!head.compare_exchange_weak(oldHead, newNode,
std::memory_order_release,
std::memory_order_relaxed));
}
bool pop_front(T& result) {
Node* oldHead = head.load(std::memory_order_relaxed);
do {
if (oldHead == nullptr) return false;
} while (!head.compare_exchange_weak(oldHead, oldHead->next.load(),
std::memory_order_acquire,
std::memory_order_relaxed));
result = oldHead->data;
// 此时其他线程可能还持有oldHead指针,不能立即delete,需要安全回收
// 简化示例:假设没有并发读,实际应使用Hazard Pointer或Epoch-Based Reclamation
delete oldHead;
return true;
}
bool search(const T& val) {
Node* cur = head.load(std::memory_order_acquire);
while (cur) {
if (cur->data == val) return true;
cur = cur->next.load(std::memory_order_acquire);
}
return false;
}
};
上面的push_front是无锁链表最经典的实现,但存在严重的内存回收问题:当线程A弹出头部节点后准备delete,线程B可能已经通过旧的head指针读取了该节点的next并准备解引用,此时delete会导致线程B访问已释放内存。在生产环境中必须引入Hazard Pointer、引用计数或Epoch-Based Reclamation等安全内存回收机制。本文为了展示原子指针的基本应用,省略了内存回收细节,读者可以在此基础上扩展。
在无锁链表中执行任意位置插入(不是只操作头部)难度更大,通常需要维护一个标记位来标识节点是否已被逻辑删除。常见的实现是Michael-Scott队列的变种,使用next指针的低位bit作为删除标记。插入时先找到前驱和后继节点,然后用CAS同时更新前驱的next和新节点的next。因为无法一次CAS两个指针,所以需要通过两阶段CAS或者辅助结构。相比之下,只支持头部插入和弹出的无锁栈是无锁链表最简单的形式,因此本文的锁自由版本只实现了push_front和pop_front,搜索操作可以无锁遍历。
完整源码整合与性能对比
将两个版本的链表整合到同一个测试程序中,我们可以直观地观察不同并发线程数下的吞吐量差异。测试代码使用std::thread生成多个生产者和消费者,对链表进行大量插入和删除操作。细粒度锁版本使用FineGrainedList的insert和remove,无锁版本使用LockFreeList的push_front和pop_front。为避免内存泄漏,无锁版本的pop_front在示例中直接delete,这仅适用于低频竞争且假设没有并发读场景,测试会串行执行pop操作来规避问题。
#include <chrono>
#include <thread>
#include <vector>
#include <cassert>
int main() {
constexpr int THREADS = 8;
constexpr int OPS_PER_THREAD = 100000;
// 细粒度锁链表测试
{
FineGrainedList<int> list;
auto start = std::chrono::high_resolution_clock::now();
std::vector<std::thread> threads;
for (int i = 0; i < THREADS; ++i) {
threads.emplace_back([&list, i]() {
for (int j = 0; j < OPS_PER_THREAD; ++j) {
list.insert(i * OPS_PER_THREAD + j);
}
for (int j = 0; j < OPS_PER_THREAD; ++j) {
list.remove(i * OPS_PER_THREAD + j);
}
});
}
for (auto& t : threads) t.join();
auto end = std::chrono::high_resolution_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
std::cout << "FineGrainedList time: " << duration.count() << " ms" << std::endl;
}
// 无锁链表测试
{
LockFreeList<int> list;
auto start = std::chrono::high_resolution_clock::now();
std::vector<std::thread> threads;
for (int i = 0; i < THREADS; ++i) {
threads.emplace_back([&list, i]() {
for (int j = 0; j < OPS_PER_THREAD; ++j) {
list.push_front(i * OPS_PER_THREAD + j);
}
int val;
for (int j = 0; j < OPS_PER_THREAD; ++j) {
while (!list.pop_front(val)) {}
}
});
}
for (auto& t : threads) t.join();
auto end = std::chrono::high_resolution_clock::now();
auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);
std::cout << "LockFreeList time: " << duration.count() << " ms" << std::endl;
}
return 0;
}
在低竞争环境下(比如只有2个线程),细粒度锁版本通常比无锁版本略快,因为CAS循环在失败时会产生额外的缓存一致性流量。而随着线程数增加到8个或更多,无锁版本的优势开始显现,原因在于锁的唤醒和上下文切换开销远大于忙等。需要强调的是,无锁链表的内存回收问题在测试中被人为简化了,真实项目中必须处理,否则无法通过压力测试。一个折中方案是使用带引用计数的节点,每次解引用都增加计数,当计数归零时才delete,但这会引入std::atomic<int>的额外开销。
总体来看,选择细粒度锁还是原子指针取决于具体的应用场景。如果链表操作逻辑复杂,例如需要按值查找后删除,细粒度锁实现更直观且不易出错。如果链表只作为无界队列使用,只允许头部出队尾部入队,那么无锁方案可以显著提高高并发下的吞吐。另外,C++标准库提供的std::atomic<std::shared_ptr<Node>>(C++20起支持)也能简化无锁链表的内存回收,但需要编译器支持且会引入引用计数的性能开销。
本文给出的两个版本都只实现了单向链表的基础操作,实际项目中还需要考虑迭代器安全、异常处理和更高效的内存分配。例如使用内存池来避免频繁new/delete,使用线程局部缓存减少锁竞争。无论采用哪种方案,理解锁粒度与原子操作的内在原理,才能在并发数据结构设计中做出合理的取舍。