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

016-condition_variable

article 2025/3/15 13:33:03

condition_variable

以下是一篇关于std::condition_variable实现异步执行同步等待的调研文章，包含完整代码及解析：

基于std::condition_variable的异步执行同步等待机制

一、引言

在多线程编程中，std::condition_variable是C++11标准库提供的核心同步原语，用于实现线程间的等待-通知机制。其核心价值在于允许线程高效等待特定条件的达成，而非通过轮询消耗CPU资源。本文结合生产者-消费者模型，探讨其在异步任务同步中的典型应用场景。

二、核心原理

组件协作

互斥锁 (std::mutex): 保护共享数据的原子性访问。
条件变量 (std::condition_variable): 实现线程阻塞与唤醒的通信机制。
唯一锁 (std::unique_lock): 配合条件变量实现自动锁管理，避免死锁。

关键操作

wait(): 阻塞线程直至收到通知，自动释放/重获锁。
notify_one()/notify_all(): 唤醒单个或全部等待线程。
带谓词的wait(): 防止虚假唤醒（Spurious Wakeup）。

三、关键问题与解决方案

虚假唤醒

现象: 线程可能未收到通知即被唤醒。
方案: 使用带谓词的wait()，循环验证条件是否成立：

cv.wait(lock,  [&]{ return !queue.empty();  });

丢失唤醒

现象: 通知先于线程进入等待状态发出，导致永久阻塞。
方案: 确保通知操作在共享状态修改后触发，且使用原子性条件检查。

性能优化

批量通知: notify_all()适用于多消费者场景，notify_one()减少无效唤醒。
超时控制: wait_for()/wait_until()避免无限期阻塞。

四、完整代码示例

线程安全任务队列实现

#include <iostream>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <future>

class ThreadSafeQueue {
public:
    void Push(std::function<void()> task) {
        std::lock_guard<std::mutex> lock(mtx_);
        queue_.push(std::move(task));
        cv_.notify_one();  // 通知一个等待线程 
    }

    bool TryPop(std::function<void()>& task) {
        std::lock_guard<std::mutex> lock(mtx_);
        if (queue_.empty()) return false;
        task = std::move(queue_.front());
        queue_.pop();
        return true;
    }

    void WaitPop(std::function<void()>& task) {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_.wait(lock, [this]{ return !queue_.empty() || stop_; });
        if (stop_) return;  // 处理终止信号 
        task = std::move(queue_.front());
        queue_.pop();
    }

    void Stop() {
        std::lock_guard<std::mutex> lock(mtx_);
        stop_ = true;
        cv_.notify_all();  // 唤醒所有线程退出 
    }

private:
    std::queue<std::function<void()>> queue_;
    std::mutex mtx_;
    std::condition_variable cv_;
    bool stop_ = false;
};

// 异步任务处理器 
class AsyncProcessor {
public:
    AsyncProcessor() : worker_([this]{ ProcessTasks(); }) {}

    ~AsyncProcessor() {
        queue_.Stop();
        worker_.join();
    }

    std::future<int> SubmitTask(int input) {
        auto promise = std::make_shared<std::promise<int>>();
        std::future<int> future = promise->get_future();
        queue_.Push([input, promise]{
            int result = input * 2; // 模拟耗时操作 
            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            promise->set_value(result);
        });
        return future;
    }

private:
    using Task = std::function<void()>;
    ThreadSafeQueue queue_;
    std::thread worker_;

    void ProcessTasks() {
        while (true) {
            Task task;
            queue_.WaitPop(task);
            if (!task) break; // 收到终止信号 
            task(); // 执行实际任务 
        }
    }
};

// 使用示例 
int main() {
    AsyncProcessor processor;
    auto future = processor.SubmitTask(42);
    
    // 同步等待结果 
    std::cout << "start wait Result ..." << std::endl;
    std::cout << "Result: " << future.get()  << std::endl;
    return 0;
}