C++ 从入门到精通之十二

发表于 阅读次数： 评论数： 本文字数： 5.3k 阅读时长 ≈ 5 分钟

大纲

C++ 并发编程

扩展阅读 - [C++ 多线程编程之一](/posts/a2a7ad9b.html)

条件变量

条件变量的概述

std::condition_variable 是 C++ 标准库中用于线程间同步的一个核心工具，它允许线程之间通过事件通知机制进行协作。

  • 核心作用：

    • (1) 阻塞线程：让一个或多个线程在特定条件不满足时进入等待（休眠）状态，从而避免无意义的忙等（Busy-Waiting），释放 CPU 资源。
    • (2) 唤醒线程：允许另一个线程在修改共享数据后，通知（唤醒）那些正在等待的线程，让它们重新检查条件并继续执行。

  • 核心函数：

    • 等待函数（阻塞）
      • wait(unique_lock<mutex>& lock)：原子性地解锁传入的互斥量并使当前线程进入阻塞状态，直到被 notify 唤醒。被唤醒后，它会重新获取互斥量（加锁）并继续执行（通常需要配合循环检查条件）。
      • wait(unique_lock<mutex>& lock, Predicate pred)：带谓词（条件检查）的重载。等效于 while (!pred()) wait(lock)，它会自动处理虚假唤醒，只有在条件满足时才真正返回。
    • 通知函数（唤醒）
      • notify_one()：唤醒正在等待队列中的任意一个线程（如果有的话）。如果多个线程在等待，系统调度选择一个唤醒。
      • notify_all()：唤醒所有正在等待的线程。这些线程会竞争互斥量，并依次检查各自的条件。
    • 带超时的等待函数
      • wait_for() 和 wait_until()：允许线程等待一段特定时间或等待到某个绝对时间点。如果超时后仍未收到通知，线程会自动醒来继续执行，避免永久阻塞。

特别注意 - `std::condition_variable` 必须与 `std::unique_lock`配合使用（不能使用`std::lock_guard`替代），因为等待期间需要原子性地释放锁。-`std::condition_variable`必须使用`while` 循环或谓词检查，因为存在虚假唤醒（Spurious Wakeup），即线程可能在未被通知的情况下醒来，必须重新验证条件是否真正满足。

条件变量的使用

!!! note 案例背景说明
在网络游戏服务器的设计中，共享数据的保护是一个典型案例：可以创建两个线程，其中一个线程负责收集玩家的命令并将命令数据写入队列，另一个线程则从队列中取出玩家发来的命令，进行解析并执行玩家所需的动作。值得一提的是，在当前业务场景下建议使用生产者消费者模型来实现，并使用 list 容器作为队列。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
#include <atomic>
#include <condition_variable>
#include <iostream>
#include <list>
#include <mutex>
#include <thread>

using namespace std;

class MyClass {
public:
    // 将收到的玩家命令写入队列
    void inMsgRecvQueue() {
        for (int i = 0; i < 1000; ++i) {
            // 加锁
            std::unique_lock<std::mutex> lock(msgRecvQueueMutex);

            // 插入队列
            msgRecvQueue.push_back(i);

            // 解锁
            lock.unlock();

            // 唤醒一个正在等待的线程
            condtion.notify_one();

            // 模拟网络收包间隔
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }

        // 加锁
        std::unique_lock<std::mutex> lock(msgRecvQueueMutex);

        // 更新程序停止标记
        stop = true;

        // 解锁
        lock.unlock();

        // 唤醒所有等待线程，让其检测 stop 标记并退出
        condtion.notify_all();
    }

