在C++中如何实现线程安全的队列

在这里插入图片描述

个人主页：个人主页
个人专栏：《数据结构》《C语言》《C++》《Linux》《网络》《redis学习笔记》

文章目录

前言
如何实现一个线程安全的队列思路
应用场景
代码实现
总结

前言

在这里插入图片描述
在一次和豆包的模拟面试中，豆包问我：“在C++中，如何实现一个线程安全的队列呢？”

根据C++标准，STL容器的线程安全性遵循以下规则：

只读操作是线程安全的：多个线程可以同时调用const 成员函数(如size, empty, at等)读取同一个容器，只要没有线程修改容器
写操作需要独占访问：如果至少有一个线程在修改容器(如push_back, earse, operator[]等)，其它线程必须通过同步机制，来保护对该容器的访问
不同容器实例独立：不同线程操作不同的容器实例。

但如果多个线程同时修改同一个容器，或一个线程修改，另一个线程读取都是线程不安全的

在这里插入图片描述

如何实现一个线程安全的队列思路

实现步骤：

使用std::queue作为底层容器
使用std::mutex保护队列的访问
使用std::condition_variable协调线程，特别是在队列空时等待
在push时获取锁，添加元素后通知一个等待的线程
在pop时，使用while循环等待队列非空，处理虚假唤醒
提供tryPop和waitAndPop等不同方法，以适用不同场景
考虑异常安全，使用lock_guard或unique_lock管理锁的声明周期

在C++中实现线程安全队列的核心就在于通过同步机制保护共享数据的访问，并协调生产者和消费者线程操作

应用场景

生产者-消费者模型：多个生产者线程向队列添加任务，消费者线程处理任务
线程池任务队列：线程池使用线程安全的队列分发任务，支持异步返回值(std::future)
事件驱动系统：管理异步事件，确保按顺序处理回调

代码实现

#include <queue>
#include <mutex>
#include <condition_variable>

template <typename T>
class ThreadSafeQueue {
public:
    // 入队操作
    // 获取锁后添加元素，并通知一个等待线程
    void push(T value) {
        std::lock_guard<std::mutex> lock(_mtx);
        if(_finish)  
            return ;

        _data_queue.push(value);
        _cv.notify_one();
    };

    // 非阻塞出队
    // 立即尝试获取元素，若队列为空则返回失败
    bool tryPop(T& value) {
        std::lock_guard<std::mutex> lock(_mtx);
        if(_data_queue.empty() || _finish)
            return false;   // 队列为空 or 终止符为true
    
        value = _data_queue.front();
        _data_queue.pop();
        return true;
    };
    
    // 阻塞出队
    // 等待队列非空后获取元素，处理虚假唤醒
    void waitAndPop(T& value) {
        std::lock_guard<std::mutex> lock(_mtx);
        _cv.wait(lock, [this](){
            return !_data_queue.empty() || _finish;
        });

        if(finish)
            return ;
        value = _data_queue.front();
        _data_queue.pop();
        return ;
    };
        
    // 终止队列
    // 设置终止标志并唤醒所有线程
    void finish() {
        std::lock_guard<std::mutex> lock(_mtx);
        _finish = true;
        _cv.notify_all();
    };

    // 判断队列是否为空
    bool empty() const {
        std::lock_guard<std::mutex> lock(_mtx);
        return _data_queue.empty();
    };

    // 获取队列元素个数
    int size()  {
        std::lock_guard<std::mutex> lock(_mtx);
        return _data_queue.size();
    }
private:
    std::queue<T> _data_queue;       // 底层容器就是queue
    std::mutex _mtx;                 // 互斥锁
    std::condition_variable _cv;     // 信号量 
    bool _finish = false;            // 终止标志
};