【Example】C++ 标准库多线程同步及数据共享 (std::future 与 std::promise)

简介: 阅读此文章前,务必读懂:【Example】C++ 标准库 std::thread 与 std::mutex 否则你会像听天书一样懵。(...)

在任何语言的多线程编程当中,必然涉及线程的同步及数据的共享,方式也有很多种。

C++ 标准库当中提供了同步及共享的方案:std::future 与 std::promise 。

头文件:

#include <future>

 

一、std::future 与 std::promise

先从最基本且最原始的形式看起,std::future 与 std::promise 是互相配合使用的。

 

【负责访问】std::future 是一个模板类,它提供了可供访问异步执行结果的一种方式。

【语法】【伪代码】std::future<Type>name(promise.get_future());

 

【负责存储】std::promise 也是一个模板类,它提供了存储异步执行的值和异常的一种方式。

【语法】【伪代码】std::promise<Type>name;

 

 

先从最简单的代码入手:

#include <thread>#include <future>voidPromiseID(std::promise<std::thread::id>&po) {
try    {
po.set_value(std::this_thread::get_id());
    }
catch (conststd::exception&e)
    {
po.set_exception(std::current_exception());
    }
return;
}
intmain()
{
std::promise<std::thread::id>p1;
std::promise<std::thread::id>p2;
std::future<std::thread::id>f1(p1.get_future());
std::future<std::thread::id>f2(p2.get_future());
std::threadt1(&PromiseID, ref(p1));
std::threadt2(&PromiseID, ref(p2));
cout<<"thread id 1: "<<f1.get() <<endl;
cout<<"thread id 2: "<<f2.get() <<endl;
t1.join();
t2.join();
returnEXIT_SUCCESS;
}

 

 

以上代码和各种在你目前看来无厘头函数展示了 Print 两个线程 ID 的操作。

首先明白,std::future 负责访问,std::promise 负责存储,同时 promise 是 future 的管理者。

 

进而就可以先讲简单明了的逻辑:

std::future

1,std::future 是由 std::promise 创建的 (std::async 、std::packaged_task 也可创建 future),也是作为它的管理者。

2,std::future 也仅在创建它的 std::promise、std::async 、std::packaged_task 有效时才可用。

3,std::future 可供异步操作创建者用各种方式查询、等待、提取需要共享的值,也可以阻塞当前线程等待到异步线程提供值。

4,std::future 一个实例只能与一个异步线程相关联。多个线程则需要使用 std::shared_future。

5,std::future 的共享状态是由异步操作所使用的、且与其关联的 std::std::promise 所修改。(当然你单线程修改也行,但抬杠又有什么意义)

6,std::future 禁用了拷贝构造,但是可以进行移动(move)操作。

公共成员函数表:

名称 作用
operator=   移动 future 对象,移动!
share() 返回一个可在多个线程中共享的 std::shared_future 对象。
get() 获取值。(类型由模板类型而定)
valid() 检查 future 是否处于被使用状态,也就是它被首次在首次调用 get() 或 share() 前
wait() 阻塞等待调用它的线程到共享值成功返回。
wait_for() 在规定时间内 阻塞等待调用它的线程到共享值成功返回。
wait_until() 在指定时间节点内 阻塞等待调用它的线程到共享值成功返回。

 

共享状态:

补充一些与 std::future 相关的枚举类型,参考自Microsoft Docs:

future_errc 枚举 : 为 future_error 类报告的所有错误提供符号名称。

名称 示意
broken_promise 0 与其关联的 std::promise 生命周期提前结束。
future_already_retrieved 1 重复调用 get() 函数。
promise_already_satisfied 2 与其关联的 std::promise 重复 set。
no_state 4 无共享状态。

 

future_status 枚举:为计时等待函数可返回的原因提供符号名称。

名称 示意
ready 0 就绪
timeout 1 等待超时
deferred 2 延迟执行(与std::async配合使用)

 

 

std::promise

1,std::promise 负责存储,注意 std::promise 应当只使用一次。

