一、问题场景

互斥锁std::mutex是一种最常见的线程间同步的手段,但是在有些情况下不太高效。

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

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

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

void function_1() {
    int count = 10;
    while (count > 0) {
        std::unique_lock<std::mutex> locker(g_mutex);
        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(g_mutex);
        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 占用率会很高,我这里达到了快将近 30%。如图:

【C++11 多线程】条件变量(Condition Variable)(七)


解决办法之一是给消费者也加一个小延时,如果一次判断后,发现队列是空的,就惩罚一下自己,延时500ms,这样可以减小 CPU 的占用率。

void function_2() {
    int data = 0;
    while ( data != 1) {
        std::unique_lock<std::mutex> locker(g_mutex);
        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
        }
    }
}

运行程序,CPU 占用变小了很多。如图:

【C++11 多线程】条件变量(Condition Variable)(七)


然后困难之处在于,如何确定这个延时时间呢,假如生产者生产的很快,消费者却延时500ms,也不是很好,如果生产者生产的更慢,那么消费者延时500ms,还是不必要的占用了CPU。

这时候我们设想,能否设计这样的一种机制,如果在队列没有数据的时候,消费者线程能一直阻塞在那里,等待着别人给它唤醒,在生产者往队列中放入数据的时候通知一下这个等待线程,唤醒它,告诉它可以来取数据了。

于是多线程中的条件变量就横空出世!

二、条件变量

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 g_mutex;
std::condition_variable cond;

void function_1() {
    int count = 10;
    while (count > 0) {
        std::unique_lock<std::mutex> locker(g_mutex);
        q.push_front(count);
        locker.unlock();
        cond.notify_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(g_mutex);
        while (q.empty())
            cond.wait(locker); // 解锁,休眠等待notify_one()唤醒
        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的占用率也很低。如图:

【C++11 多线程】条件变量(Condition Variable)(七)


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

  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没有lockunlock接口,而unique_lock提供了。这就是必须使用unique_lock的原因。

  3. 使用细粒度锁,尽量减小锁的范围,在notify_one()的时候,不需要处于互斥锁的保护范围内,所以在唤醒条件变量之前可以将锁unlock()

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

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++11 还提供了notify_all()函数,可以同时唤醒所有处于wait状态的条件变量。

三、扩展:应用场景

我们创建一个基于网络的应用程序,处理如下的任务:

  1. 与处理器进行一些握手操作;
  2. 从 xml 文件 load 数据;
  3. 处理从 xml 文件 load 的数据。

可以发现,任务 1 不依赖其他的任务,而任务 3 则依赖于任务 2,这意味着任务 1 和任务 2 可以由不同的线程并行运行,以提升程序性能。因此,让我们将其分解成一个多线程的应用程序。

线程 1 的任务是:

  • 从 xml 获取数据
  • 通知另一个线程,即等待消息

线程 2 的任务是:

  • 与服务器进行握手操作
  • 等待线程 1 从 xml 加载数据
  • 处理从 xml 获取的数据

实现代码如下:

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

class Application {
public:
    Application() {
        m_bDataLoaded = false;
    }

    // 加载xml数据线程(线程1)
    void loadData() {
        // 使该线程sleep 1秒
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        std::cout << "Loading Data from XML" << std::endl;

        // lock_guard保护数据
        std::lock_guard<std::mutex> guard(m_mutex);
        // flag设为true,表明数据已加载
        m_bDataLoaded = true;
        // 通知条件变量
        m_condVar.notify_one();
    }

    bool isDataLoaded() {
        return m_bDataLoaded;
    }

    // 主线程(线程2)
    void mainTask() {
        std::cout << "Do some handshaking" << std::endl;

        // 获取锁
        std::unique_lock<std::mutex> mlock(m_mutex);

        // 开始等待条件变量得到信号
        // wait()将在内部释放锁,并使线程阻塞
        // 一旦条件变量发出信号,则恢复线程并再次获取锁
        // 然后检测条件是否满足,如果条件满足,则继续,否则再次进入wait
        m_condVar.wait(mlock, std::bind(&Application::isDataLoaded, this));
        std::cout << "Do Processing On loaded Data" << std::endl;
    }

private:
    std::mutex m_mutex;
    std::condition_variable m_condVar;
    bool m_bDataLoaded;
};

int main() {
    Application app;
    std::thread thread_1(&Application::mainTask, &app);
    std::thread thread_2(&Application::loadData, &app);
    thread_2.join();
    thread_1.join();

    return 0;
}

/*
输出:
Do some handshaking
Loading Data from XML
Do Processing On loaded Data
*/

参考:

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

c++11多线程编程(七):条件变量说明


分类:

技术点:

相关文章: