线程初始化，同步与期望
前言
之前讲了基本的线程和互斥量，我们还留下了几个问题，比如线程之间的通信和线程的返回值的获取。本文将深入探讨这些机制，并给出一些例子供参考。
正文
unique_lock
先来看这样一个例子：
int value{};
std::mutex mutex;
int* p{};
std::unique_lock<std::mutex> getlock() //1
{
std::unique_lock<std::mutex> lock(mutex);
++value;
return lock;
}
void runadd(int i) //2
{
std::unique_lock<std::mutex> lock(getlock());
int t = ++value;
lock.unlock();
printf("%d:%d\n" , i , t);
}
void run()
{
for (int i = 0 ; i < 10 ; ++i) {
std::thread t(runadd , i);
t.detach();
}
sleep(1);
}
这里我们使用了一个新的管理互斥量的类unique_lock，它的用法和std::lock_guard差不多，不过就像unique_ptr一样，unique_lock支持移动,所以它可以用来作为函数的返回值(1)，这是lock_guard做不到的。
另外，unique_lock也更加灵活，它允许你在不同的位置解锁并在其它位置重新上锁。这对于效率的提升很明显，比如在 io 操作时 CPU 基本是空闲的，如果此时该线程还持有锁，那么其他线程也运行不了，导致这一段时间被浪费了（有点占着茅坑不拉屎的感觉）。在 io 之前解锁，可以提高 cpu 的利用率。
不过unique_lock的灵活是有代价的，它需要一些额外的空间来帮助自己实现这种灵活性。具体使用std::unique_lock还是std::lock_guard要根据情况而定。
多线程初始化问题
考虑如果一个变量需要延迟初始化，你可能为了避免数据竞争会这样写：
int* p{};
void errorinit()
{
if (!p) {
std::lock_guard<std::mutex> lock(mutex);
if (!p) {
p = new int(5);
}
}
cout << *p << '\n';
}
这是声名狼藉的双重检查锁模式，因为其会造成条件竞争。我们注意到第一次对p的检查没有用互斥量保护，那么当另外一个线程在对p进行分配内存写入时，另外有一个线程有对 p 进行读取的风险，产生条件竞争。另外，当编译器对分配内存的三步（分配内存，写入数据，复制地址到 p ）进行优化重排时，比如先复制地址再写入数据，可能会导致在某一时刻 p 已经有了地址但其值未定义，此时另外一个线程读取 p 发现有值并执行下一步，会造成未定义行为，这是极其危险的数据竞争。
C++标准库提供了一种标志来解决这个问题。
std::once_flag flag;
void safeinit(int i)
{
std::call_once(flag , []() { p = new int(5); });
printf("%d:%d\n" , i , *p);
}
标志once_flag很类似mutex，但他是一次性的，所有的线程对call_once的可调用对象只会执行一次，他的参数是一个可调用对象。
当然，你可能会想到静态局部变量，只有第一次运行时会分配内存，也就是延迟初始化。这在多线程中也是成立的。所以你完全可以这样写：
void staticinit(int i)
{
static int sp{5};
printf("%d:%d\n" , i , sp);
}
这和上面的效果相同。
线程的同步
有时候我们的线程中会有一些依赖关系，线程 A 只有在线程 B 执行完毕后才能执行，但是两个线程怎么通知对方呢？进程间有共享内存和消息传递两种方式，线程也是相似的道理。首先我们想到用一个变量来传递消息：
bool start{};
void sleepforwhile()
{
std::unique_lock<std::mutex> lock(mutex);
while (!start) {
lock.unlock();
std::this_thread::sleep_for(std::chrono::milliseconds(100));
cout << "Sleep 100ms...\n";
lock.lock();
}
cout << "Sleep down\n";
}
void run(){
std::unique_lock<std::mutex> lock(mutex , std::defer_lock);
std::thread t1(sleepforwhile);
t1.detach();
sleep(1);
lock.lock();
start = true;
lock.unlock();
}
先创建一个锁备用，std::defer_lock表示延迟上锁。然后我们创建一个等待线程，这个线程检查 start 变量，不为真则睡眠一段时间再继续检查。
std::this_thread::sleep_for可以让当前线程睡眠一段时间
我们的主线程修改 start ，然后子线程可以执行他的下一步工作。这种方式很简单快捷，不过可拓展性太差，我们无法为不知道数量的线程都各自给一个变量，这样太难管理；无法做到即时通知，因为当我们让线程一直重复检查变量时其需要不断上锁解锁，这是非常消耗资源的；还有一些其他的缺陷。
我们可以使用条件变量std::condition_variable帮助我们实现同步。
std::condition_variable cv{};
std::queue<int> datas{};
std::mutex mutex;
void pushdata()
{
int i{};
while (start) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
std::lock_guard<std::mutex> lock(mutex);
datas.push(i++);
cv.notify_one();
}
cout << "end pushdata\n";
}
void coutdata()
{
int i{};
while (start) {
std::unique_lock<std::mutex> lock(mutex);
cv.wait(lock , []() { return !datas.empty(); });
int t = datas.front();
datas.pop();
lock.unlock();
printf("%d:%d\n" , i++ , t);;
}
cout << "end coutdata\n";
}
注意这里多个线程对start都是读取操作，没有写入操作，不会引起数据竞争
我们定义一个条件变量cv，当线程 1 向队列中传入数据后，他会调用cv.notify_one来通知其他的一个阻塞在cv.wait上的线程，然后这个线程可以检查条件并判断是否要进行下一步操作。
对于cv.wait()，我们必须给他传入一个unique_lock锁，这个锁可以是局部变量，但是锁的互斥量必须要能够锁住我们用到的消息队列。然后wait解锁并阻塞，等待一次唤醒。当锁被唤醒时，其上锁并检查后面的调用对象返回的条件，如果条件为假再次解锁并阻塞，为真才退出阻塞并运行。
这里必须保证进行同步的线程使用的条件变量和互斥量都是同一个，才能正确的执行同步操作。
期望
到现在为止，我们获取返回值的方式都是通过函数的输出参数，但是有时候我们想运行的函数通过返回值返回结果，我们该怎么获取这个返回值呢？
我们把想要线程返回的值叫做期望std::future，即我们期望这个线程返回一个答案。期望是一个模板类，持有一个任务状态的引用。调用期望的成员函数get()，当任务状态为就绪时，我们可以获取想要的答案，或者一个异常。
期望
std::future不可复制，但支持移动，可复制的期望是std::shared_future。
async异步与同步
std::async提供了一个对线程的包装实现，我们可以像创建普通线程一样调用async：
constexpr int num{77};
class Num
{
int i{77};
public:
[[nodiscard]] int geti() const
{
return i;
}
};
int geti()
{
cout << "geti ";
return num;
}
void run()
{
std::future<int> f1 = std::async(std::launch::async,geti);
auto f2 = std::async(&Num::geti , Num());
sleep(1);
cout << "f1: " << f1.get() << "\nf2: " << f2.get() << '\n';
auto f3 = std::async(std::launch::deferred , add , 1 , 2);
sleep(1);
cout << "f3: " << f3.get() << '\n';
}
geti f1: 77
f2: 77
f3: run add 3
async创建一个共享的状态，这个状态包含你想要执行的任务函数，任务的结果，启动任务的策略，等等东西，并返回一个包含该状态的引用的期望。
任务的策略有两种，async让任务以新线程的方式立刻执行，而deferred让任务一直等到期望调用get()或wait()时，使用本线程而不是新建线程来执行任务。这些从输出都可以看出来。当没有指定策略时，它会根据实现选择一个合适的策略来运行。
任务
async是一个高抽象的函数模板，它很方便，但不太灵活。当我们有大量的任务时，使用async可能会创建过多线程直到耗尽资源，而且我们也无法看到async实现的细节。标准库提供了对任务的包装std::packaged_task，让我们可以自己管理任务，并分配线程。
void run(){
std::packaged_task<int(int , int)> addtask(add);
auto f4 = addtask.get_future();
addtask(1 , 3);
cout << f4.get() << '\n';
}
这看起来很像普通的函数调用，不过通过任务包装，我们把函数的执行和返回的结果解耦，使其可以在两个不同的阶段进行。通过get_future函数获取任务的期望，然后我们可以把这个任务交给其他线程来执行，当我们需要结果的时候只需要对期望调用get()函数即可。
注意，任务只是一个可调用对象，并没有创建新的线程。
承诺
std::packaged_task已经非常强大了，但它还是属于高抽象的封装。当我们不满足于只期望任务的返回值，而是期望任务运行过程中的一些状态，比如哪一步是否发生了异常或者计算的结果对不对，我们就需要自己定制这些期望。我们使用承诺std::promise。
std::future<int> addpromisefuture()
{
auto p = std::make_shared<std::promise<int>>(); //1
auto f = p->get_future();
std::thread t([p]
{
p->set_value(5);
});
t.detach();
return f;
}
void waitfutures()
{
std::promise<int> p;
std::shared_future<int> f = p.get_future();
std::thread t([](std::promise<int> promise)
{
sleep(1);
promise.set_value(6);
},std::move(p));
t.detach();
for (int i = 0 ; i < 10 ; ++i) {
std::thread tem([f,i]
{
printf("%d:%d\n" , i , f.get());
});
tem.detach();
}
sleep(1);
}
void run(){
cout << addpromisefuture().get() << '\n';
waitfutures();
}
就像它的名字一样，std::promise承诺会给我们一个结果，保证我们的期望不落空。这是我们需要保证的事情，如果承诺没有得到实现，那期望会一直等，而如果承诺直到被析构都没有给我们结果，get()就会抛出错误。
1我们用一个共享智能指针来创建一个promise对象，然后调用get_future获取其对应的期望，然后我们创建一个线程执行我们的任务，任务执行过程中我们的承诺通过set_value回应了期望，所以我们可以获得期望的值。承诺也可以使用set_exception回应一个异常。
这里使用共享指针的目的是在当前函数执行完毕时延长
promise的生命周期，并允许我们复制指针作为lambda的成员。当然这是一个危险的做法，因为promise本身不可复制，使用指针允许复制后可能会导致多次写入。最好的办法是像下面的函数一样将承诺移动到我们的任务中去。
2另外一边，我们用shared_future接收承诺的期望，并创建了 10 个线程等待这个承诺响应。最终，这 10 个子线程几乎是同时输出其结果。
限制等待时间
有时候我们不想让一个线程等待另外一个线程太久，毕竟如果对方一直不回应我们也没有必要一直等下去。有很多类都提供了等待函数，分为wait_for和wait_until，一个是等待多长时间，一个是等到某个时间点。
void sleeptask()
{
sleep(1);
}
void run()
{
std::packaged_task<void()> task(sleeptask);
auto f1 = task.get_future();
std::thread t(std::move(task));
t.detach();
if (f1.wait_for(std::chrono::milliseconds(100)) == std::future_status::ready) {
cout << "won!\n";
}
else {
cout << "not won!\n";
}
}
这里是future的等待示例，wait方法在等待期间期望状态改变或是等待完成后会返回一个状态，然后我们可以进行一个判断。
bool done{};
std::mutex mutex;
std::condition_variable cv;
void setcv()
{
sleep(1);
std::lock_guard<std::mutex> lock(mutex);
done = true;
cv.notify_all();
}
bool waitloop()
{
std::unique_lock<std::mutex> lock(mutex);
while (!done) {
if (cv.wait_for(lock , std::chrono::seconds(1)) == std::cv_status::timeout) {
break;
}
}
return done;
}
void run(){
std::packaged_task<bool()> task2(waitloop);
auto f2 = task2.get_future();
std::thread t2(std::move(task2));
t2.detach();
setcv();
if (f2.get()) {
cout << "won!\n";
}
else {
cout << "not won!\n";
}
}
我们的条件变量也有wait方法，当我们有一个线程完成任务后，它想通知所有的其他线程，所以它调用notify_all，然后其他的线程从wait中唤醒，并只有一个线程抢到了锁，其他进程阻塞在锁的获取队列上。如果这一批次线程没有完成任务，那么他们就继续等待下一次唤醒。直到超时或任务完成，才退出循环。这种循环结构的好处是，线程被假唤醒后可以重新进入循环等待。不过使用wait_for有无限等待的可能，这种情况下，使用wait_until是更好的选择：
bool waitloop2()
{
std::unique_lock<std::mutex> lock(mutex);
while (!done) {
if (cv.wait_until(lock , std::chrono::steady_clock::now() + std::chrono::seconds(1)) ==
std::cv_status::timeout) {
break;
}
}
return done;
}
这样可以保证最大等待时间，不会一直因为假唤醒而重置等待时间。
总结
本文算是给上一篇文章填坑。首先是 std::unique_lock，它比 std::lock_guard 灵活：支持移动、可以中途解锁再重新上锁，代价是多一点点的空间和间接开销，两者按需选择即可。
然后是延迟初始化问题。双重检查锁在 C++ 里声名狼藉，无保护的首次检查和编译器的指令重排都可能引发竞争；标准库给出的方案是 std::call_once 配合 std::once_flag，而更省心的办法是直接利用 C++11 对静态局部变量初始化的线程安全保证。
线程间的同步从"轮询一个共享变量"这种朴素做法讲起，指出了它无法即时通知、反复加解锁开销大的缺陷，进而引出条件变量 std::condition_variable：配合 unique_lock 与谓词，让线程在条件不满足时真正地睡眠，被唤醒后再重新检查条件，这才是高效又正确的等待方式。
最后是重头戏——期望 std::future。std::async 是最省事的封装，但线程数量不可控；std::packaged_task 把任务的执行和结果的获取解耦，让我们自己安排线程；std::promise 则把主动权完全交给使用者，可以定制结果、传递异常，配合 std::shared_future 还能让多个线程等待同一个结果。此外还介绍了用 wait_for 等待一段时长、用 wait_until 等到某个时间点，避免线程陷入无限等待。
到这里，线程从创建、管理、同步到取回结果的主线就算打通了。本人才疏学浅，文章中有不妥之处还请指正。
$ 正在加载评论…