C++11多线程---互斥量、锁、条件变量的总结

关于互斥量`std::mutex`的总结

互斥量用于组成代码的临界区。C++的多线程模型是基于内存的，或者说是基于代码片段的，这和我们操作系统学习的临界区概念基本一致，但是与Golang不同，Golang是基于消息模型的。

一个std::mutex的lock()和unlock()之间的代码片段组成一个临界区，这个临界区内部同时最多只能有一个线程进行访问，可以理解为这个片段内部的代码是受到保护的，不会被多线程同时访问造成不可预知的问题。

std::mutex mtx;
mtx.lock();
// 这里的代码，同时最多只能有一个线程进行访问
mtx.unlokc();

当一个线程获取到一个std::mutex并且调用lock()后，必须由同一个线程调用unlock()操作，因此一定要注意两个函数成对出现，否则会造成死锁。

当一个线程获取一个std::mutex并调用lock()函数后，其他线程的调用会失败。可以使用try_lock进行判别，具体细节参考：https://en.cppreference.com/w/cpp/thread/mutex

关于`std::lock_guard`和`std::unique_lock`的总结

std::lock_guard比较好理解，因为调用mutex需要时刻记着解锁，所用这个类封装了一系列的操作，在一个模块中构造了一个std::lock_guard后，相当于对该结构块加锁，当线程离开结构块后，std::lock_guard自动析构，相当于解锁。它的结构简单、速度快，但是功能比较少。

std::unique_lock是对std::lock_guard功能的一个拓展，功能更多，但是速度会慢一些，具体参照：https://en.cppreference.com/w/cpp/thread/unique_lock

关于条件变量`std::condition_variable`的总结

这个相当于操作系统中的wait和signal原语操作，需要结合一个std::unique_lock组成的临界区共同完成功能。wait & signal原语操作最典型的特点是 “阻塞自己，唤醒别人”。可以这么理解，如果当前满足特定条件不满足，那么就不能进入临界区，当前线程阻塞。当其他线程处理完后，使得条件满足了，线程会唤醒那些处于阻塞状态的线程，使之重新进入。直接通过下面的代码来说明，经典的生产者和消费者问题。

注意只有一个互斥量的时候，唤醒顺序的问题，参照官网：https://en.cppreference.com/w/cpp/thread/condition_variable

在这里给出一个更加简洁的例子：

void worker_thread()
{// Wait until main() sends datastd::unique_lock<std::mutex> lk(m);cv.wait(lk, pred});// 这里执行工作代码，注意下面两个语句的顺序lk.unlock();cv.notify_one();
}

上述代码中，先执行lk.unlock()，说明当前线程放弃对临界区的所有权，此时再调用notify_one会唤醒其他线程来对临界区执行操作。如果先唤醒其他线程，则可能unlock未执行完毕，就有线程到临界区了，此时新来的线程又会阻塞了。

代码示例

下面代码总共是3个例子。第一个Application是模拟一个事件处理系统的，但是没有使用条件变量，自己实现了一下，第二个Application使用了条件变量。第三个ProduceAndConsume是典型的生产者和消费者模型。