    // 从队列中读取玩家命令
    void outMsgRecvQueue() {
        while (true) {
            // 每轮循环重新初始化
            int command = -1;

            // 加锁
            std::unique_lock<std::mutex> lock(msgRecvQueueMutex);

            // 等待条件满足
            // 如果 Lambda 表达式返回 true，当前线程会继续往下执行（持有锁）
            // 如果 Lambda 表达式返回 false，那么 wait() 将解锁互斥量（释放锁），并让当前线程阻塞等待直到被唤醒
            // wait(lock, predicate) 内部会循环判断 predicate（断言），可以有效避免虚假唤醒
            condtion.wait(lock, [this]() { return stop || !msgRecvQueue.empty(); });

            // 如果程序停止，并且消息队列已经处理完，则退出线程
            if (stop && msgRecvQueue.empty()) {
                break;
            }

            // 操作队列
            if (!msgRecvQueue.empty()) {
                // 取出队列元素
                command = msgRecvQueue.front();

                // 移除队列元素
                msgRecvQueue.pop_front();
            }

            // 解锁
            lock.unlock();

            // 打印
            if (command != -1) {
                std::cout << "已处理玩家命令: " << command << std::endl;
            }

            // 模拟业务执行耗时
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
    }

private:
    bool stop = false;                 // 程序停止标记
    std::list<int> msgRecvQueue;       // 消息队列（共享数据）
    std::mutex msgRecvQueueMutex;      // 保护消息队列线程安全的互斥锁
    std::condition_variable condtion;  // 条件变量
};

int main() {
    std::cout << "main thread start." << std::endl;

    // 局部变量
    MyClass mc;

    // 创建并启动写线程
    std::thread t_write(&MyClass::inMsgRecvQueue, &mc);

    // 创建并启动读线程
    std::thread t_read(&MyClass::outMsgRecvQueue, &mc);

    // 等待写线程执行完毕
    t_write.join();

    // 等待读线程执行完毕
    t_read.join();

    std::cout << "main thread end." << std::endl;

    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
7
已处理玩家命令: 0
已处理玩家命令: 1
已处理玩家命令: 2
......
已处理玩家命令: 997
已处理玩家命令: 998
已处理玩家命令: 999

std::async

std::async 的概述

  • 概述：

    • std::async 是 C++ 11 引入的异步任务启动机制，用于将函数异步执行，并返回一个 std::future 对象，以便获取任务执行结果或等待任务完成。

  • 特点：

    • 简化多线程编程：无需手动创建和管理 std::thread。
    • 线程同步调用：调用 get() 时，如果任务尚未完成，会阻塞当前线程等待执行结果。
    • 异步结果获取：支持通过 std::future 从异步任务中获取返回结果。
    • 一次性读取结果：std::future 的结果只能通过 get() 获取一次，获取后对象将失效。
    • 支持异常传播：任务中抛出的异常会保存在 std::future 中，在调用 get() 时重新抛出异常。
    • 自动资源管理：任务结束后线程资源自动回收，无需 join() 或 detach()。
    • 支持状态查询：可以通过 wait_for()、wait_until() 判断任务是否完成。

  • 启动策略：

    • std::async 支持三种启动方式：
      启动策略说明
      std::launch::async立即创建新线程执行任务。
      std::launch::deferred延迟执行，直到调用 get() 或 wait() 时才在当前线程（不会创建子线程）执行。
      默认策略由实现决定采用立即执行还是延迟执行。

  • 常用成员函数：

    • get()：获取任务返回值（只能调用一次）。
    • wait()：等待任务完成。
    • wait_for()：等待指定时间。
    • wait_until()：等待直到指定时间点。

  • 优点：

    • 使用简单，代码量少。
    • 自动管理线程生命周期。
    • 支持返回值和异常传递。
    • 适合执行独立的异步计算任务。

  • 缺点：

    • 默认启动策略具有不确定性，可能不会真正创建线程。
    • 无法直接控制线程属性（如线程名称、优先级等）。
    • 大量创建异步任务可能带来线程创建开销，不适合高并发场景。

  • 适用场景：

    • 后台计算任务。
    • 并行执行多个独立任务。
    • 需要获取任务返回值的异步操作。
    • 临时性的异步任务，而非长期运行的线程。

!!! note 总结
std::async 是 C++ 11 提供的高级异步任务接口，它可以自动创建并管理线程，返回 std::future 用于获取执行结果，是比 std::thread 更易用的异步编程方式。

std::async 的使用

使用案例一
  • std::async + 普通函数的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

int process(int milliseconds) {
    std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;
    // 模拟业务处理耗时
    const std::chrono::milliseconds ms(milliseconds);
    std::this_thread::sleep_for(ms);
    std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;
    // 返回业务处理结果
    return 5;
}

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;
    // 启动一个子线程，第二个参数是线程函数的参数
    std::future<int> result = std::async(process, 5000);
    std::cout << "continue ..." << std::endl;
    // 获取子线程的执行结果（阻塞等待线程执行完成）
    const int num = result.get();
    std::cout << "result = " << num << std::endl;
    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
main() run, thread id 1
continue ...
process() start, current thread id 2
process() end, current thread id 2
result = 5
main() end, thread id 1

特别注意

在上面的 std::async 案例代码中，即使不手动调用 std::future 的 get() 函数，main 线程也会等待子线程执行完成。

使用案例二
  • std::async + 类成员函数的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

class MyClass {
public:
    int process(const int milliseconds) {
        std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;
        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);
        std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;
        // 返回业务处理结果
        return 5;
    }
};

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;
    // 启动一个子线程
    MyClass mc;
    // 第二个参数是对象引用，第三个参数是线程函数的参数
    std::future<int> result = std::async(&MyClass::process, &mc, 5000);
    std::cout << "continue ..." << std::endl;
    // 获取子线程的执行结果（阻塞等待线程执行完成）
    const int num = result.get();
    std::cout << "result = " << num << std::endl;
    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
main() run, thread id 1
continue ...
process() start, current thread id 2
process() end, current thread id 2
result = 5
main() end, thread id 1

特别注意

在上面的 std::async 案例代码中，即使不手动调用 std::future 的 get() 函数，main 线程也会等待子线程执行完成。

使用案例三
  • std::async + std::launch::deferred 的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

class MyClass {
public:
    int process(const int milliseconds) {
        std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;
        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);
        std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;
        // 返回业务处理结果
        return 5;
    }
};

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;
    MyClass mc;
    // 第一个参数是执行策略，第三个参数是对象引用，第四个参数是线程函数的参数
    std::future<int> result = std::async(std::launch::deferred, &MyClass::process, &mc, 5000);
    std::cout << "continue ..." << std::endl;
    // 使用 std::launch::deferred 执行策略后，不会创建子线程，当 get() 被调用后才会在当前主线程（非子线程）开始执行任务
    const int num = result.get();
    std::cout << "result = " << num << std::endl;
    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
