当前位置：首页 > news >正文

SPSC无锁队列

news 2026/6/14 4:08:39

定位

通信模型

核心原理

内部存有两块缓冲区，写缓冲区与读缓冲区
只允许两个线程对SPSC无锁队列进行操作
1个生产者线程只往写缓冲区写入数据
1个消费者线程只往读缓冲区拿取数据
惰性机制:
消费者线程在拿取数据时，会先判断读缓冲区是否为空
只有在读缓冲区空了的时候才会触发原子交换
原子交换：
SPSC无锁队列内部存有读缓冲区与写缓冲区的原子指针
这两个指针的交换为原子交换
并且只有消费者线程才会触发原子交换，生产者碰不到

声明与实现

#ifndef SPSC_QUEUE_H
#define SPSC_QUEUE_H
#include <memory>
#include <queue>
#include <atomic>
#include <thread>
#include <functional>template <class T>
class SPSCqueue {
public:typedef std::shared_ptr<SPSCqueue> ptr;SPSCqueue();void        push(T task);                       //添加任务,生产者线程专属T           pop();                              //拿取任务,消费者线程专属bool        empty();                            //判空,读缓冲区的判空
private:void        change();                           //原子交换std::queue<T>                   m_writeQueue;   //写缓冲区std::queue<T>                   m_readQueue;    //读缓冲区std::atomic<std::queue<T> *>    m_writePtr;     //写缓冲区的原子指针std::atomic<std::queue<T> *>    m_readPtr;      //读缓冲区的原子指针
};
template <class T>
SPSCqueue<T>::SPSCqueue() {m_writePtr  = &m_writeQueue;m_readPtr   = &m_readQueue;
}
//添加任务,生产者线程专属
template <class T>
void SPSCqueue<T>::push(T task) {(*m_writePtr).push(std::move(task));
}
//拿取任务,消费者线程专属
template <class T>
T SPSCqueue<T>::pop() {if ((*m_readPtr).empty()) {change();}if (!(*m_readPtr).empty()) {auto task = (*m_readPtr).front();(*m_readPtr).pop();return task;}return nullptr;
}
//判空,读缓冲区的判空
template <class T>
bool SPSCqueue<T>::empty() {return (*m_readPtr).empty();
}
//原子交换
template <class T>
void SPSCqueue<T>::change() {std::queue<T> *writePtr = m_writePtr;std::queue<T> *readPtr = m_readPtr;m_writePtr.exchange(readPtr);m_readPtr.exchange(writePtr);
}#endif

使用示例

#include <chrono>
#include "SPSCqueue.h"SPSCqueue<std::function<void()>>::ptr que;
uint64_t cnt = 0;//生产者线程的入口函数
void producer() {for (long long i = 0; i < 1000000LL; ++i) {que->push([]() {for(int z =0;z<1000;++z) {//模拟具体业务耗时}++cnt; });}
}
//消费者线程的入口函数
void consumer() {while (true) {auto it = que->pop();if (it != nullptr) {it();}}
}
int main() {que = std::make_shared<SPSCqueue<std::function<void()>>>();auto before = std::chrono::system_clock::now(); // 记录开始时刻std::thread t2(consumer);                       // 开启消费者线程std::thread t1(producer);                       // 开启生产者线程while (true) {// 这里会线程不安全// 但是后果只是读取旧值// 所以实际测得的的总耗时结果一定更少if (cnt == 1000000) {auto after = std::chrono::system_clock::now(); // 记录结束时刻auto before_ms = std::chrono::duration_cast<std::chrono::milliseconds>(before.time_since_epoch()).count();auto after_ms = std::chrono::duration_cast<std::chrono::milliseconds>(after.time_since_epoch()).count();std::cout << "开始时刻 : " << before_ms << std::endl;std::cout << "结束时刻 : " << after_ms  << std::endl;std::cout << "耗时 : " << after_ms - before_ms << std::endl; // 输出总耗时break;}}t1.detach();t2.detach();return 0;
}