[c++11]多线程编程(六)——条件变量(Condition Variable)

互斥锁std::mutex是一种最多见的线程间同步的手段，可是在有些状况下不过高效。

假设想实现一个简单的消费者生产者模型，一个线程往队列中放入数据，一个线程往队列中取数据，取数据前须要判断一下队列中确实有数据，因为这个队列是线程间共享的，因此，须要使用互斥锁进行保护，一个线程在往队列添加数据的时候，另外一个线程不能取，反之亦然。用互斥锁实现以下：

#include <iostream>
#include <deque>
#include <thread>
#include <mutex>

std::deque<int> q;
std::mutex mu;

void function_1() {
    int count = 10;
    while (count > 0) {
        std::unique_lock<std::mutex> locker(mu);
        q.push_front(count);
        locker.unlock();
        std::this_thread::sleep_for(std::chrono::seconds(1));
        count--;
    }
}

void function_2() {
    int data = 0;
    while ( data != 1) {
        std::unique_lock<std::mutex> locker(mu);
        if (!q.empty()) {
            data = q.back();
            q.pop_back();
            locker.unlock();
            std::cout << "t2 got a value from t1: " << data << std::endl;
        } else {
            locker.unlock();
        }
    }
}
int main() {
    std::thread t1(function_1);
    std::thread t2(function_2);
    t1.join();
    t2.join();
    return 0;
}

//输出结果
//t2 got a value from t1: 10
//t2 got a value from t1: 9
//t2 got a value from t1: 8
//t2 got a value from t1: 7
//t2 got a value from t1: 6
//t2 got a value from t1: 5
//t2 got a value from t1: 4
//t2 got a value from t1: 3
//t2 got a value from t1: 2
//t2 got a value from t1: 1

能够看到，互斥锁其实能够完成这个任务，可是却存在着性能问题。

首先，function_1函数是生产者，在生产过程当中，std::this_thread::sleep_for(std::chrono::seconds(1));表示延时1s，因此这个生产的过程是很慢的；function_2函数是消费者，存在着一个while循环，只有在接收到表示结束的数据的时候，才会中止，每次循环内部，都是先加锁，判断队列不空，而后就取出一个数，最后解锁。因此说，在1s内，作了不少无用功！这样的话，CPU占用率会很高，可能达到100%（单核）。如图：

解决办法之一是给消费者也加一个小延时，若是一次判断后，发现队列是空的，就惩罚一下本身，延时500ms，这样能够减少CPU的占用率。

void function_2() {
    int data = 0;
    while ( data != 1) {
        std::unique_lock<std::mutex> locker(mu);
        if (!q.empty()) {
            data = q.back();
            q.pop_back();
            locker.unlock();
            std::cout << "t2 got a value from t1: " << data << std::endl;
        } else {
            locker.unlock();
            std::this_thread::sleep_for(std::chrono::milliseconds(500));
        }
    }
}

如图：

而后困难之处在于，如何肯定这个延时时间呢，假如生产者生产的很快，消费者却延时500ms，也不是很好，若是生产者生产的更慢，那么消费者延时500ms，仍是没必要要的占用了CPU。

这就引出了条件变量（condition variable）,c++11中提供了#include <condition_variable>头文件，其中的std::condition_variable能够和std::mutex结合一块儿使用，其中有两个重要的接口，notify_one()和wait()，wait()可让线程陷入休眠状态，在消费者生产者模型中，若是生产者发现队列中没有东西，就可让本身休眠，可是不能一直不干活啊，notify_one()就是唤醒处于wait中的其中一个条件变量（可能当时有不少条件变量都处于wait状态）。那什么时刻使用notify_one()比较好呢，固然是在生产者往队列中放数据的时候了，队列中有数据，就能够赶忙叫醒等待中的线程起来干活了。

使用条件变量修改后以下：

#include <iostream>
#include <deque>
#include <thread>
#include <mutex>
#include <condition_variable>

std::deque<int> q;
std::mutex mu;
std::condition_variable cond;

void function_1() {
    int count = 10;
    while (count > 0) {
        std::unique_lock<std::mutex> locker(mu);
        q.push_front(count);
        locker.unlock();
        cond.notify_one();  // Notify one waiting thread, if there is one.
        std::this_thread::sleep_for(std::chrono::seconds(1));
        count--;
    }
}

void function_2() {
    int data = 0;
    while ( data != 1) {
        std::unique_lock<std::mutex> locker(mu);
        while(q.empty())
            cond.wait(locker); // Unlock mu and wait to be notified
        data = q.back();
        q.pop_back();
        locker.unlock();
        std::cout << "t2 got a value from t1: " << data << std::endl;
    }
}
int main() {
    std::thread t1(function_1);
    std::thread t2(function_2);
    t1.join();
    t2.join();
    return 0;
}

此时CPU的占用率也很低。

上面的代码有三个注意事项：

1. 在function_2中，在判断队列是否为空的时候，使用的是while(q.empty())，而不是if(q.empty())，这是由于wait()从阻塞到返回，不必定就是因为notify_one()函数形成的，还有可能因为系统的不肯定缘由唤醒（可能和条件变量的实现机制有关），这个的时机和频率都是不肯定的，被称做伪唤醒，若是在错误的时候被唤醒了，执行后面的语句就会错误，因此须要再次判断队列是否为空，若是仍是为空，就继续wait()阻塞。
2. 在管理互斥锁的时候，使用的是std::unique_lock而不是std::lock_guard，并且事实上也不能使用std::lock_guard，这须要先解释下wait()函数所作的事情。能够看到，在wait()函数以前，使用互斥锁保护了，若是wait的时候什么都没作，岂不是一直持有互斥锁？那生产者也会一直卡住，不可以将数据放入队列中了。因此，wait()函数会先调用互斥锁的unlock()函数，而后再将本身睡眠，在被唤醒后，又会继续持有锁，保护后面的队列操做。而lock_guard没有lock和unlock接口，而unique_lock提供了。这就是必须使用unique_lock的缘由。
3. 使用细粒度锁，尽可能减少锁的范围，在notify_one()的时候，不须要处于互斥锁的保护范围内，因此在唤醒条件变量以前能够将锁unlock()。

还能够将cond.wait(locker);换一种写法，wait()的第二个参数能够传入一个函数表示检查条件，这里使用lambda函数最为简单，若是这个函数返回的是true，wait()函数不会阻塞会直接返回，若是这个函数返回的是false，wait()函数就会阻塞着等待唤醒，若是被伪唤醒，会继续判断函数返回值。

void function_2() {
    int data = 0;
    while ( data != 1) {
        std::unique_lock<std::mutex> locker(mu);
        cond.wait(locker, [](){ return !q.empty();} );  // Unlock mu and wait to be notified
        data = q.back();
        q.pop_back();
        locker.unlock();
        std::cout << "t2 got a value from t1: " << data << std::endl;
    }
}

除了notify_one()函数，c++还提供了notify_all()函数，能够同时唤醒全部处于wait状态的条件变量。

参考

1. C++并发编程实战
2. C++ Threading #6: Condition Variable