main() run, thread id 1
continue ...
process() start, current thread id 1
process() end, current thread id 1
result = 5
main() end, thread id 1
使用案例四
  • std::async + std::launch::async 的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

class MyClass {
public:
    int process(const int milliseconds) {
        std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;
        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);
        std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;
        // 返回业务处理结果
        return 5;
    }
};

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;
    MyClass mc;
    // 第一个参数是执行策略，第三个参数是对象引用，第四个参数是线程函数的参数
    std::future<int> result = std::async(std::launch::async, &MyClass::process, &mc, 5000);
    std::cout << "continue ..." << std::endl;
    // 使用 std::launch::async 执行策略后，会立即创建子线程执行任务，主线程调用 get() 后会阻塞等待任务执行完成并获取执行结果
    const int num = result.get();
    std::cout << "result = " << num << std::endl;
    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
main() run, thread id 1
continue ...
process() start, current thread id 2
process() end, current thread id 2
result = 5
main() end, thread id 1

std::packaged_task

std::packaged_task 的概述

  • 概述：

    • std::packaged_task 是 C++ 11 引入的异步任务封装机制，用于将一个可调用对象（普通函数、函数对象、Lambda 表达式等）封装成任务，并将任务执行结果与 std::future 关联起来，以便后续获取返回值或异常信息。

  • 特点：

    • 任务封装管理：可以将任意可调用对象包装成一个可执行任务，统一管理任务执行过程。
    • 结果异步获取：任务执行完成后的返回值会保存到共享状态中，并通过关联的 std::future 获取。
    • 执行与获取分离：任务的创建、执行和结果获取可以独立进行，提高任务调度灵活性。
    • 支持异常传播：任务执行过程中抛出的异常会被保存起来，并在调用 future::get() 时重新抛出。
    • 一次执行机制：一个 std::packaged_task 对象通常只能执行一次，执行完成后任务状态失效。
    • 支持移动语义：std::packaged_task 不支持拷贝，只支持移动，适合在线程池、任务队列中传递。
    • 无需手动同步结果：内部自动管理任务结果与 std::future 之间的关联。

  • 常用成员函数：

    • get_future()：获取与任务结果关联的 std::future 对象，通常应该在任务执行前调用。
    • operator()：执行封装的任务。
    • valid()：判断当前 std::packaged_task 是否有效。
    • reset()：重新绑定共享状态，使任务可以再次执行。

  • 与其他异步组件的关系：

    组件作用
    std::future获取异步任务执行结果。
    std::async自动启动异步任务，并返回 std::future。
    std::packaged_task封装任务，将任务执行和结果获取分离。
    std::promise手动设置异步结果，通常由一个线程生产结果，另一个线程通过 std::future 获取。

  • 优点：

    • 任务封装和线程管理分离，灵活性更高。
    • 支持在线程池、任务队列中实现任务调度。
    • 自动处理返回值和异常传递。
    • 相比 std::async，可以更精细地控制任务执行方式。

  • 缺点：

    • 使用复杂度高于 std::async。
    • 需要手动安排任务执行线程，不会自动创建线程。
    • 一个任务对象只能执行一次，需要重新执行时必须调用 reset()。

  • 适用场景：

    • 线程池任务封装。
    • 任务队列异步执行。
    • 需要将任务提交和任务执行分离的场景。
    • 多线程环境下需要统一管理异步任务结果的场景。