2,std::promise 的统一初始化构造 "(p)" 是被禁用的,同时赋值运算符 "operator=" 作用为移动,std::promise 不可拷贝,但是可以被引用。

【注:此处应额外补充 alloc 构造函数】

3,std::promise 与 std::future 的状态相关联,它负责将共享值存入并给 std::future 访问使用,值类型也有可能是void、异常,当 std::future 端的阻塞函数接收到后,会立即解除阻塞状态。

4,std::promise 在作为使用者的异步线程当中,应当注意共享变量的生命周期、是否被 set 的问题。如果没有共享值没有被 set,而异步线程却结束,future 端会抛出异常。

5,std::promise 的 set 操作函数只能被调用一次。

6,std::promise 的 get_future() 函数只能被调用一次。

7,std::promise<void> 空类型创建是可以的,任何 set 函数不接受任何形式的参数,此操作用于传递通知,通知与其关联的 std::future 端解除阻塞。

 

公共成员函数表:

名称 作用
operator= 从另一个 std::promise 移动到当前对象。
swap() 交换移动两个 std::promise。
get_future() 获取与它关联的 std::future。
set_value() 设置值,类型由初始化时的模板类型而定。
set_value_at_thread_exit() 设置值,但是到该线程结束时才会发出通知。
set_exception() 设置异常,类型为 exception_ptr。
set_exception_at_thread_exit() 设置异常,但是到该线程结束时才会发出通知。

 

 

一个简单的例子:

#include <iostream>usingstd::cout;
usingstd::endl;
#include <vector>usingstd::vector;
#include <algorithm>#include <thread>#include <future>voidGetVectorMaxToPromise(constvector<int>&vec, std::promise<int>&po) {
try    {
autoit=std::max_element(vec.begin(), vec.end());
po.set_value_at_thread_exit(*it);
    }
catch (conststd::exception&)
    {
po.set_exception(std::current_exception());
    }
return;
}
voidPrintIntValue(std::future<int>&fu) {
cout<<"Value: "<<fu.get() <<endl;
return;
}
intmain()
{
vector<int>vec= { 1, 2, 3, 4, 5 };
std::promise<int>po;
std::future<int>fu(po.get_future());
std::threadt1(&GetVectorMaxToPromise, ref(vec), ref(po));
std::threadt2(&PrintIntValue, ref(fu));
t1.join();
t2.join();
returnEXIT_SUCCESS;
}

 

这个例子是一个线程获取 vector 当中的最大值并给另一个线程去 print。

在这个非常简单的例子当中可以看到通过 promise to future 做到了线程的同步与值的传递,还有异常的处理。

 

 

std::shared_future 与 std::packaged_task

std::future 有个非常明显的问题,就是只能和一个 std::promise 成对绑定使用,也就意味着仅限于两个线程之间使用。

那么多个线程是否可以呢,可以!就是 std::shared_future。

 

std::shared_future

它的语法是:

【语法】【伪代码】std::shared_future<Type>s_fu(pt.get_future());

 

std::shared_future 也是一个模板类,它的功能定位、函数接口和 std::future 一致,不同的是它允许给多个线程去使用,让多个线程去同步、共享:

#include <iostream>usingstd::cout;
usingstd::endl;
#include <vector>usingstd::vector;
#include <sstream>#include <string>usingstd::string;
#include <algorithm>#include <thread>#include <future>intGetVectorMax(constvector<int>&vec) {
return*(std::max_element(vec.begin(), vec.end()));
}
voidPrintIntValueOnShared(std::shared_future<int>&s_fu) {
s_fu.wait();
std::stringstreamss;
ss<<std::this_thread::get_id() <<" Value: "<<s_fu.get();
cout<<ss.str() <<endl;
return;
}
intmain()
{
vector<int>vec= { 1, 2, 3, 4, 5 };
std::packaged_task<int(constvector<int>&)>pt(GetVectorMax);
std::shared_future<int>s_fu(pt.get_future());
std::threadt1(&PrintIntValueOnShared, ref(s_fu));
std::threadt2(&PrintIntValueOnShared, ref(s_fu));
std::threadt3(&PrintIntValueOnShared, ref(s_fu));
Sleep(500); // Windows.hstd::thread(ref(pt), ref(vec)).join();
t1.join();
t2.join();
t3.join();
returnEXIT_SUCCESS;
}


