并发编程 —— 使用条件变量构建线程安全队列

使用条件变量处理数据等待,看下面的例子：

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

using namespace std;

bool more_data_to_prepare();
int prepare_data();
bool data_queue_empty();
void process(int data);
bool is_last_data(int data);

std::mutex mut;
std::queue<int> data_queue;
std::condition_variable data_cond;

//数据接收线程
void data_preparation_thread() {
    while(more_data_to_prepare()) {
        int const data = prepare_data();
        std::lock_guard<std::mutex> lock(mut);
        data_queue.push(data);   //压入元素
        data_cond.notify_one();  //唤醒等待的线程
    }
}

//数据处理线程
void data_processing_thread() {
    while(true) {
        std::unique_lock<std::mutex> lock(mut);
        data_cond.wait(lock, []{return !data_queue_empty();});
        int data = data_queue.front(); //返回队首元素
        data_queue.pop();  //弹出队首元素
        lock.unlock();     //及时释放锁资源
        process(data);     //处理数据
        if(is_last_data(data))
            break;
    }
}

有几点需要注意：

• wait函数的谓词参数返回真时，即条件满足时，wait才会返回，否则wait函数会将第一个参数unique_lock中的互斥量解锁，并将该线程置于阻塞或等待状态。
• wait()函数中使用unique_lock()，而不是lock_guard()，是因为在条件不满足时，要调用unlock()解锁互斥量，而lock_guard只能在生命周期结束时解锁，不支持用户解锁。

在多个线程中使用队列转移数据很常见，我们可以重写一个通用线程安全队列，将同步操作限制在队列内部。

队列实现如下:

#include <iostream>
#include <mutex>
#include <memory>
#include <queue>
#include <condition_variable>

bool more_data_to_prepare();
int prepare_data();
bool data_queue_empty();
void process(int data);
bool is_last_data(int data);

template <typename T>
class threadsafe_queue {
private:
    mutable std::mutex mut;   //互斥量必须是可变的
    std::queue<T> data_queue;
    std::condition_variable data_cond;
public:
    threadsafe_queue() {}
    threadsafe_queue(threadsafe_queue const& other) {
        std::lock_guard<std::mutex> lk(other.mut);
        data_queue = other.data_queue;
    }

    void push(T new_value) {
        std::lock_guard<std::mutex> lk(mut);
        data_queue.push(new_value);
        data_cond.notify_one();
    }

    void wait_and_pop(T value) {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk, [this]{return !data_queue_empty();});
        value = data_queue.front();
        data_queue.pop();
    }

    std::shared_ptr<T> wait_and_pop() {
        std::unique_lock<std::mutex> lk(mut);
        data_cond.wait(lk, [this]{return !data_queue_empty();});
        std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return res;
    }

    bool try_pop(T& value) {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
            return false;
        value = data_queue.front();
        data_queue.pop();
        return true;
    }

    std::shared_ptr<T> try_pop() {
        std::lock_guard<std::mutex> lk(mut);
        if(data_queue.empty())
            return std::shared_ptr<T>();
        std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
        data_queue.pop();
        return true;
    }

    bool empty() const {
        std::lock_guard<std::mutex> lk(mut);
        return data_queue.empty();
    }
};

threadsafe_queue<int> data_queue;

//数据接收线程
void data_preparation_thread() {
    while(more_data_to_prepare()) {
        int const data = prepare_data();
        data_queue.push(data);
    }
}

//数据处理线程
void data_processing_thread() {
    while(true) {
        int data;
        data_queue.wait_and_pop(data);
        process(data);     //处理数据
        if(is_last_data(data))
            break;
    }
}