总结

std::packaged_task 是 C++ 11 提供的任务封装工具，它将可调用对象、任务执行和结果获取进行解耦，通过 std::future 获取执行结果。相比 std::async，它提供了更灵活的任务调度能力，常用于线程池等高级异步编程场景。

特别注意

std::packaged_task 的 get_future() 通常应在任务执行前调用，以获取与任务结果关联的 std::future 对象。虽然部分编译器实现（例如 Windows 下的 MSVC）允许在任务执行完成后再调用 get_future()，代码也可能正常运行，但这并不是推荐的使用方式。为了保证代码的跨平台兼容性，并符合异步任务的设计思想，应在执行 packaged_task 前完成 std::future 获取，然后通过 future::get() 获取任务执行结果。

std::packaged_task 的使用

使用案例一
  • std::packaged_task + 普通函数的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

int process(const int milliseconds) {
    std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;

    // 模拟业务处理耗时
    const std::chrono::milliseconds ms(milliseconds);
    std::this_thread::sleep_for(ms);

    std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;

    // 返回业务处理结果
    return 5;
}

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建 packaged_task，参数是可调用对象（比如普通函数）
    std::packaged_task<int(int)> task(process);

    // 获取子线程的执行结果，特别注意：get_future() 必须在 packaged_task 执行前调用
    std::future<int> result = task.get_future();

    // 创建子线程（第二个参数是线程函数的参数），子线程会直接执行
    std::thread t1(std::ref(task), 5000);

    // 等待子线程执行完成
    t1.join();

    // 获取执行结果
    const int num = result.get();

    std::cout << "result = " << num << std::endl;

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;

    return 0;
}

程序运行的结果如下：

1
2
3
4
5
main() run, thread id 1
process() start, current thread id 2
process() end, current thread id 2
result = 5
main() end, thread id 1
使用案例二
  • std::packaged_task + Lambda 表达式的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建 packaged_task，参数是可调用对象（比如 Lambda 表达式）
    std::packaged_task<int(int)> task([](const int milliseconds) -> int {
        std::cout << "process start, current thread id " << std::this_thread::get_id() << std::endl;

        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);

        std::cout << "process end, current thread id " << std::this_thread::get_id() << std::endl;

        // 返回业务处理结果
        return 5;
    });

    // 获取子线程的执行结果，特别注意：get_future() 必须在 packaged_task 执行前调用
    std::future<int> result = task.get_future();

    // 创建子线程（第二个参数是线程函数的参数），子线程会直接执行
    std::thread t1(std::ref(task), 5000);

    // 等待子线程执行完成
    t1.join();

    // 获取执行结果
    const int num = result.get();

    std::cout << "result = " << num << std::endl;

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;

    return 0;
}

程序运行的结果如下：

1
2
3
4
5
main() run, thread id 1
process start, current thread id 2
process end, current thread id 2
result = 5
main() end, thread id 1
使用案例三
  • std::packaged_task 作为可调用对象直接调用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建 packaged_task，参数是可调用对象（比如 Lambda 表达式）
    std::packaged_task<int(int)> task([](const int milliseconds) -> int {
        std::cout << "process start, current thread id " << std::this_thread::get_id() << std::endl;

        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);

        std::cout << "process end, current thread id " << std::this_thread::get_id() << std::endl;

        // 返回业务处理结果
        return 5;
    });

    // 获取执行结果，特别注意：get_future() 必须在 packaged_task 执行前调用
    std::future<int> result = task.get_future();

    // 直接调用 packaged_task，在当前线程执行（相当于普通函数调用），不会创建子线程，参数是可调用对象的参数
    task(5000);

    // 获取任务执行结果
    const int num = result.get();

    std::cout << "result = " << num << std::endl;

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;

    return 0;
}

程序运行的结果如下：

1
2
3
4
5
main() run, thread id 1
process start, current thread id 1
process end, current thread id 1
result = 5
main() end, thread id 1
使用案例四
  • std::packaged_task + std::vector 容器的使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
