---
title: "线程初始化，同步与期望"
date: 2026-10-14
category: "操作系统"
tags: ["C++", "并发", "内存模型", "atomic"]
summary: "前言\n\n之前讲了基本的线程和互斥量，我们还留下了几个问题，比如线程之间的通信和线程的返回值的获取。本文将深入探讨这些机制，并给出一些例子供参考。\n\n 正文\n\n unique_lock\n\n先来看这样一个例子：\n\ncpp\nint value{}…"
series:
  name: "操作系统并发"
  order: 0
---

## 前言

之前讲了基本的线程和互斥量，我们还留下了几个问题，比如线程之间的通信和线程的返回值的获取。本文将深入探讨这些机制，并给出一些例子供参考。

## 正文

### unique_lock

先来看这样一个例子：

```cpp
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`要根据情况而定。

### 多线程初始化问题

考虑如果一个变量需要延迟初始化，你可能为了避免数据竞争会这样写：

```cpp
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++标准库提供了一种标志来解决这个问题。

```cpp
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`的可调用对象只会执行一次，他的参数是一个可调用对象。

当然，你可能会想到静态局部变量，只有第一次运行时会分配内存，也就是延迟初始化。这在多线程中也是成立的。所以你完全可以这样写：

```cpp
void staticinit(int i)
{
    static int sp{5};
    printf("%d:%d\n" , i , sp);
}
```

这和上面的效果相同。

### 线程的同步

有时候我们的线程中会有一些依赖关系，线程 A 只有在线程 B 执行完毕后才能执行，但是两个线程怎么通知对方呢？进程间有共享内存和消息传递两种方式，线程也是相似的道理。首先我们想到用一个变量来传递消息：

```cpp
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`帮助我们实现同步。

```cpp
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`：

```cpp
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';
}
```

```bash
geti f1: 77
f2: 77
f3: run add 3
```

`async`创建一个共享的状态，这个状态包含你想要执行的任务函数，任务的结果，启动任务的策略，等等东西，并返回一个包含该状态的引用的期望。

任务的策略有两种，`async`让任务以新线程的方式立刻执行，而`deferred`让任务一直等到期望调用`get()`或`wait()`时，使用本线程而不是新建线程来执行任务。这些从输出都可以看出来。当没有指定策略时，它会根据实现选择一个合适的策略来运行。

#### 任务

`async`是一个高抽象的函数模板，它很方便，但不太灵活。当我们有大量的任务时，使用`async`可能会创建过多线程直到耗尽资源，而且我们也无法看到`async`实现的细节。标准库提供了对任务的包装`std::packaged_task`，让我们可以自己管理任务，并分配线程。

```cpp
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`。

```cpp
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`，一个是等待多长时间，一个是等到某个时间点。
```cpp
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`方法在等待期间期望状态改变或是等待完成后会返回一个状态，然后我们可以进行一个判断。
```cpp
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`是更好的选择：
```cpp
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` 等到某个时间点，避免线程陷入无限等待。

到这里，线程从创建、管理、同步到取回结果的主线就算打通了。本人才疏学浅，文章中有不妥之处还请指正。