是的,你还看到了另一个奇怪的东西:std::packaged_task。(...)

 


std::packaged_task

std::packaged_task 的作用是包装一个可调用对象(可能是函数,也可能是lambda)去给异步线程调用,简化 promise to future 的流程。

它的语法:

【语法】【伪代码】【Callback】std::packaged_task<Type(ArgsType, ...)>name(Callable);
【语法】【伪代码】【Lambda】std::packaged_task<Type(ArgsType, ...)>pl([](ArgsType, ...) {
returnTypeData;
});

 

是的,就像 std::function 那样。只不过它是用来给异步线程调用的:

成员函数表:

名称 作用
operator= 移动 std::packaged_task 对象,移动!
valid() 检查可调用对象是否有效。
swap() 交换移动两个 std::packaged_task。
get_future() 返回具有相关联异步状态的 std::future 对象。
operator() 执行该可调用对象。
make_ready_at_thread_exit 执行该可调用对像,但是到该线程结束时才会发出通知。
reset() 重置,并清空之前的值。

 

将上文例子变种演示一下:

std::packaged_task<int(constvector<int>&)>pl([](constvector<int>&vec) {
return*(std::max_element(vec.begin(), vec.end()));
});
std::shared_future<int>s_fu(pl.get_future());
if (pl.valid())
{
std::threadt1(&PrintIntValueOnShared, ref(s_fu));
std::threadt2(&PrintIntValueOnShared, ref(s_fu));
std::threadt3(&PrintIntValueOnShared, ref(s_fu));
Sleep(500); // Windows.hstd::thread(ref(pl), ref(vec)).join();
t1.join();
t2.join();
t3.join();
}

 

 

使用它需要注意的事项:

1,std::packaged_task 不能被拷贝,但是可以被移动,也可以被引用。

2,std::packaged_task 可以默认无参构造,但此时没有任何作用,执行会发生异常,valid() 值为 false。

3,std::packaged_task 的 get_future() 函数只能被调用一次。

4,std::packaged_task 绑定了可调用对象并已经运行,它的共享状态会一直持续到与它关联的 std::future 或最后一个 std::shared_future 结束为止。

5,std::packaged_task 应谨慎操作,它本身的生命周期应持续到所有与它关联的 future 结束后为止。

 

 

std::async

std::async 是一个函数模板,作用是异步运行可调用对象,最终将调用结果返回到 std::future 当中。

它的语法是:

【语法】【伪代码】std::async(LaunchEnum, Callable, Args, ...);
or【语法】【伪代码】std::async(Callable, Args, ...);

 

 

std::async的第一个枚举参数

launch 枚举: 展示描述模板函数 async 的可能模式的位掩码类型

名称 示意
async 0 异步调用 主动
deferred 1 延迟调用 被动

 

这两个枚举代表什么效果呢?请仔细看非常简单的例子:

#include <iostream>usingstd::cout;
usingstd::endl;
#include <string>usingstd::string;
#include <vector>usingstd::vector;
#include <chrono>#include <thread>#include <future>voidPrintFiveStr(conststring&str) {
for (size_ti=0; i<5; i++)
    {
cout<<str;
std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
return;
};
intmain()
{
vector<std::launch>launchs= {std::launch::async, std::launch::deferred};
for (auto&launch : launchs)
    {
std::future<void>add=std::async(launch, ref(PrintFiveStr), "+");
std::future<void>sub=std::async(launch, ref(PrintFiveStr), "-");
add.get();
sub.get();
cout<<endl;
    }
returnEXIT_SUCCESS;
}

 

三次运行效果:

【第一次】