#include <chrono>
#include <future>
#include <iostream>
#include <thread>
#include <vector>

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建 packaged_task，参数是可调用对象（比如 Lambda 表达式）
    std::packaged_task<int(int)> task([](const int milliseconds) -> int {
        std::cout << "process start, current thread id " << std::this_thread::get_id() << std::endl;

        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);

        std::cout << "process end, current thread id " << std::this_thread::get_id() << std::endl;

        // 返回业务处理结果
        return 5;
    });

    // 创建容器
    std::vector<std::packaged_task<int(int)>> tasks;

    // 将 packaged_task 放入容器
    tasks.push_back(std::move(task));

    // 从容器获取 packaged_task
    const auto iter = tasks.begin();
    std::packaged_task<int(int)> task2 = std::move(*iter);

    // 从容器移除 packaged_task，后续不能再使用 iter，否则会出现未定义行为
    tasks.erase(iter);

    // 获取执行结果，特别注意：get_future() 必须在 packaged_task 执行前调用
    std::future<int> result = task2.get_future();

    // 直接调用 packaged_task，在当前线程执行（相当于普通函数调用），不会创建子线程，参数是可调用对象的参数
    task2(5000);

    // 获取任务执行结果
    const int num = result.get();
    std::cout << "result = " << num << std::endl;

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
main() run, thread id 1
process start, current thread id 1
process end, current thread id 1
result = 5
main() end, thread id 1
使用案例五
  • std::packaged_task + std::queue 实现任务队列（常用于线程池）
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
#include <chrono>
#include <future>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建任务队列
    std::queue<std::packaged_task<int(int)>> tasks;

    // 创建互斥锁，保护任务队列
    std::mutex mutex;

    // 创建 packaged_task，参数是可调用对象（比如 Lambda 表达式）
    std::packaged_task<int(int)> task([](const int milliseconds) -> int {
        std::cout << "process start, current thread id " << std::this_thread::get_id() << std::endl;

        // 模拟业务处理耗时
        const std::chrono::milliseconds ms(milliseconds);
        std::this_thread::sleep_for(ms);

        std::cout << "process end, current thread id " << std::this_thread::get_id() << std::endl;

        // 返回业务处理结果
        return 5;
    });

    // 获取 future，必须在 packaged_task 执行前调用
    std::future<int> result = task.get_future();

    {
        // 加锁，将任务放入任务队列
        std::lock_guard<std::mutex> lock(mutex);

        // 将 packaged_task 放入任务队列
        tasks.push(std::move(task));
    }

    // 创建工作线程，模拟任务消费者
    std::thread worker([&tasks, &mutex]() {
        std::cout << "worker thread start, thread id " << std::this_thread::get_id() << std::endl;

        std::packaged_task<int(int)> current_task;

        {
            // 加锁，从任务队列获取任务
            std::lock_guard<std::mutex> lock(mutex);

            // 判断任务队列是否为空
            if (!tasks.empty()) {
                // 从任务队列获取 packaged_task
                current_task = std::move(tasks.front());

                // 从队列移除 packaged_task
                tasks.pop();
            }
        }

        // 判断是否成功获取任务
        if (current_task.valid()) {
            // 执行 packaged_task，在当前线程执行，不会创建新的线程，参数是可调用对象的参数
            current_task(5000);
        }

        std::cout << "worker thread end, thread id " << std::this_thread::get_id() << std::endl;
    });

    // 等待工作线程执行完成
    worker.join();

    // 获取执行结果
    const int num = result.get();
    std::cout << "result = " << num << std::endl;

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;

    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
7
main() run, thread id 1
worker thread start, thread id 2
process start, current thread id 2
process end, current thread id 2
worker thread end, thread id 2
result = 5
main() end, thread id 1

std::promise

std::promise 的概述

  • 概述：

    • std::promise 是 C++ 11 引入的异步结果传递机制，用于在线程之间传递数据或异常。它与 std::future 配合使用，一个线程通过 std::promise 设置结果，另一个线程通过关联的 std::future 获取结果。

  • 特点：

    • 线程间数据传递：支持一个线程向另一个线程传递异步执行结果。
    • 主动设置结果：与 std::async 和 std::packaged_task 自动保存结果不同，std::promise 需要开发者手动调用接口设置结果。
    • 支持异常传递：可以在线程中保存异常信息，并通过 future::get() 在另一线程重新抛出异常。
    • 一次性赋值：一个 std::promise 对象只能设置一次结果，重复设置会产生异常。
    • 与 std::future 配合使用：通过 get_future() 获取关联的 std::future 对象，实现结果同步。
    • 支持移动语义：std::promise 不支持拷贝，只支持移动，适合在线程间转移所有权。

  • 常用成员函数：

    • get_future()：获取与当前 promise 关联的 std::future 对象。
    • set_value()：设置异步任务的返回结果。
    • set_exception()：设置异常结果，并传递给等待获取结果的线程。
    • swap()：交换两个 std::promise 对象的状态。

  • 与其他异步组件的关系：

    组件作用
    std::future获取异步操作的结果。
    std::promise手动设置异步操作的结果或异常。
    std::packaged_task封装任务，并自动将执行结果保存到 std::future。
    std::async自动启动异步任务，并返回 std::future。

  • 优点：

    • 提供灵活的线程间通信方式。
    • 可以精确控制结果产生的时机。
    • 支持结果和异常统一传递。
    • 适合实现自定义异步任务模型。

  • 缺点：

    • 需要手动管理任务执行和结果设置。
    • 使用复杂度高于 std::async。
    • 一个 promise 只能关联一个 future，不适合一对多结果通知场景。

  • 适用场景：

    • 多线程之间传递计算结果。
    • 线程任务完成通知。
    • 自定义异步任务框架。
    • 线程池中任务执行结果返回。
    • 需要手动控制异步结果产生时机的场景。

