在高并发编程场景中，多线程对共享数据结构的访问需要保证线程安全，传统互斥锁在竞争时会触发线程上下文切换，带来额外的性能开销。自旋锁通过忙等待的方式避免上下文切换，更适合短临界区的并发场景，而C++标准库提供的atomic_flag是实现自旋锁的理想原子类型，它保证了操作的原子性且无锁特性。
自旋锁的实现原理
自旋锁的核心逻辑是当一个线程尝试获取锁时，如果锁已经被其他线程持有，该线程会循环检查锁的状态直到获取成功，而不是进入休眠状态。C++的atomic_flag类型只有两种状态：设置和清除，它的test_and_set操作是原子性的，非常适合用来实现自旋锁的获取和释放逻辑。
基础自旋锁实现
下面是一个基于atomic_flag的简单自旋锁实现：
#include <atomic>
class SpinLock {
private:
// atomic_flag默认初始化为清除状态
std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
// 获取锁，自旋等待直到成功
void lock() {
// test_and_set原子地设置flag为true，返回之前的状态
// 如果之前是false，说明获取锁成功，退出循环
// 如果之前是true，说明锁被持有，继续循环
while (flag.test_and_set(std::memory_order_acquire)) {
// 可以在这里加入pause指令优化CPU占用，不同平台实现不同
// 例如x86平台可以用__asm__ __volatile__("pause")
}
}
// 释放锁
void unlock() {
// 清除flag状态，释放锁
flag.clear(std::memory_order_release);
}
};
这里的memory_order_acquire和memory_order_release是内存序参数，保证了锁操作前后的内存可见性，避免指令重排带来的问题。
基于自旋锁的高并发队列实现
有了自旋锁之后，我们可以构建一个线程安全的队列，队列的入队和出队操作都需要加锁保证同一时间只有一个线程修改队列结构。这里采用单向链表作为队列的底层存储结构，方便实现入队和出队操作。
队列结构设计
队列包含头节点和尾节点，头节点是哨兵节点，不存储实际数据，尾节点指向最后一个存储数据的节点。自旋锁用来保护对头尾节点和节点数据的修改操作。
完整源码实现
#include <atomic>
#include <memory>
// 自旋锁定义，同上
class SpinLock {
private:
std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
void lock() {
while (flag.test_and_set(std::memory_order_acquire)) {
// 可选：加入CPU pause优化
}
}
void unlock() {
flag.clear(std::memory_order_release);
}
};
// 队列节点结构
template <typename T>
struct QueueNode {
T data;
std::unique_ptr<QueueNode<T>> next;
QueueNode(T val) : data(std::move(val)), next(nullptr) {}
};
// 高并发队列实现
template <typename T>
class ConcurrentQueue {
private:
// 哨兵头节点
std::unique_ptr<QueueNode<T>> head;
// 尾节点指针，指向最后一个数据节点
QueueNode<T>* tail;
// 自旋锁实例
SpinLock lock;
public:
ConcurrentQueue() {
// 初始化头节点为哨兵节点，尾节点指向头节点
head = std::make_unique<QueueNode<T>>(T());
tail = head.get();
}
// 禁止拷贝构造和赋值
ConcurrentQueue(const ConcurrentQueue&) = delete;
ConcurrentQueue& operator=(const ConcurrentQueue&) = delete;
// 入队操作
void enqueue(T val) {
// 创建新节点
auto new_node = std::make_unique<QueueNode<T>>(std::move(val));
QueueNode<T>* new_node_ptr = new_node.get();
lock.lock();
// 将新节点接到尾节点后面
tail->next = std::move(new_node);
// 更新尾节点指针
tail = new_node_ptr;
lock.unlock();
}
// 出队操作，成功返回true，失败（队列为空）返回false
bool dequeue(T& result) {
lock.lock();
// 检查队列是否为空：头节点的next为空说明没有数据节点
if (head->next == nullptr) {
lock.unlock();
return false;
}
// 取出第一个数据节点
auto first_data_node = std::move(head->next);
result = std::move(first_data_node->data);
// 更新头节点为原来的第一个数据节点，原来的哨兵节点会被自动释放
head = std::move(first_data_node);
lock.unlock();
return true;
}
// 判断队列是否为空
bool empty() {
lock.lock();
bool is_empty = (head->next == nullptr);
lock.unlock();
return is_empty;
}
};
代码测试示例
下面是一个简单的多线程测试示例，验证队列的线程安全性：
#include <iostream>
#include <thread>
#include <vector>
int main() {
ConcurrentQueue<int> queue;
const int THREAD_NUM = 4;
const int OPER_NUM = 1000;
// 生产者线程，每个线程入队1000个数据
std::vector<std::thread> producers;
for (int i = 0; i < THREAD_NUM; ++i) {
producers.emplace_back([&queue, i]() {
for (int j = 0; j < OPER_NUM; ++j) {
queue.enqueue(i * OPER_NUM + j);
}
});
}
// 消费者线程，每个线程出队直到队列为空
std::vector<std::thread> consumers;
std::atomic<int> total_count(0);
for (int i = 0; i < THREAD_NUM; ++i) {
consumers.emplace_back([&queue, &total_count]() {
int val;
while (queue.dequeue(val)) {
total_count.fetch_add(1, std::memory_order_relaxed);
}
});
}
// 等待所有生产者完成
for (auto& t : producers) {
t.join();
}
// 等待所有消费者完成
for (auto& t : consumers) {
t.join();
}
std::cout << "Total dequeued elements: " << total_count << std::endl;
// 预期输出4000，即4个线程每个入队1000个
return 0;
}
注意事项
- 自旋锁适合临界区执行时间短的场景，如果临界区执行时间长，忙等待会浪费大量CPU资源，此时更适合用互斥锁。
atomic_flag是C++中唯一保证无锁的原子类型，其他原子类型不一定是无锁的，使用前可以通过is_always_lock_free检查。- 本示例的队列实现中，自旋锁保护了整个入队出队操作，实际场景中可以根据需求优化锁粒度，进一步提升并发性能。
- 内存序的选择需要根据实际场景调整，本示例使用的acquire和release语义可以满足大部分自旋锁的场景需求。
C++atomic_flag自旋锁高并发队列修改时间：2026-07-19 18:57:29