+--++--+-+

+++++-----


【第二次】

+-+--++--+

+++++-----


【第三次】

+-+-+-+-+-

+++++-----

 

是的,最直观的就是:

std::launch::async 是在 std::async 初始化所有线程局域对象后执行可调用对象。

std::launch::deferred 是在 std::async 初始化后(期间完成内部std::thread对象创建),不执行可调用对象(内部std::thread也没有被初始化),在 std::async 返回的 std::future 首次调用非定时等待函数后,再去执行。

这就是[异步调用主动]与[延迟调用被动]的区别。

注意的是,如果不传第一个枚举参数,那么,std::async 优先使用哪种 launch 取决于编译器的实现机制。

 

额外技术细节请参考 C++ Reference:

函数模板 async 异步地运行函数 f (潜在地在可能是线程池一部分的分离线程中),并返回最终将保有该函数调用结果的 std::future 。

1) 表现如同以 policy 为 std::launch::async | std::launch::deferred 调用 (2) 。换言之, f 可能执行于另一线程,或者它可能在查询产生的 std::future 的值时同步运行。

2) 按照特定的执行策略 policy ,以参数 args 调用函数 f :

  • 若设置 async 标志(即 (policy & std::launch::async) != 0 ),则 async 在新的执行线程(初始化所有线程局域对象后)执行可调用对象 f ,如同产出 std::thread(std::forward<F>(f), std::forward<Args>(args)...) ,除了若 f 返回值或抛出异常,则于可通过 async 返回给调用方的 std::future 访问的共享状态存储结果。
  • 若设置 deferred 标志(即 (policy & std::launch::deferred) != 0 ),则 async 以同 std::thread 构造函数的方式转换 f 与 args... ,但不产出新的执行线程。而是进行惰性求值:在 async 所返回的 std::future 上首次调用非定时等待函数,将导致在当前线程(不必是最初调用 std::async 的线程)中,以 args... (作为右值传递)的副本调用 f (亦作为右值)的副本。将结果或异常置于关联到该 future 的共享状态,然后才令它就绪。对同一 std::future 的所有后续访问都会立即返回结果。
  • 若 policy 中设置了 std::launch::async 和 std::launch::deferred 两个标志,则进行异步执行还是惰性求值取决于实现。
  • 【C++ 14 开始】若 policy 中未设置 std::launch::async 或 std::launch::deferred 或任何实现定义策略标志,则行为未定义。

任何情况下,对 std::async 的调用同步于(定义于 std::memory_order )对 f 的调用,且 f 的完成先序于令共享状态就绪。若选择 async 策略,则关联线程的完成同步于首个等待于共享状态上的函数的成功返回,或最后一个释放共享状态的函数的返回,两者的先到来者。

 

完工!

2022-03-19 凌晨 4:23

AirChip




====================================

芯片烤电池 C++ Example 2022-Spring Season Pass :

【Example】C++ 标准库常用容器全面概述

【Example】C++ 回调函数及 std::function 与 std::bind

【Example】C++ 运算符重载

【Example】C++ 标准库智能指针 unique_ptr 与 shared_ptr

【Example】C++ 接口(抽象类)概念讲解及例子演示

【Example】C++ 虚基类与虚继承 (菱形继承问题)

【Example】C++ Template (模板)概念讲解及编译避坑

【Example】C++ 标准库 std::thread 与 std::mutex

【Example】C++ 标准库多线程同步及数据共享 (std::future 与 std::promise)

【Example】C++ 标准库 std::condition_variable

【Example】C++ 用于编译时封装的 Pimpl 演示 (编译防火墙 Private-IMPL)

【Example】C++ 单例模式 演示代码 (被动模式、兼容VS2022编译)

====================================