未使用条件变量，仅仅借助循环的加载数据应用

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <chrono>
#include <random>
#include <cstdlib>const int MAXT = 1000;
std::uniform_int_distribution<int>dis(1, MAXT);
std::random_device rd;
std::mt19937 gen(rd());class Application {public:void mainTask() {std::cout << "Do some main task...\n";auto t = dis(gen);  // 随机时间模拟主线任务std::this_thread::sleep_for(std::chrono::milliseconds(t));std::cout << "Finish main task in " << t << " ms\n";mtx.lock();while (!m_bDataLoaded) {mtx.unlock();std::this_thread::sleep_for(std::chrono::microseconds(100));mtx.lock();}mtx.unlock();std::cout << "Get loaded data\n";}void loadData() {std::cout << "Loading data...\n";auto t = dis(gen);  // 随机时间模拟主线任务std::this_thread::sleep_for(std::chrono::milliseconds(t));std::lock_guard<std::mutex>lck(mtx);m_bDataLoaded = true;std::cout << "Finish loading data in " << t << " ms\n";}bool isDataLoaded()const {return m_bDataLoaded;}private:bool m_bDataLoaded{ false };std::mutex mtx;std::condition_variable m_convVar;
};int main() {Application app;std::thread t1(&Application::mainTask, &app);std::thread t2(&Application::loadData, &app);t1.join();t2.join();system("pause");return 0;
}

结果：

使用了条件变量的加载数据应用

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <chrono>
#include <random>
#include <cstdlib>const int MAXT = 1000;
std::uniform_int_distribution<int>dis(1, MAXT);
std::random_device rd;
std::mt19937 gen(rd());class Application {public:void mainTask() {std::cout << "Do some main task...\n";auto t = dis(gen);  // 随机时间模拟主线任务std::this_thread::sleep_for(std::chrono::milliseconds(t));std::cout << "Finish main task in " << t << " ms\n";std::unique_lock<std::mutex> lck(mtx);m_convVar.wait(lck, std::bind(&Application::isDataLoaded, this));std::cout << "Get loaded data\n";}void loadData() {std::cout << "Loading data...\n";auto t = dis(gen);  // 随机时间模拟加载数据任务std::this_thread::sleep_for(std::chrono::milliseconds(t));std::cout << "Finish loading task in " << t << " ms\n";std::unique_lock<std::mutex> lck(mtx);m_bDataLoaded = true;lck.unlock();   // 最好是添加上这一句，本例子无所谓m_convVar.notify_one();}bool isDataLoaded()const {return m_bDataLoaded;}private:bool m_bDataLoaded{ false };std::mutex mtx;std::condition_variable m_convVar;
};int main() {Application app;std::thread t1(&Application::mainTask, &app);std::thread t2(&Application::loadData, &app);t1.join();t2.join();system("pause");return 0;
}

运行结果：

生产者和消费者模型

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <random>
#include <functional>
#include <algorithm>
#include <queue>
#include <vector>
#include <cstdlib>const int MAXT = 1000;
const int MAXN = 5;
std::random_device rd;
std::mt19937 gen(rd());
std::uniform_int_distribution<int>dis(MAXT / 10, MAXT);class ProducerAndConsumer {public:void prodece() {// 随机时间，模拟货物生产过程，生产过程本身不是在临界区std::this_thread::sleep_for(std::chrono::milliseconds(2 * dis(gen)));std::cout << "Prodece\n";std::unique_lock<std::mutex> lck(mtx);m_convPro.wait(lck, std::bind(&ProducerAndConsumer::notFull, this));m_qCargo.push(m_iCargoNum);++m_iCargoNum;m_convCon.notify_one();}void consume() {std::unique_lock<std::mutex> lck(mtx);m_convCon.wait(lck, std::bind(&ProducerAndConsumer::notEmpty, this));int n = m_qCargo.front();m_qCargo.pop();m_convPro.notify_one();lck.unlock();  // 一定要先解锁// 随机时间，模拟货物消费过程，消费过程本身不是在临界区！！！std::this_thread::sleep_for(std::chrono::milliseconds(dis(gen)));std::cout << "Consume\n";}inline bool notFull()const {return m_qCargo.size() < m_iMaxCargoNum;}inline bool notEmpty()const {return m_qCargo.size() > 0;}inline int getBufferSize()const {return m_qCargo.size();}inline int getCargoNum()const {return m_iCargoNum;}private:std::queue<int> m_qCargo;   // 货物队列int m_iMaxCargoNum{ MAXN }; // 最大容量int m_iCargoNum{ 0 };        // 货物总数std::mutex mtx;std::condition_variable m_convPro, m_convCon;
};int main() {ProducerAndConsumer pac;auto N = std::thread::hardware_concurrency();std::cout << "Thread num: " << N << std::endl;std::vector<std::thread>producerThreads;std::vector<std::thread>consumerThreads;for (int i = 0; i < N; ++i) {producerThreads.emplace_back(std::thread(&ProducerAndConsumer::prodece, &pac));consumerThreads.emplace_back(std::thread(&ProducerAndConsumer::consume, &pac));}std::for_each(producerThreads.begin(), producerThreads.end(),std::mem_fn(&std::thread::join));std::for_each(consumerThreads.begin(), consumerThreads.end(),std::mem_fn(&std::thread::join));std::cout << "Cargo in buffer: " << pac.getBufferSize() << std::endl;std::cout << "Cargo count: " << pac.getCargoNum() << std::endl;system("pause");return 0;
}

运行结果：