高性能 C++ 实战 (三)：工程可用版 CAS 无锁队列，解决 ABA 与内存泄漏，高并发实测（附源码）

上一版最简无锁队列，适合理解原理，但不能直接上生产：存在 ABA 风险、节点内存释放不安全，多线程下直接delete节点会触发野指针问题。 无锁开发最大难点不是 CAS，而是内存安全回收。 本篇基于 Michael-Scott 无锁队列，增加风险指针 Hazard Pointer 简易内存回收解决 ABA 与并发析构问题，附带压测代码，以实际运行效果为准。

项目仓库：high_performance_cpp_demo，代码路径：lockfree/lockfree_queue.hpp

1. 基础版本存在的 2 个工程致命问题

单纯靠版本号只能缓解指针 ABA，并发场景节点不能随便 delete，需要安全内存回收机制，这里选用业界常用 Hazard Pointer（风险指针）。

2. Hazard Pointer 简易实现原理

• 线程访问节点前，把节点指针存入自己的 HP（风险指针），标记：这个节点我正在用，别删
• 回收节点时，先放入待回收链表；只有没有任何线程 HP 引用该节点，才真正释放内存。

3. 完整工程化源码 lockfree/lockfree_queue.hpp

#ifndef LOCKFREE_QUEUE_HPP
#define LOCKFREE_QUEUE_HPP

#include <atomic>
#include <vector>
#include <thread>
#include <cassert>

// 简易HazardPointer 解决无锁内存安全回收，避免ABA+野指针delete
class HazardPointer {
public:
static constexpr int MAX_HP = 100;
struct HPRecord {
std::atomic<std::thread::id> owner;
std::atomic<void*> ptr;
};
static HPRecord hp_records[MAX_HP];
std::atomic<HPRecord*> my_hp{nullptr};
std::vector<void*> retired_list;

HazardPointer() {
// 分配当前线程的HP记录
for (int i = 0; i < MAX_HP; ++i) {
std::thread::id empty_id{};
if (hp_records[i].owner.compare_exchange_strong(empty_id, std::this_thread::get_id())) {
my_hp = &hp_records[i];
break;
}
}
}
~HazardPointer() {
retire_all();
if(my_hp) my_hp.load()->owner.store(std::thread::id{});
}

// 标记ptr为正在使用
void* protect(std::atomic<void*>& src) {
void* p;
do {
p = src.load();
if (!p) return nullptr;
my_hp.load()->ptr.store(p);
} while (src.load() != p);
return p;
}

// 取消保护
void unprotect() {
if(my_hp) my_hp.load()->ptr.store(nullptr);
}

// 待回收，延迟删除
void retire(void* p) {
retired_list.push_back(p);
if(retired_list.size() >= 100) {
scan_retire();
}
}

void scan_retire() {
std::vector<void*> remain;
for(auto* p : retired_list) {
bool hazard = false;
for(int i=0;i<MAX_HP;i++) {
void* hp_p = hp_records[i].ptr.load();
if(hp_p == p) {
hazard = true;
break;
}
}
if(hazard) remain.push_back(p);
else delete p;
}
retired_list.swap(remain);
}
void retire_all() {
scan_retire();
for(auto* p : retired_list) delete p;
retired_list.clear();
}
};
HazardPointer::HPRecord HazardPointer::hp_records[HazardPointer::MAX_HP]{};

template<typename T>
class LockFreeQueue
{
private:
struct Node
{
T* data;
std::atomic<Node*> next;
Node() : data(nullptr), next(nullptr) {}
explicit Node(T&& val) : data(new T(std::move(val))), next(nullptr) {}
~Node() { delete data; }
};

std::atomic<Node*> head_;
std::atomic<Node*> tail_;
public:
LockFreeQueue()
{
Node* dummy = new Node();
head_.store(dummy);
tail_.store(dummy);
}

~LockFreeQueue()
{
Node* cur = head_.load();
while (cur != nullptr)
{
Node* temp = cur;
cur = cur->next.load();
delete temp;
}
}

// 入队
void enqueue(T&& val)
{
Node* new_node = new Node(std::move(val));
Node* old_tail;
while (true)
{
old_tail = tail_.load();
Node* next = old_tail->next.load();
if (old_tail != tail_.load()) continue;
if (next == nullptr)
{
if (old_tail->next.compare_exchange_weak(next, new_node))
{
tail_.compare_exchange_weak(old_tail, new_node);
break;
}
}
else
{
tail_.compare_exchange_weak(old_tail, next);
}
}
}

// 出队，增加HP内存保护，安全回收节点
bool dequeue(T& out)
{
HazardPointer hp;
Node* old_head;
while (true)
{
old_head = static_cast<Node*>(hp.protect(reinterpret_cast<std::atomic<void*>&>(head_)));
if (!old_head) return false;
Node* old_tail = tail_.load();
Node* next = old_head->next.load();
if (old_head != head_.load()) continue;
if (old_head == old_tail)
{
if (next == nullptr)
{
hp.unprotect();
return false;
}
tail_.compare_exchange_weak(old_tail, next);
}
else
{
if (head_.compare_exchange_weak(old_head, next))
{
out = *(next->data);
hp.retire(old_head);
hp.unprotect();
return true;
}
}
}
}
};

#endif

4. 压测代码 main.cpp（实测吞吐）

#include "lockfree/lockfree_queue.hpp"
#include <iostream>
#include <thread>
#include <vector>
#include <chrono>
#include <atomic>

int main()
{
LockFreeQueue<int> q;
const int per_thread_cnt = 500000;
const int producer_cnt = 4;
const int consumer_cnt = 4;
std::vector<std::thread> threads;
auto start = std::chrono::high_resolution_clock::now();

// 生产者
for(int i = 0; i < producer_cnt; i++)
{
threads.emplace_back([&q, i, per_thread_cnt](){
for(int j = 0; j < per_thread_cnt; j++)
{
q.enqueue(std::move(i * per_thread_cnt + j));
}
});
}

std::atomic<long long> sum{0};
// 消费者
for(int i = 0; i < consumer_cnt; i++)
{
threads.emplace_back([&q, &sum](){
int val;
while(q.dequeue(val))
{
sum += val;
}
});
}

for(auto& t : threads)
t.join();
auto end = std::chrono::high_resolution_clock::now();
auto dur = std::chrono::duration_cast<std::chrono::milliseconds>(end – start);
std::cout << "耗时：" << dur.count() << " ms\\n";
std::cout << "sum = " << sum.load() << std::endl;
return 0;
}

编译命令：

g++ main.cpp -o lockfree_test -std=c++17 -pthread -O2
./lockfree_test

必须加 -O2 优化，否则性能完全失真，生产环境也是开启优化。

5. 实测效果 & 对比（重点，贴合实际使用）

测试环境：4 核 8 线程 CPU

• 4 生产者 + 4 消费者，每个生产者写入 50 万条数据

✅ 适合场景：高并发网关、日志队列、IO 事件队列、消息分发 ⚠️ 不适合场景：低并发、CPU 资源紧张机器，自旋会浪费 CPU，此时 mutex 更合适。

6. 生产环境优化建议【新增章节】

上面代码是工程可用基础版，落地到线上项目，还可以做下面几轮优化，兼顾性能、稳定性、资源开销。

6.1 内存分配优化

alignas(64) std::atomic<Node*> head_;
alignas(64) std::atomic<Node*> tail_;

6.2 自旋策略优化（降低 CPU 空转）

原生 while (true) 无限自旋会把 CPU 打满。

• 短自旋：CAS 失败先循环几十次重试；多次失败后调用std::this_thread::yield()让出时间片；
• 低负载场景：自旋一定次数后短暂 sleep，减少 CPU 占用；

适用：不想占用过高 CPU 的服务；代价：延迟轻微上升。

6.3 Hazard Pointer 优化

6.4 功能与健壮性优化

6.5 编译与平台调优

6.6 业务选型优化

7. 面试 & 工程问答

核心两点：节点并发 delete 会野指针崩溃；存在 ABA 问题。单纯版本号无法解决指针内存回收，Hazard Pointer 是工业界主流方案。

线程访问节点前标记该节点正在使用；回收节点延迟释放，确认没有任何线程引用后再 delete。

不是！低并发场景 mutex 性能更好。 只有线程多、锁竞争激烈时，无锁才能体现优势；代价是 CPU 自旋，功耗上升。

Epoch Based Reclamation (EBR)、Quiescent State Based Reclamation，HP 实现最简单，适合入门。

8. 系列预告

高性能 C++ 实战系列：

全部代码上传开源仓库，O2 编译直接跑压测。