基于std::mutex std::lock_guard std::condition_variable 和std::async实现的简单同步队列

Posted 2020-10-18 自由真实个性

C++多线程编程中通常会对共享的数据进行写保护，以防止多线程在对共享数据成员进行读写时造成资源争抢导致程序出现未定义的行为。通常的做法是在修改共享数据成员的时候进行加锁--mutex。在使用锁的时候通常是在对共享数据进行修改之前进行lock操作，在写完之后再进行unlock操作，进场会出现由于疏忽导致由于lock之后在离开共享成员操作区域时忘记unlock，导致死锁。

针对以上的问题，C++11中引入了std::unique_lock与std::lock_guard两种数据结构。通过对lock和unlock进行一次薄的封装，实现自动unlock的功能。std::unique_lock 与std::lock_guard都能实现自动加锁与解锁功能，但是std::unique_lock要比std::lock_guard更灵活，但是更灵活的代价是占用空间相对更大一点且相对更慢一点。

std::condition_variable 是条件变量，更多有关条件变量的定义参考维基百科。Linux 下使用 Pthread 库中的 pthread_cond_*() 函数提供了与条件变量相关的功能， Windows 则参考 MSDN。

当 std::condition_variable 对象的某个 wait 函数被调用的时候，它使用 std::unique_lock(通过 std::mutex) 来锁住当前线程。当前线程会一直被阻塞，直到另外一个线程在相同的 std::condition_variable 对象上调用了 notification 函数来唤醒当前线程。

std::condition_variable 对象通常使用 std::unique_lock<std::mutex> 来等待，如果需要使用另外的 lockable 类型，可以使用 std::condition_variable_any 类，本文后面会讲到 std::condition_variable_any 的用法。

template<typename T>
class SimpleSyncQueue
{
public:
    SimpleSyncQueue(){}
    ~SimpleSyncQueue(){}
    void Put(const T& element)
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        m_queue.push_back(element);
        m_NotEmpty.notify_all();
    }
    void Take(T& result)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_NotEmpty.wait(locker, [this](){cout << "take wait!" << endl; return !m_queue.empty(); });
        result = m_queue.front();
        m_queue.pop_front();
    }
    bool Empty()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.empty();
    }
    size_t Size()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_queue.size();
    }
private:
    std::list<T> m_queue;
    std::mutex m_mutex;
    std::condition_variable m_NotEmpty;
};

SimpleSyncQueue<int> queue;
void Thread1func()
{
    for (int i = 0; i < 10; i++)
    {
        queue.Put(i);
        cout << "size: "<<queue.Size() << endl;
    }
}
void Thread2func()
{
    int res;
    for (int i = 0; i < 5; i++)
    {
        queue.Take(res);
        cout << "teke res: " << res << endl;
    }
}
void Test14()
{
    std::future<void> future1 = std::async(std::launch::async,Thread1func);
    //Sleep(1000);
    std::future<void> future2 = std::async(std::launch::async, Thread2func);
    future1.wait();
    future2.wait();
}

某次执行结果如下：