点击上方“C语言与CPP编程”，选择“关注/置顶/星标公众号”
干货福利，第一时间送达！
你好，我是飞宇。
昨天在朋友圈分享了一下自己关于《C++并发编程实战》这本书的读书笔记，收到不少点赞，今天就在公众号上分享一下自己以前的读书笔记，今天继续更新。
前段时间也更新了第一部分：《C++并发编程实战》读书笔记（1）：并发、线程管控
这里也顺便放一下自己的个人联系方式的二维码，听说以后公众号后不可以放二维码了，我也会经常在微信上分享一些计算机学习经验以及工作体验，还有一些内推工作机会，一般不闲聊，欢迎来做点赞之交。
C++国内并发编程的书少得可怜，唯一本质量还稍微靠谱点的就这本了《C++并发编程实战》，算是矮子里了挑将军挑出来的吧。
第4章 并发操作的同步
4.1 等待事件或等待其他条件
如果线程甲需要等待线程乙完成任务，可以使用C++标准库的条件变量来等待事件发生。
例如可以实现一个生产者消费者模型，通过队列来传递数据，一端准备数据另一端处理数据，其中条件变量的作用是消费者线程取出数据前检查队列是否非空，否则释放锁并等待生产者线程准备数据。
std::mutex mut;std::queuedata_queue; std::condition_variable data_cond;void data_preparation_thread() {while (...) {const Widget data = prepare_data();{std::lock_guard<std::mutex> lk(mut);data_queue.push(data);}// 通知消费者线程data_cond.notify_one();}}void data_processing_thread() {while (...) {// 需要多次加锁解锁，所以用unique_lockstd::unique_lock<std::mutex> lk(mut);// wait首先判断lambda，成立则返回，否则解锁互斥进入阻塞// 每次被notify后解除阻塞并获取锁，重复上述过程data_cond.wait(lk, [] { return !data_queue.empty(); });Widget data = data_queue.front();data_queue.pop();lk.unlock();process(data);}}
也可以实现一个简略的线程安全的队列：
template <typename T>class threadsafe_queue {private:mutable std::mutex mut;std::queuedata_queue; std::condition_variable data_cond;public: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();}};
4.2 使用future等待一次性事件发生
若线程需等待某一次性事件，可以以适当方式取得一个代表目标事件的future，此后线程就可以一边执行其他任务一边在future上等待。一旦目标事件发生，future就进入就绪状态，无法重置。
4.2.1 从后台任务返回值
并不急需某任务的返回值时，可以用async异步地启动任务，获得一个future对象；对后者调用get会阻塞当前线程，等待future准备完并返回该值。
int f() { ... }std::future<int> answer = std::async(f);...std::cout << answer.get();
给async的任务函数传递参数类似给thread传递参数。
// 调用成员函数的情况// p=&x; p->foo("hello")auto f1 = std::async(&X::foo, &x, "hello");// tmpx=x; tmpx.foo("hello")auto f2 = std::async(&X::foo, x, "hello");// 调用仿函数的情况// tmpy=y; tmpy(3.14)auto f3 = std::async(Y(), 3.14);// y(3.14)auto f4 = std::async(std::ref(y), 3.14);// 函数形参为引用的情况X baz(X&);auto f5 = std::async(baz, std::ref(x));
可以给async传递参数指定运行方式，deferred代表直到在future上调用wait/get才执行任务函数，async代表开启专属线程来执行；默认为deferred|async。
auto f = std::async(std::launch::async, Y(), 1.2);4.2.2 关联future实例与任务
类模板packaged_task把任务包装起来，可作为任务调度器、线程池的构建单元，其模板参数是函数签名，例如int(int,double*)。它具备函数调用操作符，参数取决于上述模板参数，调用时将参数传递给任务函数，通过get_future获取future对象，异步运行得到结果后保存到该对象。
例如图形用户界面需要接收其他线程的消息来更新界面。
std::mutex m;std::deque<std::packaged_task<void()> > tasks;// 图形用户界面的线程函数void gui_thread() {while (...) {get_and_process_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();}}// 其他线程通过该函数传递消息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;}
4.2.3 创建std::promise
有些任务无法以简单的函数调用表达，或者执行结果来自多个部分的代码，那么就需要使用std::promise显式地异步求值。
promise通过get_future获取关联的future对象，等待数据的线程在future上阻塞，提供数据的线程通过set_value设置数据，设置完后future即就绪。若promise销毁时仍未set_value，则传递异常。
下面是单线程处理多个连接的例子。这里假设传入的数据包含有ID与荷载数据，接收后将ID与promise对应，将相关值设为荷载数据。对于传出的数据而言，promise的相关值是代表是否成功的bool。
void process_connections(connection_set& connections) {while (...) {for (connection_iterator connection = ...) {if (connection->has_incoming_data()) {data_packet data = connection->incoming();std::promise& p = connection->get_promise(data.id); p.set_value(data.payload);}if (connection->has_outgoing_data()) {outgoing_packet data = connection->top_of_outgoing_queue();connection->send(data.payload);data.promise.set_value(true);}}}}
async与packaged_task运行的函数抛出异常时会保存在future对象中，调用get时再次抛出。对于promise而言，应用set_exception保存异常
some_promise.set_exception(std::make_exception_ptr(std::logic_error("foo")));4.2.4 多个线程一起等待
shared_future可以让多个线程等待同一个目标事件。每个线程复制一份shared_future副本，成为各线程独有的局部变量；通过该局部变量访问将由标准库自动同步，可以安全地访问。
std::promise<int> p1;auto f1 = p1.get_future();assert(f1.valid());std::shared_future<int> sf1 = std::move(f1);assert(!f1.valid());assert(sf1.valid());std::promise<int> p2;auto sf2 = p2.get_future().share();
4.3 限时等待
之前介绍的所有可能阻塞的调用，其阻塞都可能漫无止境。为此可以采用一些超时机制：延迟超时表示等待一定时间，后缀为for，绝对超时表示等待到某时间点，后缀为until。
std::chrono库中时钟是时间信息的来源，每个时钟类都提供当前时刻now、时间值的类型time_point、计时单元的长度ratio<>、计时速率是否恒定is_steady。常用时钟类包括system_clock，steady_clock，high_resolution_clock。
时长类duration<>，其模板参数有两个，第一个指采用何种类型表示计时单元的数量，第二个指每个计时单元代表多少秒。例如std::chrono::duration
auto f = std::async(some_task);if(f.wait_for(std::chrono::milliseconds(35))==std::future_status::ready){process(f.get());}
时间点类time_point<>，模板参数有两个，第一个指参考时钟，第二个指计时单元，即特化的duration。
std::condition_variable cv;bool done;std::mutex m;bool wait_loop() {auto const timeout =std::chrono::steady_clock::now() + std::chrono::milliseconds(500);std::unique_lock<std::mutex> lk(m);while (!done) {if (cv.wait_until(lk, timeout) == std::cv_status::timeout) break;}return done;}
4.4 运用同步操作简化代码
在并发实战中可以使用贴近函数式编程的风格，函数调用的结果完全取决于参数而非任何外部状态。线程间不会直接共享数据，而是由各任务分别预先准备妥自己所需的数据，随后通过future将结果发送到其他有需要的线程。
例如可以实现并行的快排：
template <typename T>std::listparallel_quick_sort(std::list input) { if (input.empty()) {return input;}// 将input的开头剪切到result// 以此为分界值，将input分为两段std::listresult; result.splice(result.begin(), input, input.begin());T const& pivot = *result.begin();auto divide_point = std::partition(input.begin(), input.end(),[&](T const& t) { return t < pivot; });// 异步处理较小的一段std::listlower_part; lower_part.splice(lower_part.end(), input, input.begin(), divide_point);std::future<std::list> new_lower(std::async(¶llel_quick_sort , std::move(lower_part))); // 本线程处理较大的一段auto new_higher(parallel_quick_sort(std::move(input)));// 汇合所有结果result.splice(result.end(), new_higher);result.splice(result.begin(), new_lower.get());return result;}
除了函数式编程，CSP（通信式串行线程）也有同样特性，其中线程完全隔离，没有共享数据，通过管道传递消息。具体代码这里不再演示。
C++20中还提出两个新特性：latch和barrier。latch是一个同步对象，内含计数器，减到0时就绪。
void foo() {unsigned const thread_count = ...;latch done(thread_count);my_data data[thread_count];std::vector<std::future<void> > threads;for (unsigned i = 0; i < thread_count; ++i)threads.push_back(std::async(std::launch::async, [&, i] {data[i] = make_data(i);done.count_down();...}));done.wait();process_data(data, thread_count);}
而barrier针对一组给定的线程，每个线程运行到barrier处就阻塞，直到同组的所有线程都抵达才释放。
void process_data(data_source &source, data_sink &sink) {unsigned const num_threads = ...barrier sync(num_threads);std::vectorthreads(num_threads); std::vectorchunks; result_block result;for (unsigned i = 0; i < num_threads; ++i) {threads[i] = joining_thread([&, i] {while (...) {if (!i) {data_block current_block = source.get_next_data_block();chunks = divide_into_chunks(current_block, num_threads);}sync.arrive_and_wait();result.set_chunk(i, num_threads, process(chunks[i]));sync.arrive_and_wait();if (!i) {sink.write_data(std::move(result));}}});}}
你好，我是飞宇，本硕均于某中流985 CS就读，先后于百度搜索以及字节跳动电商等部门担任Linux C/C++后端研发工程师。
同时，我也是知乎博主@韩飞宇，日常分享C/C++、计算机学习经验、工作体会，欢迎点击此处查看我以前的学习笔记&经验&分享的资源。
我组建了一些社群一起交流，群里有大牛也有小白，如果你有意可以一起进群交流。
欢迎你添加我的微信，我拉你进技术交流群。此外，我也会经常在微信上分享一些计算机学习经验以及工作体验，还有一些内推机会。
加个微信，打开另一扇窗