相关文章
|
16天前
|
缓存 Java 调度
Java并发编程:深入解析线程池与Future任务
【7月更文挑战第9天】线程池和Future任务是Java并发编程中非常重要的概念。线程池通过重用线程减少了线程创建和销毁的开销,提高了资源利用率。而Future接口则提供了检查异步任务状态和获取任务结果的能力,使得异步编程更加灵活和强大。掌握这些概念,将有助于我们编写出更高效、更可靠的并发程序。
|
28天前
|
存储 前端开发 安全
C++一分钟之-未来与承诺:std::future与std::promise
【6月更文挑战第27天】`std::future`和`std::promise`是C++异步编程的关键工具,用于处理未完成任务的结果。`future`代表异步任务的结果容器,可阻塞等待或检查结果是否就绪;`promise`用于设置`future`的值,允许多线程间通信。常见问题包括异常安全、多重获取、线程同步和未检查状态。解决办法涉及智能指针管理、明确获取时机、确保线程安全以及检查未来状态。示例展示了使用`std::async`和`future`执行异步任务并获取结果。
31 2
|
25天前
|
消息中间件 存储 开发工具
消息队列 MQ产品使用合集之C++如何使用Paho MQTT库进行连接、发布和订阅消息
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
7天前
|
C++
C++一分钟之-文件系统库(fs)的使用
【7月更文挑战第18天】C++17的`&lt;filesystem&gt;`库简化了文件系统操作,包括`path`类和`directory_iterator`。`path`用于表示路径,`directory_iterator`用于遍历目录。常用功能有路径拼接、分解,创建/删除目录,以及遍历。错误处理、跨平台兼容性和性能是使用时需考虑的关键点。示例代码展示了如何初始化`path`、创建目录、删除目录以及处理异常。
11 1
|
24天前
|
存储 安全 Linux
网络请求的高效处理:C++ libmicrohttpd库详解
网络请求的高效处理:C++ libmicrohttpd库详解
|
12天前
|
网络协议 安全 Python
我们将使用Python的内置库`http.server`来创建一个简单的Web服务器。虽然这个示例相对简单,但我们可以围绕它展开许多讨论,包括HTTP协议、网络编程、异常处理、多线程等。
我们将使用Python的内置库`http.server`来创建一个简单的Web服务器。虽然这个示例相对简单,但我们可以围绕它展开许多讨论,包括HTTP协议、网络编程、异常处理、多线程等。
|
18天前
|
存储 C++ 容器
C++一分钟之-正则表达式库(regex)
【7月更文挑战第7天】C++从C++11开始支持正则表达式,通过`&lt;regex&gt;`库提供功能。本文涵盖基本概念如`std::regex`、`std::smatch`,以及`regex_search`和`regex_match`的使用。常见问题包括大小写敏感性、特殊字符转义、贪婪与非贪婪匹配和捕获组。提供的代码示例展示了如何进行匹配、不区分大小写的匹配、特殊字符匹配、贪婪与非贪婪匹配和捕获组的使用。理解并练习正则表达式能提升文本处理效率。
15 0
|
22天前
|
存储 算法 程序员
C++基础知识(八:STL标准库(Vectors和list))
C++ STL (Standard Template Library标准模板库) 是通用类模板和算法的集合,它提供给程序员一些标准的数据结构的实现如 queues(队列), lists(链表), 和 stacks(栈)等. STL容器的提供是为了让开发者可以更高效率的去开发,同时我们应该也需要知道他们的底层实现,这样在出现错误的时候我们才知道一些原因,才可以更好的去解决问题。
|
22天前
|
算法 前端开发 C++
C++基础知识(八:STL标准库 deque )
deque在C++的STL(Standard Template Library)中是一个非常强大的容器,它的全称是“Double-Ended Queue”,即双端队列。deque结合了数组和链表的优点,提供了在两端进行高效插入和删除操作的能力,同时保持了随机访问的特性。
|
22天前
|
存储 C++ 索引
C++基础知识(八:STL标准库 Map和multimap )
C++ 标准模板库(STL)中的 map 容器是一种非常有用的关联容器,用于存储键值对(key-value pairs)。在 map 中,每个元素都由一个键和一个值组成,其中键是唯一的,而值则可以重复。

热门文章

最新文章