C++同实例成员函数中多消费者条件变量等待实现方案

初始bool标记方案的问题 #

你第一版用单bool标记新数据的写法存在本质缺陷：

• 只要有一个消费者醒来把m_newData置为false，其他还阻塞在wait的消费者要么错过本次更新，要么要等下一次生产者发信号才能醒来，无法保证所有消费者都拿到最新数据
• 如果不及时重置标记，所有线程会被同一份数据反复唤醒，完全不符合不重复处理旧数据的要求
• 根本无法实现「错过中间更新直接拿最新值」的语义，只能支持单消费者的简单通知场景。

线程ID映射版本号的思路评价 #

你第二版基于线程ID存储每个线程已处理版本号的思路方向完全正确，是实现这类「多消费者只拿最新值、跳过中间帧」通知机制的标准思路，但现有代码有几个明确问题：

• 存在未定义行为：Wait函数持有的是共享读锁，最后直接修改m_threadCtrMap里的线程版本号属于写操作，共享锁不允许并发写，会触发数据竞争。
• 使用成本高：强制要求消费者提前调用RegisterWaiter注册，容易漏调用引发bug。
• 存在无意义的内存占用：如果消费线程动态创建销毁，map里残留的退出线程条目不会自动清理，长时间运行会持续占用内存。

修正后的生产可用实现 #

下面的实现完全满足你的所有需求：

• 彻底规避虚假唤醒，没有新数据时线程不会误唤醒
• 同一份数据永远不会被同一个线程重复处理
• 线程忙处理时错过的中间更新直接跳过，唤醒后永远拿到最新值
• 自动完成线程注册，不需要手动调用注册接口
• 保留共享读锁特性，多消费者读数据可以完全并发，读多写少场景下性能远高于普通互斥锁

#include <functional>
#include <shared_mutex>
#include <condition_variable>
#include <unordered_map>
#include <thread>
#include <cstdint>
#include <chrono>
#include <cstdlib>

class LatestDataSignaller {
public:
    // 生产者调用：执行写操作更新数据，版本号自增后通知所有等待线程
    void Signal(const std::function<void()>& write_fn) {
        std::unique_lock lock(m_mtx);
        write_fn();
        m_version++;
        m_cv.notify_all();
    }

    // 消费者调用：阻塞直到有比上次处理更新的数据，持读锁执行读回调
    void Wait(const std::function<void()>& read_fn) {
        std::shared_lock read_lock(m_mtx);
        const std::thread::id tid = std::this_thread::get_id();
        
        // 首次进入自动注册当前线程，不需要单独调用注册接口
        if (!m_thread_ver.contains(tid)) {
            read_lock.unlock();
            {
                std::unique_lock write_lock(m_mtx);
                m_thread_ver.try_emplace(tid, m_version);
            }
            read_lock.lock();
        }

        // 谓词判断天然防虚假唤醒：只有全局版本和当前线程已处理版本不一致才会唤醒
        m_cv.wait(read_lock, [&]() {
            return m_thread_ver[tid] != m_version;
        });

        // 持共享锁执行读操作，多消费者可并发执行，不会阻塞其他消费者读
        read_fn();

        // 更新当前线程已处理版本，短暂升级为独占锁避免写竞争
        read_lock.unlock();
        {
            std::unique_lock write_lock(m_mtx);
            m_thread_ver[tid] = m_version;
        }
    }

private:
    std::condition_variable_any m_cv;
    std::shared_mutex m_mtx;
    uint64_t m_version = 0; // 全局数据版本，每次生产者更新自增
    std::unordered_map<std::thread::id, uint64_t> m_thread_ver; // 每个线程最后处理的版本号
};

// Example使用方式和你原有逻辑完全兼容
class Example {
public:
    void ConsumerLoop() 
    {
        int latestData = 0;
        while (true) {
            m_signaller.Wait([this, &latestData]() {
                latestData = m_latestData;
            });
            // 直接处理latestData即可，永远是最新值，不会重复处理同一份数据
            // 处理期间产生的多次更新会直接合并，下次唤醒直接拿最新的，不需要回溯中间版本
        }
    }

    void ProducerLoop() {
        while (true) {
            int newData = rand();
            m_signaller.Signal([this, newData]() {
                m_latestData = newData;
            });
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
    }

private:
    LatestDataSignaller m_signaller;
    int m_latestData = 0;
};

优化提示 #

• 如果你的消费线程是固定数量（比如启动时创建N个消费者，运行期间不销毁），当前实现已经是最优，map大小固定为消费者数量，没有额外开销。如果需要动态创建销毁消费线程，可以在更新版本号的逻辑里加简单的清理，删掉已经退出线程的条目即可，固定线程池场景完全不需要这个逻辑。
• 版本号用uint64_t不需要考虑溢出：哪怕每秒产生10亿次更新，也要近600年才会回绕，所有实际业务场景都不会触发问题。
• 如果不需要多消费者并发读的性能优势，可以直接把std::shared_mutex换成普通std::mutex，std::condition_variable_any换成std::condition_variable，代码逻辑完全不变，锁开销会更低。
• 这个实现完全符合条件变量的使用规范，所有谓词判断都在锁保护下执行，不存在竞态问题。

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