单生产者多消费者FIFO队列锁机制异常问题求助

问题根源 #

• 无锁状态下读取队列大小：get()和put()方法中，对my_queue.size()的检查都未加锁保护。以get()为例：当多个消费者线程同时看到队列非空（my_queue.size() != 0），其中一个线程先获取锁并取走最后一个元素，后续线程获取锁时队列已为空，此时调用my_queue.front()会触发未定义行为（崩溃）。
• 检查与操作的原子性被破坏：队列状态检查和后续的入队/出队操作必须在同一个锁的保护下，否则两次操作之间的队列状态可能被其他线程修改，导致逻辑错误。

修复后的代码 #

#pragma once
#include <optional>
#include <queue>
#include <mutex>
#include <condition_variable>

template<class T, uint32_t MAX_SIZE>
class spmc_fifo_queue
{
public:
    using optional_queue_pair = std::optional<T>;

    bool put(optional_queue_pair&& p)
    {
        std::unique_lock my_lock(my_mutex);
        // 等待队列有空闲空间，条件判断在锁保护下执行
        my_cv_remove.wait(my_lock, [this] { return my_queue.size() < MAX_SIZE; });
        
        my_queue.push(std::move(p));
        my_cv_add.notify_one();
        return true;
    }

    optional_queue_pair get()
    {
        std::unique_lock my_lock(my_mutex);
        // 等待队列非空，条件判断在锁保护下执行
        my_cv_add.wait(my_lock, [this] { return my_queue.size() != 0; });
        
        optional_queue_pair ret = std::move(my_queue.front());
        my_queue.pop();
        my_cv_remove.notify_one();
        return ret;
    }

private:
    std::queue<optional_queue_pair> my_queue;
    std::mutex my_mutex;
    std::condition_variable my_cv_add;
    std::condition_variable my_cv_remove;
};

关键修改说明 #

• 移除无锁状态检查：直接使用std::unique_lock获取锁，调用wait()方法——wait()会自动释放锁并等待条件，被唤醒后重新获取锁并检查条件，确保条件满足后才继续执行。
• 合并锁的作用范围：将队列状态检查、入队/出队操作全部放在同一个锁的保护下，保证这些操作的原子性，彻底避免竞态条件。
• 清理冗余定义：移除原代码中多次重复定义的optional_queue_pair，仅在类的public区域定义一次，提升代码整洁度。

补充说明 #

• std::condition_variable::wait()的谓词参数会在线程被唤醒后重新检查，能有效避免虚假唤醒带来的逻辑问题。
• 单生产者多消费者场景下，notify_one()足以唤醒一个等待的消费者线程；若扩展为多生产者场景，可根据需求改用notify_all()，但会带来一定性能开销。

内容的提问来源于stack exchange，提问作者Suno