总结

std::promise 是 C++ 11 提供的底层异步通信工具，它负责生产异步结果，而 std::future 负责获取异步结果。相比 std::async 和 std::packaged_task，std::promise 提供了更灵活的结果控制能力，常用于线程间通信和自定义异步任务实现。

std::promise 的使用

使用案例一
  • std::promise + std::future 的简单使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

void process(std::promise<int>& result, int milliseconds) {
    std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;

    // 模拟业务处理耗时
    const std::chrono::milliseconds ms(milliseconds);
    std::this_thread::sleep_for(ms);

    std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;

    // 设置业务处理结果
    result.set_value(5);
}

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建 Promise 对象
    std::promise<int> mpro;

    // 获取子线程的执行结果，特别注意：get_future() 应该在子线程执行前调用
    std::future<int> result = mpro.get_future();

    // 创建子线程，第一个参数是线程函数，第二个参数是 Promise 对象的引用，第三个参数毫秒数
    std::thread t1(process, std::ref(mpro), 5000);

    // 阻塞等待子线程执行完成
    t1.join();

    // 获取子线程的执行结果
    const int num = result.get();
    std::cout << "result = " << num << std::endl;

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
main() run, thread id 1
process() start, current thread id 2
process() end, current thread id 2
result = 5
main() end, thread id 1
使用案例二
  • std::promise + std::future 在不同线程间通信的典型使用方式
    • process() 线程负责生产结果，通过 promise.set_value() 设置数据结果；
    • process2() 线程负责消费结果，通过 future.get() 等待并获取数据结果。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
#include <chrono>
#include <future>
#include <iostream>
#include <thread>

void process(std::promise<int>& result, int milliseconds) {
    std::cout << "process() start, current thread id " << std::this_thread::get_id() << std::endl;

    // 模拟业务处理耗时
    const std::chrono::milliseconds ms(milliseconds);
    std::this_thread::sleep_for(ms);

    std::cout << "process() end, current thread id " << std::this_thread::get_id() << std::endl;

    // 设置业务处理结果
    result.set_value(5);
}

void process2(std::future<int> result) {
    std::cout << "process2() start, current thread id " << std::this_thread::get_id() << std::endl;

    // 阻塞等待任务执行结果
    const int num = result.get();
    std::cout << "result = " << num << std::endl;

    std::cout << "process2() end, current thread id " << std::this_thread::get_id() << std::endl;
}

int main() {
    std::cout << "main() run, thread id " << std::this_thread::get_id() << std::endl;

    // 创建 Promise 对象
    std::promise<int> mpro;

    // 获取执行结果，特别注意：get_future() 应该在子线程执行前调用
    std::future<int> result = mpro.get_future();

    // 创建子线程 1，第一个参数是线程函数，第二个参数是 Promise 对象的引用，第三个参数毫秒数
    std::thread t1(process, std::ref(mpro), 5000);

    // 创建子线程 2，第一个参数是线程函数，第二个参数是 Future 对象
    // future 不支持拷贝，通过移动转移所有权
    std::thread t2(process2, std::move(result));

    // 阻塞等待子线程执行完成
    t1.join();
    t2.join();

    std::cout << "main() end, thread id " << std::this_thread::get_id() << std::endl;
    return 0;
}

程序运行的结果如下：

1
2
3
4
5
6
7
main() run, thread id 1
process2() start, current thread id 3
process() start, current thread id 2
process() end, current thread id 2
result = 5
process2() end, current thread id 3
main() end, thread id 1