C++技能系列
Linux通信架构系列
C++高性能优化编程系列
深入理解软件架构设计系列
高级C++并发线程编程
期待你的关注哦！！！有更多博文系列等着看哦，会经常更新！！！
因为你的关注激励着我的创作！！！
青春是薄脆的蛋壳，你总要打破它，去经历风吹雨淋，在历练中成长。
Youth is tucking eggshell, you always want to break it, to experience the wind rain, grow in experience.
某线程只有先等另一线程的任务完成，才可以执行自己的任务。一般而言，线程常常需要等待特定的事件的发生，或等待某个条件成立。只要设置一个“任务完成”的标志，或者利用共享数据存储一个类似的标志，通过定期查验该标志就可以满足需求，但这远非理想的方法。C++标准库提供了工具：条件变量和future。
C++标准库提供了条件变量的两种实现：std::condition_variable和std::condition_variable_any。它们都在标准库的头文件<condition_variable>内声明。两者都需要配合互斥，方能提供妥当的同步操作。std::condition_variable仅限于于std::mutex一起使用；然而，只要某一类型符合成为互斥的最低标准，足以充当互斥，std::condition_variable_any即可与之配合使用，因此它的后缀是"_any”。由于std::condition_variable_any更加通用，他可能产生额外的开销，涉及其性能、自身的体积或系统资源等，因此std::condition_variable应予优先采用，除非有必要令程序更加灵活。
那么前面介绍的问题，我们该如何利用std::condition_variable？为了让线程A休眠，直至有数据需要处理才被唤醒，我们该如何做？
接下来我们运用条件变量std::condition_variable等待数据处理：
std::mutex mut;
std::queue<data_chunk> data_queue;
std::condition_variable data_cond;
void data_preparation_thread()
{
while(more_data_to_prepare())
{
data_chunk const data = perpare_data();
{
std::lock_quard<std::mutex> lk(mut);
data_queue.push(data);
}
data_cond.notify_one();
}
}
void data_processing_thread()
{
while(true)
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, []{return !data_queue.empty();});
data_chunk data = data_queue.front();
data_queue.pop();
lk.unlock();
process(data);
if(is_last_chunk(data))
break;
}
}
在线程传递数据的常见方法是运用队列，若队列的实现到位，同步操作就可以被限制到内部，从而大幅减少可能出现的同步问题和条件竞争。
线程安全队列的完整定义，其中采用了条件变量：
#include <queue>
#include <memory>
#include <mutex>
#include <condition_variable>
template<typename T>
class threadsafe_queue
{
private:
mutable std::mutex mut;
std::queue<T> data_queue;
std::condition_variable data_cond;
piblic:
threadsafe_queue()
{}
threadsafe_queue(threadsafe_queue const& other)
{
std::lock_guard<std::mutex> lk(other.mut);
data_queue = other.data_queue;
}
void push(T new_value)
{
std::lock_guard<std::mutex> lk(mut);
data_queue.push(new_value);
data_cond.notify_one();
}
void wait_and_pop(T& value)
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this]{return !data_queue.empty();});
value = data_queue.front();
data_queue.pop();
}
std::shared_ptr<T> wait_and_pop()
{
std::unique_lock<std::mutex> lk(mut);
data_cond.wait(lk, [this]{return !data_queue.empty();});
std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
data_queue.pop();
return res;
}
bool try_pop(T& value)
{
std::lock_guard<std::mutex> lk(mut);
if(data_queue.empty())
return false;
value = data_queue.front();
data_queue.pop();
return true;
}
std::shared_ptr<T> try_pop()
{
std::lock_guard<std::mutex> lk(mut);
if(data_queue.empty())
return std::shared_ptr<T>();
std::shared_ptr<T> res(std::make_shared<T>(data_queue.front()));
data_queue.pop();
return res;
}
bool empty();
{
std::lock_guard<std::mutex> lk(mut);
return data_queue.empty();
}
};
只要我们不急需线程运算的值，就可以使用std::async()按异步方式启动任务。我们从std::async()函数处获得std::future对象，运行的函数一旦完成，其返回值就有该对象最后持有。若要用到这个值，只需在future对象上调用get(),当前线程就会阻塞，以便future准备妥当并返回该值。
async使用方法和thread类类似，但是多了一个函数参数。async(A, B, C,....)：
其中这个A：是表示你是否创建一个新线程，参数类型是std::launch类型。
std::launch::deferred：表示该函数会延迟启动，直到future上调用 wait()或get()；
std::launch::async：表示该函数必须运行在他自己的线程上；
std::launch::deferred | std::launch::async：表明该函数可以有具体实现来选择(默认最后一个)；
B和C：分别是需要传入的函数和函数的参数，和thread一样，函数的参数是通过副本传入，如果需要传入引用则需要使用std::ref类。
future可以从异步任务中获取结果，一般与std::async配合使用，std::async用于创建异步任务，实际上就是创建一个线程执行相应任务。
std::future::get获取结果，如果调用过程中，任务尚未完成，则主线程阻塞至任务完成。
std::future::wait_for等待结果返回，wait_for可设置超时时间，如果在超时时间之内任务完成，则返回std::future_status::ready状态；如果在超时时间之内任务尚未完成，则返回std::future_status::timeout状态。
string fun_2() {
this_thread::sleep_for(chrono::seconds(3));
return "hello";
}
int main() {
auto x = async(fun_2);
cout << "456 ";
if (x.wait_for(chrono::seconds(1)) == future_status::timeout)
cout << "time out" << endl;
else
cout << x.get();
}
使用std::future取得std::sync异步任务的返回值，如下代码：
#include<iostream>
#include<future>
bool func(std::string &str)
{
str = "helloworld.";
std::this_thread::sleep_for(std::chrono::seconds(2));
return true;
}
int main() {
std::cout << "Hello, World!" << std::endl;
std::string stri;
std::future<bool> fut = std::async(func, std::ref(stri));
bool b = fut.get();
std::cout << "b: " << b << " stri: " << stri << std::endl;
return 0;
}
std::sync()向任务函数传递参数代码如下，详见代码备注：
#include<string>
#include<future>
struct X
{
void foo(int, std::string const&);
std::string bar(std::string const&);
}
X x;
// 1、调用p->foo(42, "hello"),其实p的值是&x,即x的地址
auto f1 = std::async(&X::foo, &x, 42, "hello");
// 2、调用tmpx.bar("goodbye"),其中tmpx是x副本
auto f2 = std::async(&X::bar, x, "goodbye");
struct Y
{
double operator()(double);
};
Y y
// 3、调用tmp(3.14)。其中，由Y()生成一个匿名变量，传递给std::async(),进而发生移动构造。在std::async()内部产生对象tmpy, 在tmpy上执行Y::operator()
auto f3 = std::async(Y(), 3.141);
// 4、调用y(2.78)
auto f4 = std::async(std::ref(y), 2.718);
X baz(X&);
// 5、baz(x)
std::async(baz, std::ref(x));
class move_only
{
public:
move_only();
move_only(move_only&&);
move_only(move_only const&) = delete;
move_only& operator=(move_only&&);
move_only& operator=(move_only const&) = delete;
void operator()();
};
// 6、调用tmp(),其中tmp等价于std::move(move_only()),它的产生过程与3相似
auto f5 = std::async(move_only);
std::packaged_task<>连结了future对象与函数。std::packaged_task<>对象在执行任务时，会调用关联函数，把返回值保存为future的内部数据，并令future准备就绪。它可作为线程池的构件单元，亦可用于其它任务管理。
例如，为各个任务分别创建专属的独立运行的线程，或者在某个特定的后台线程上依次执行全部任务。若一项庞杂的操作能分解为多个子任务，则可把它们分别包装到多个std::packaged_task<>实例中，再传递给任务调度器或者线程池。
（1）std::packaged_task使用详解，代码如下，详见备注：
#include<iostream>
#include<future>
#include<thread>
#include<functional>
//使用auto、decltype自动追踪返回值类型的泛型编程模版，返回两参数之和
template<class T1, class T2>
auto func(T1& t1, T2& t2)->decltype(t1, t2){return t1 + t2;}
//伪函数，定义一个类重载operator()实现实例化后成为可调用对象
class test{
int a, b;
public:
test(int a, int b) : a(a), b(b){}
int operator()(int _a, int _b){return a + b + _a + _b;}
};
//普通函数，充当函数指针
int count(std::string s, std::string t){
s.pop_back();
t.pop_back();
return sizeof(s + t) / sizeof(wchar_t);
}
int add(int a, int b){
return a+ b;
}
void func2(future<int>& fut, packaged_task<int(int, int)>& task){
packaged_task<int(int, int)> task1(add);
task = move(task1);
fut = task.get_future();
task.make_ready_at_thread_exit(1, 2);
}
std::future<int> launcher(std::packaged_task<int(int)>& task, int arg){
if(task.valid())
{
std::future<int> ret = task.get_future();
std::thread(std::move(task), arg).detach();
return ret;
}
else return std::future<int>();
}
int main()
{
//代码片段1: function pointer
{
auto data1 = 1.11;
auto data2 = 2.22;
std::packaged_task<decltype(data1 + data2)(decltype(data1)&, decltype(data2)&)> task1(func<decltype(data1)&, decltype(data2)&>);
auto thread1 = std::thread(std::move(task1), std::ref(data1), std::ref(data2));
thread.join();
std::cout << "This is pointer task, task1 resut is : " << fut1.get() << std::endl;
}
//代码片段2: lamda function
{
auto u = 1;
auto v = 2;
std::packaged_task<decltype(u + v)(decltype(u), decltype(v))>
task2([u, v](decltype(u), decltype(v)) -> decltype(u + v){ return u + v;});
std::future<decltype(u + v)> fut2 = task2.get_future();
std::this_thread::sleep_for(std::chrono::seconds(3));
task2(u, v);
std::cout << "This is lamda task, task2 resut is : " << fut2.get() << std::endl;
}
//代码片段3: lamda function
{
std::packaged_task<int<int, int>> task3(test(1, 2));
auto fut3 = task3.get_future();
std::this_thread::sleep_for(std::chrono::seconds(3));
task3(3, 4);
std::cout << "This is class test task, task3 result is: " << fut3.get() << std::endl;
}
//代码片段4: class function
{
std::string s1 = "Allen", s2 = "Su";
std::packaged_task<int()> task4(std::bind(count, s1, s2));
auto fut4 = task4.get_future();
auto thread4 = std::thread(std::chrono::seconds(3));
thread.join();
std::cout << "This is bind task, task4 result is: " << fut4.get() << std::endl;
}
//代码片段5: std::packaged_task::reset()
//重置状态，摒弃之前运行的结果
{
std::packaged_task<int(int, int)> task([](int a, int b)){
return a + b;
});
std::future<int> result = task.get_future();
task(2, 9);
std::cout << "2 + 9 = " result.get() << std::endl;
task.reset();
result = task.get_future();
std::thread task_td(std::move(task), 2, 10);
task_td.join();
std::cout << "2 + 10 = " result.get() << std::endl;
}
//代码片段6: std::packaged_task::make_ready_at_thread_exit()
//在被调用线程退出后，将共享状态的标志设置ready。
//如果std::future对象尝试使用get()获取共享状态的值（通常是在其他线程里），
//则会被阻塞直到本线程退出。
{
std::packaged_task<int(int, int)> task;
std::future<int> fut;
auto thd = thread(func2, std::ref(fut), std::ref(task));
thd.join();
auto result = fut.wait_for(chrono::second(2));
if(result == future_state::ready)
std::cout << fut.get() << std::endl;
}
//代码片段7: std::packaged_task::valid()
//检查是否是有效的共享状态，有则返回ture,否则返回false。
{
std::packaged_task<int(int)> task([](int x){ return x * 2});
std::future<int> fut = launcher(task, 25);
std::cout << "The double of 25 is " << fut.get() << std::endl;
}
return 0;
}
std::packaged_task是一种将函数和其期望的返回值打包为可调用对象的类模板。这个可调用对象可以在另一个线程中异步执行，也可以像普通函数一样直接调用。
（2）下面是一个简单的案例代码，使用std::packaged_task实现了一个计算阶乘的函数，并在另一个线程中异步执行：
#include <iostream>
#include <future>
using namespace std;
long long factorial(int n) {
long long result = 1;
for (int i = 2; i <= n; ++i) {
result *= i;
}
return result;
}
int main() {
int n = 10;
// 创建一个std::packaged_task，打包计算阶乘的函数
std::packaged_task<long long(int)> task(factorial);
// 获取与该std::packaged_task关联的std::future
std::future<long long> future = task.get_future();
// 在另一个线程中异步执行该std::packaged_task
std::thread thread(std::move(task), n);
// 等待异步执行的结果，并输出
cout << "factorial(" << n << ") = " << future.get() << endl;
// 关闭线程
thread.join();
return 0;
}
（3）使用std::packaged_task在GUI线程上运行代码：
#include <deque>
#include <mutex>
#include <future>
#include <thread>
#include <utility>
std::mutex m;
std::deque<std::packaged_task<void>> tasks;
bool gui_shutdown_message_received();
void get_and_process_gui_message();
void gui_thread()
{
while(!gui_shutdown_message_received())
{
get_and_rpocess_gui_message();
std::packaged_task<void> task;
{
std::lock_guard<std::mutex> lk(m);
if(tasks.empty())
continue;
task = std::move(tasks.front);
tasks.pop_front();
}
task();
}
}
std::thread gui_bg_thread(gui_thread);
template<typename Func>
std::future<void> post_task_for_gui_thread(Func f)
{
std::packaged_task<void()> task(f);
std::future<void> res = task.get_future();
std::lock_guard<std::mutex> lk(m);
tasks.push_back(std::move(task));
return res;
}
std::promise<T>给出一个异步求值的方法，某个std::future对象与之关联，能延后读出需要求取的值。配对的std::promise和std::future可实现下面的工作机制：等待数据的线程在future上阻塞，而提供数据的线程利用相配的promise设定关联的值，使future准备就绪。std::promise的值通过成员函数set_value()设置，只要设置好，std::future即准备就绪，凭借它就能获取该值。如果promise在被销毁时仍未曾设置值，保存的数据则由异常代替。
void test(std::promise<double> &promise)
{
promise.set_value(1.1);
}
int main()
{
std::promise<double> promise;
std::future<double> fut = promise.get_future();
thread t(test, ref(promise));
t.detach();
double y = fut.get();
}
std::async和std::packaged_task是支持异常抛出的，不会陷入死等状态。std::promise它是通过成员函数显示调用实现，假如，我们不想保存值，而想保存异常，就不应该调用set_value()而应该调用成员函数set_exception()。可以看下如下代码块：
void test(std::promise<double> &promise)
{
try {
vector<int> vec = {1};
int a = vec.at(2); //数组越界，抛异常
}catch (...){
cout << "exception" << endl;
promise.set_exception(std::current_exception());
}
}
int main()
{
std::promise<double> promise;
std::future<double> fut = promise.get_future();
test(promise);
double y = fut.get();
}
这里的std::current_exception()用于捕获当前的异常。此外，我们还能用std::make_exception_ptr()直接保存新异常，而不触发抛出行为：
promise.set_exception(std::current_exception(std::make_exception_ptr(std::logic_error("foo"))));
假设，某个线程按计划仅仅等待一次，只要条件成立一次，它就不再理会条件变量。条件变量未必是这种同步模式的最佳选择。若我们所等待的条件变量需要判定某份数据是否可用， std::future 更适合此场景。
C++标准库提供有两种future，分别由两个类模版提供，其声明在头文件<future>内，独占future（unique future,即std::future<>）和共享future（share future, 即std::shared_future<>）。同一事件仅仅允许关联唯一一个std::future实例，但可以关联多个std::shared_future实例，只要目标事件发生，与后者关联的所有实例就会同时就绪，并且它们全都可以访问与该目标关良数据的类型。
虽然future能用于线程间通信，但是future对象本身不提供同步访问。若多个线程需访问同一个std::future对象，必须用互斥或其他同步方式进行保护。
一个std::shared_future<>对象可能派生出多个副本，这些副本都指向同一个异步结果，由多个线程分别独占，它们可以访问属于自己的那个副本而无须互相同步。
(1) 使用std::move向其默认构造函数传递归属权
std::promise<int> p;
std::future<int> f(p.get_future());
assert(f.valid); //future对象f有效
std::share_future<int> sf(std::move(f));
assert(!f.valid()); //future对象f不再有效
assert(sf.valid()); //future对象sf开始生效
有时候某线程的值不止被一个线程所需要，而get()却只能只用一次，这时可以通过std::shared_future达到多次通过get()获取值的目的：
std::future<int>myf = mypt.get_future();
std::shared_future<int>myf_s(myf.share());
std::thread t2(mythread1,std::ref(myf_s));
t2.join();
auto mysf = myf_s.get();
cout << mysf << " -" << endl;
cout << "---------------" << endl;
auto mysf2 = myf_s.get();
cout << mysf2 << " -" << endl;
(2) 隐式转移归属权
std::promise<std::string> p;
std::shared_future<std::string> sf(p.get_future());
(3) 使用future的成员函数share(),直接创建新的std::shared_future对象
std::promise<std::string> p;
auto sf = p.get_future().share();
如果出现阻塞动作可能漫无止境，只要锁等待的目标事件还未发生，线程就一直暂停。通常情况下没有问题，但是某种场景下会限制时长。
c++中的有两种超时机制可供选用：
迟延超时：线程根据指定时长而继续等待。
（处理迟延超时函数变体以“_for“为后缀。）
绝对超时：在某特定时间点来临之前，线程一直等。
（处理绝对超时的函数变体以“_until”为后缀。）
c++中的三种时钟：
steady_clock 是单调的时钟，相当于教练手中的秒表；只会增长，适合用于记录程序耗时；system_clock 是系统的时钟；因为系统的时钟可以修改；甚至可以网络对时； 所以用系统时间计算时间差可能不准。（可能发生时间跳变）high_resolution_clock 是当前系统能够提供的最高精度的时钟；它也是不可以修改的。相当于steady_clock的高精度版本。一般限时等待都用单调时钟，也叫恒稳时钟。system_clock 是系统的时钟，可能发生时间跳变。
（1）std::condition_variable限时等待的处理
std::condition_variable cv;
std::mutex cv_m;
bool flag;
// condition_variable wait_for超时触发
{
std::unique_lock<std::mutex> lk(cv_m);
auto now = std::chrono::steady_clock::now();
if(cv.wait_for( lk, std::chrono::seconds (2) ) == std::cv_status::timeout){
}
}
// condition_variable wait_for条件触发
{
std::unique_lock<std::mutex> lk(cv_m);
auto now = std::chrono::system_clock::now();
if(!cv.wait_for(lk, std::chrono::milliseconds(200) , [](){return flag == ture;})){
std::cout << "timeout." << std::endl;
}
}
//condition_variable waits_until条件触发
{
std::unique_lock<std::mutex> lk(cv_m);
auto now = std::chrono::system_clock::now();
if(!cv.wait_until(lk, now + std::chrono::milliseconds(200), [](){return i == 1;})){
std::cout << "timeout." << std::endl;
}
}
//跟据时机通知解锁操作
cv.notify_all();
（2）std::future如何使用限时等待的处理
wait函数：
【1】等待共享状态就绪。
【2】如果共享状态尚未就绪(即提供者尚未设置其值或异常)，则该函数将阻塞调用的线程直到就绪。
【3】当共享状态就绪后，则该函数将取消阻塞并void返回。
wait_for函数：
【1】等待共享状态在指定的时间内(time span)准备就绪。
【2】 如果共享状态尚未就绪(即提供者尚未设置其值或异常)，则该函数将阻塞调用的线程直到就绪或已达到设置的时间。
【3】此函数的返回值类型为枚举类future_status。此枚举类有三种label：ready：共享状态已就绪；timeout：在指定的时间内未就绪；deferred：共享状态包含了一个延迟函数(deferred function),延迟执行，当std::async()第一个参数为std::lanuch::deferred时生效。
wait_until函数：
【1】等待共享状态在指定的时间点(time point)准备就绪。
【2】如果共享状态尚未就绪(即提供者尚未设置其值或异常)，则该函数将阻塞调用的线程直到就绪或已达到指定的时间点。
【3】此函数的返回值类型为枚举类future_status。
返回值类型为future_status，该函数将本线程阻塞在当前，并等待一段时间，后继续执行，若在等待时间内wait_for()绑定线程执行完毕，则返回ready，未执行完毕则返回timeout。
bool func(std::string &str)
{
str = "helloworld.";
std::this_thread::sleep_for(std::chrono::seconds(1));
return true;
}
int main() {
std::string stri;
std::future<bool> fut = std::async(std::launch::async, func, std::ref(stri));
//枚举类型
//设置等待3s,根据设置等待时间与子线程执行消耗时间得到返回值。决定程序在当前行阻塞时间。
std::future_status status = fut.wait_for(std::chrono::seconds(3));
if (status == std::future_status::timeout)//子线程还没执行完
{
std::cout << "timeout..." << std::endl;
}
else if (status == std::future_status::ready)//子线程已执行完
{
std::cout << "ready..." << std::endl;
std::cout << fut.get() << std::endl;
}
当std::async()第一个参数为std::lanuch::deferred时生效。此时线程不在阻塞在wait_for()处，而是继续执行直到遇见属于该future对象的get()，如代码片段：
std::future<bool> fut = std::async(std::launch::deferred, func, std::ref(stri));
else if (status == std::future_status::deferred)//直接进入分支
{
std::cout << "ready..." << std::endl;
std::cout << fut.get() << std::endl; //等待
}
注意：无论std::async()是否延迟执行，异步线程都将会指向完程序才能结束，三种结束方式：
wait_for()处等待异步线程结束get()处等待异步线程结束return 0;处等待异步线程结束get()函数只能使用一次，因为get()函数的设计是一个移动语义，相当于将future对象中的值转移到了get()调用者中，所以再次get()就报告了异常。
所谓消息传递机制，注册函数回调，如下代码lambda表达式实现的消息回调：
#include <iostream>
#include <functional>
#include <thread>
class Test{
public:
using callback_t = std::function<void(std::string msg)>;
Test() = default;
~Test() = default;
void SetCallBack(callback_t cbk){
callback_ = cbk;
}
void Execute(){
std::thread t([&](){
//do something
int b = 4;
std::string str = std::to_string(b);
callback_(str);
});
if (t.joinable())
t.join();
}
private:
callback_t callback_;
};
int main() {
std::cout << "Hello, World!" << std::endl;
Test test;
test.SetCallBack([](std::string str){
std::cout << str << std::endl;
});
test.Execute();
return 0;
}
这里不再详细讲述boost库线程，大家可以去查阅学习。
boost::thread_group线程组类的用法，等待多个并发线程的处理的结果：
#include <boost/atomic.hpp>
#include <boost/thread.hpp>
#include <cassert>
boost::atomic_int g_counter(0);
void some_function() {
++ g_counter;
}
int main(int argc, char* argv[]) {
// 使用线程组创建10个线程
boost::thread_group threads;
for(unsigned i=0; i<10; ++i) {
threads.create_thread(&some_function);
}
threads.join_all();
assert(g_counter == 13);
}
灵活运用：
1、如何使用条件变量std::condition_variable线程同步？
2、如何使用线程安全队列线程同步？
4、如何使用std::future取得std::sync异步任务的返回值？
5、如何通过std::sync()向任务函数传递参数的？
6、如何使用std::packaged_task在线程间传递任务？
7、如何使用std::promise进行线程间同步？
8、如何将std::promise异常保存到std::future中？
9、如何使用std::shared_future多个线程一起等待？
10、std::condition_variable和std::future如何使用限时等待的处理？
11、如何使用消息传递进行同步？
12、如何等待多个并发线程的std::future的处理？