C++并发编程 - 同步并发操作

前言

 一个好的程序员是那种过单行线马路都要往两边看的人。

  线程同步的本质是防止临界区(公共资源)并发操作，即多个线程禁止同时操作临界区。为此，在程序中以某种手段，将多个线程按照先后顺序访问临界区。

  临界区的操作一直要保持谨慎。多线程访问临界区，同时读还好；假如一个写的同时，另一个读，那这个读的值可能不是确定的，有可能是写之前的也有可能是写之后的，这可能会引发重大bug，且难以排查。

  本文主要介绍如何使用c++11中条件变量以及期望来设计并发操作。

条件变量 std::condition_variable

  在多线程任务中，线程通常使用条件变量阻塞自身，直至条件发生。即A线程阻塞等待某个条件变量，B线程通知条件变量变化解除A线程阻塞。为了防止竞争，条件变量的使用总是和互斥锁结合在一起。

  C++标准库对条件变量有两套实现: std::condition_variable 和 std::condition_variable_any。其中前者仅限于与std:mutex配合使用，后者可以和任何满足最低标准的互斥量一起使用。

  从体积、性能以及系统资源的使用方面，推荐使用 std::condition_variable。仅当对灵活性有硬性要求的情况下，才会选择 std::condition_variable_any。

std::condition_variable 实例：

   

    

     

      

     
     

      

       static std::mutex mut;
      
     
    

     

      

     
     

      

       static bool cond_value = false;
      
     
    

     

      

     
     

      

       static std::condition_variable cond;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       // 阻塞线程
      
     
    

     

      

     
     

      

       void wait_thread()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           std::unique_lock<std::mutex> lk(mut); //a
      
     
    

     

      

     
     

      

           cond.wait(
      
     
    

     

      

     
     

      

               lk, []{ return cond_value; }); 
      
     
    

     

      

     
     

      

           //lk.unlock();    // b
      
     
    

     

      

     
     

      

           //...  //c
      
     
    

     

      

     
     

      

       }
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       // 通知线程
      
     
    

     

      

     
     

      

       void notify()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           std::lock_guard<std::mutex> lk(mut); 
      
     
    

     

      

     
     

      

           cond_value = true;
      
     
    

     

      

     
     

      

           cond.notify_one();
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  
  

• 流程:
    step 1. wait_thread先获取锁，然后查询cond_value为false时，先解锁，然后阻塞线程，等待其他线程通知。
    step2. notify()获取锁，更改cond_value为true，并通过notify_one()通知阻塞的线程，并解锁。
    step3. wait_thread接收到通知，重新获取锁，检查cond_value为true，从wait()返回（解除阻塞）。返回时仍处于持有锁状态，直至互斥锁被析构或者手动解锁。

  在多线程中持有锁时间过长是一件糟糕的事情，当处理完与互斥锁相关的共享数据时，就应该立刻解锁。故上述c处还有其他业务时，b处有必要解锁。

  另外，还存在多个线程等待同一事件。此种场景，可通过notify_all()通知所有阻塞的线程检查条件。

「注」：wait()会去检查这些条件(通过调用所提供的lambda函数)， 当条件满足(lambda函数返回true)时返回。如果条件不满足(lambda函数返回false)， wait()函数将解锁互斥量， 并且将这个线程(上段提到的处理数据的线程)置于阻塞或等待状态。当准备数据的线程调用notify_one()通知条件变量时， 处理数据的线程从睡眠状态中苏醒， 重新获取互斥锁， 并且对条件再次检查，在条件满足的情况下， 从wait()返回并继续持有锁。当条件不满足时， 线程将对互斥量解锁，并且重新开始等待。

期望 std::future

  是指某个线程只等待一个特定的一次性事件。C++标准库将这种一次性事件称为“期望” (future)。

  当一个线程需要等待一个特定的一次性事件时，在某种程度上来说它就需要知道这个事件在未来的表现形式。之后，这个线程会周期性(较短的周期)的等待或检查，事件是否触发(检查信息板)；在检查期间也会执行其他任务。另外，在等待任务期间它可以先执行另外一些任务，直到对应的任务触发，而后等待期望的状态会变为“就绪”(ready)。

  在C++标准库中， 有两种“期望”， 使用两种类型模板实现， 声明在头文件中: 唯一期望(uniquefutures)( std::future<> )和共享期望(shared futures)( std::shared_future<> )。这是仿照 std::unique_ptr 和 std::shared_ptr 。std::future 的实例只能与一个指定事件相关联，而 std::shared_future 的实例就能关联多个事件。后者的实现中， 所有实例会在同时变为就绪状态， 并且他们可以访问与事件相关的任何数据。

「注」: 以上两段描述，摘抄于《C++并发编程实战》

  std::future并非单独使用，在C++标准库std::async、std::packaged_task和std::promise关联了std::future。即std::async、std::packaged_task和std::promise会返回std::future类型，线程通过std::future获取任务执行的结果。

  了解这些，我们可以使用std::future程序上实现业务与任务的分离。即业务线程只负责处理逻辑，任务线程负责任务执行，业务线程又能获取到任务执行的结果或其他的设计。

std::async 带返回值的后台任务

  当需要执行一个耗时的任务，在不阻塞主线程的条件下，还需要关心这个任务执行的结果时(例如是获取计算结果)。在不考虑使用全局变量情况下（全局变量，代码大忌），std::thread就不能满足这个需求了，此时std::async可以满足需求。

std::async使用案例

  std::async可启动一个异步任务。与std::thread对象等待运行方式的不同，std::async会返回一个std::future对象， 这个对象持有最终计算出来的结果。当需要这个结果时，只需要调用这个对象的get()成员函数；并且直到“期望”状态为就绪的情况下，线程才会阻塞；之后，返回计算结果。

「std::async 实例」:

   

    

     

      

     
     

      

       void async_test()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           struct SThreadParam
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      

               int func(int val1, int val2)
      
     
    

     

      

     
     

      

               {
      
     
    

     

      

     
     

      

                   LOGD("async func start...\n");
      
     
    

     

      

     
     

      

                   sleep(2);           // Simulated 2S time consumption
      
     
    

     

      

     
     

      

                   return val1 + val2;
      
     
    

     

      

     
     

      

               }
      
     
    

     

      

     
     

      

           };
      
     
    

     

      

     
     

      

           SThreadParam paramFunc;
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

           LOGD("create async...\n");
      
     
    

     

      

     
     

      

           std::future<int> f4 = std::async(std::launch::async, &SThreadParam::func, &paramFunc, 10, 12);
      
     
    

     

      

     
     

      

           LOGD("async debug...\n");
      
     
    

     

      

     
     

      

           //sleep(2);
      
     
    

     

      

     
     

      

           LOGD("async return %d.\n", f4.get());
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  
  

「执行结果」：

   

    

     

      

     
     

      

       2022-08-27 00:46:44.951  179 D: create async...
      
     
    

     

      

     
     

      

       2022-08-27 00:46:44.951  181 D: async debug...
      
     
    

     

      

     
     

      

       2022-08-27 00:46:44.951  173 D: async func start...
      
     
    

     

      

     
     

      

       2022-08-27 00:46:46.953  183 D: async return 22.
      
     
   
  

  一般情况下，std::async()任务函数默认情况下在创建时启动，在调用f4.get()时，若线程未执行完，则阻塞至执行完；否则，立刻返回结果（主线程sleep(2)是否注释，后两条打印都相差2s）。当然也有不一般的情况，见下文「注」。

指定async任务执行时机

  std::async()任务执行时机可在创建时传入额外参数指定，这个参数的类型是std::launch，有以下含义：

• std::launch::deferred: 函数调用被延迟到wait()或get()函数调用时才执行

• std::launch::async: 函数必须在其 所在的独立线程上执行

• std::launch::deferred | std::launch::async: 表明实现可以选择这两种方式的一种

   

    

     

      

     
     

      

       auto f6=std::async(std::launch::async,Y(),1.2); // 在新线程上执行
      
     
    

     

      

     
     

      

       auto f7=std::async(std::launch::deferred, func, std::ref(x)); // 在wait()或get()调用时执行
      
     
    

     

      

     
     

      

       auto f8=std::async(
      
     
    

     

      

     
     

      

               std::launch::deferred | std::launch::async,
      
     
    

     

      

     
     

      

               func, std::ref(x)); // 实现选择执行方式
      
     
    

     

      

     
     

      

       auto f9=std::async(func, std::ref(x));
      
     
    

     

      

     
     

      

       f7.wait(); // 调用延迟函数
      
     
   
  

「注」：
  std::async与std::thread作用相似，两者最明显的区别在于async采用默认启动策略时并不一定创建新的线程。
  如果系统资源紧张，std::thread创建线程可能失败，系统报告异常。而std::async则不会，它在无法创建新线程时，会将任务分配给后续调用future.get()函数的线程，并以同步的方式执行（即不创建新线程）。

std::async避坑

接下说的很重要!!!
  在std::async时，一定要处理其返回的std::future。否则，async会以同步的方式执行任务，即当前线程会阻塞到任务执行完。

「错误的示例」：

   

    

     

      

     
     

      

       void async_case()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           struct SThreadParam
      
     
    

     

      

     
     

      

           {
      
     
    

     

      

     
     

      

               int func(int val1, int val2)
      
     
    

     

      

     
     

      

               {
      
     
    

     

      

     
     

      

                   LOGD("async func start...\n"); // a
      
     
    

     

      

     
     

      

                   sleep(2);           // Simulated 2S time consumption
      
     
    

     

      

     
     

      

                   return val1 + val2;
      
     
    

     

      

     
     

      

               }
      
     
    

     

      

     
     

      

           };
      
     
    

     

      

     
     

      

           SThreadParam paramFunc;
      
     
    

     

      

     
     

      

           LOGD("create async...\n"); // b
      
     
    

     

      

     
     

      

           //std::future<int> f4 = std::async(&SThreadParam::func, &paramFunc, 10, 12);
      
     
    

     

      

     
     

      

           std::async(std::launch::async, &SThreadParam::func, &paramFunc, 10, 12);
      
     
    

     

      

     
     

      

           LOGD("async debug...\n"); // c
      
     
    

     

      

     
     

      

           //sleep(2);
      
     
    

     

      

     
     

      

           //LOGD("async return %d.\n", f4.get());
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  
  

「执行结果」：

   

    

     

      

     
     

      

       2022-08-27 00:55:15.886  179 D: create async...
      
     
    

     

      

     
     

      

       2022-08-27 00:55:15.886  173 D: async func start...
      
     
    

     

      

     
     

      

       2022-08-27 00:55:17.887  182 D: async debug...
      
     
   
  

  设计上我们希望async不会阻塞，即SThreadParam::func应该在另一个线程执行，b和c的打印时间应该接近相同。但实际上却相差2s，刚好是SThreadParam::func的执行时间。

  这是很隐蔽且很重大的bug，假设任务时间更长，会导致主线长时间阻塞。这个bug网上大佬们已经给出很多解释了，这里不再赘诉，可参考https://blog.csdn.net/weixin_44537992/article/details/108438379。

  这就警告我们在使用新的接口时，还是要搞清楚其原型及原理，规范地使用未知的接口。说到这里我也慢慢懂了，为什么有些接口明明部分参数可以缺省，但是大佬们在使用时还是会明确指定这些参数值，虽然使用起来更复杂，但是也更安全。

std::packaged_task 任务与期望

  std::packaged_task<>对一个函数或可调用对象， 绑定一个期望。当std::packaged_task<>对象被调用，它就会调用相关函数或可调用对象，将期望状态置为就绪，返回值也会被存储为相关数据。

  与std::async不同的是，std::packaged_task<>可以将任务与期望打包，移动到指定线程显示调用packaged_task，future才会就绪；而std::async()不能指定线程运行任务。这是两者最显著的差异。

「packaged_task实例」：

   

    

     

      

     
     

      

       void task_execute_thread()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           std::thread th5([](const char *name) {
      
     
    

     

      

     
     

      

               static std::packaged_task<int(int)> task;
      
     
    

     

      

     
     

      

               
      
     
    

     

      

     
     

      

               while(!packaged_shutdown) {
      
     
    

     

      

     
     

      

                   std::unique_lock<std::mutex> lk(package_mut);
      
     
    

     

      

     
     

      

                   LOG("%s: wait task...\n", name);
      
     
    

     

      

     
     

      

                   pacakge_cond.wait(lk, [](){
      
     
    

     

      

     
     

      

                       LOG("check queue %ld.\n", task_queue.size());
      
     
    

     

      

     
     

      

                       return !task_queue.empty();
      
     
    

     

      

     
     

      

                   });
      
     
    

     

      

     
     

      

                   lk.unlock();
      
     
    

     

      

     
     

      

                   LOG("%s: task execute.\n", name);
      
     
    

     

      

     
     

      

                   task = std::move(task_queue.front()); // 5
      
     
    

     

      

     
     

      

                   task_queue.pop_front();
      
     
    

     

      

     
     

      

                   task(2); // a. 执行任务，参数未使用
      
     
    

     

      

     
     

      

               }
      
     
    

     

      

     
     

      

           }, "thread5");
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

           th5.detach();
      
     
    

     

      

     
     

      

       }
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

       void task_post()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           static int i = 0;
      
     
    

     

      

     
     

      

           auto f = [](int num){
      
     
    

     

      

     
     

      

               return num;
      
     
    

     

      

     
     

      

           };
      
     
    

     

      

     
     

      

           std::packaged_task<int(int)> task(std::bind(f, ++i));
      
     
    

     

      

     
     

      

           std::future<int> fut = task.get_future();
      
     
    

     

      

     
     

      

           std::unique_lock<std::mutex> lk(package_mut);
      
     
    

     

      

     
     

      

           
      
     
    

     

      

     
     

      

           task_queue.push_back(std::move(task));
      
     
    

     

      

     
     

      

           lk.unlock();
      
     
    

     

      

     
     

      

           pacakge_cond.notify_one();
      
     
    

     

      

     
     

      

           LOG("task return %d.\n", fut.get()); // b. 阻塞至a执行完成
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  
  

「执行结果」：

   

    

     

      

     
     

      

       input: f
      
     
    

     

      

     
     

      

       thread5: wait task...
      
     
    

     

      

     
     

      

       check queue 0.
      
     
    

     

      

     
     

      

       Input: g
      
     
    

     

      

     
     

      

       check queue 1.
      
     
    

     

      

     
     

      

       thread5: task execute 0.
      
     
    

     

      

     
     

      

       thread5: wait task...
      
     
    

     

      

     
     

      

       check queue 0.
      
     
    

     

      

     
     

      

       task return 1.
      
     
   
  

「注」 std::future只适用于一次期望。即获取的future只允许使用一次get。std::shared_future似乎可以解决这个问题，未确认。

std::promises 承诺与期望

std::promise对象可以保存某一类型 T 的值，该值可被 future 对象读取（可能在另外一个线程中），因此 promise 也提供了一种线程同步的手段。在 promise 对象构造时可以和一个共享状态（通常是std::future）相关联，并可以在相关联的共享状态(std::future)上保存一个类型为 T 的值。

  与std::packaged_task差异在于，std::promise可在线程运行时，通过set_value()向阻塞线程传递值。即A线程通过std::promise::set_value()传递某类型T的值，B线程std::future::get()会解除阻塞，同时获取到A线程传的T类型的值。

「std::promise实例」：

   

    

     

      

     
     

      

       void pomise_case()
      
     
    

     

      

     
     

      

       {
      
     
    

     

      

     
     

      

           std::promise<int> promise;  // save type is int
      
     
    

     

      

     
     

      

           std::future<int> future = promise.get_future(); // from promise
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

           LOGD("-> create thread6.\n");
      
     
    

     

      

     
     

      

           std::thread th6([](const char *name, std::promise<int> &prm) {
      
     
    

     

      

     
     

      

               static int value = 1024;
      
     
    

     

      

     
     

      

               std::this_thread::sleep_for(std::chrono::seconds(2)); // To highlight the delay of the effect
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

               prm.set_value(++value);  // a. refer step b
      
     
    

     

      

     
     

      

               LOGD("%s: set value: %d.\n", name, value);
      
     
    

     

      

     
     

      

           }, "thread6", std::ref(promise));
      
     
    

     

      

     
     

      

           
      
     
    

     

      

     
     

      

           LOGD("-> create thread7.\n"); 
      
     
    

     

      

     
     

      

           std::thread th7([](const char *name, std::future<int> &fut) {
      
     
    

     

      

     
     

      

               auto value = fut.get(); // b. wait step a set_value
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

               LOGD("%s: get value: %d.\n", name, value);
      
     
    

     

      

     
     

      

           }, "thread7", std::ref(future));
      
     
    

     

      

     
     

      
 
      
     
    

     

      

     
     

      

           th6.join();
      
     
    

     

      

     
     

      

           th7.join();
      
     
    

     

      

     
     

      

       }
      
     
   
  

   
  
  

「执行效果」：

   

    

     

      

     
     

      

       2022-08-27 02:24:08.826  233 D: -> create thread6.
      
     
    

     

      

     
     

      

       2022-08-27 02:24:08.826  242 D: -> create thread7.
      
     
    

     

      

     
     

      

       2022-08-27 02:24:10.827  239 D: thread6: set value: 1025.
      
     
    

     

      

     
     

      

       2022-08-27 02:24:10.827  246 D: thread7: get value: 1025.
      
     
   
  

std::promise还是挺有意思的，可以实现线程间的值传递且无并发问题。原来我们通过全局变量实现线程间通信的方法，还要考虑上锁，以后就可以通过promise来实现了。

总结

• 本文主要总结了同步并发常用的一些技巧，其中条件变量使用被广泛使用，而c++11新引进的std::future相对来说很少被用到。

• 大致概括一下std::async、std::packaged_task、std::promise使用的场景：
  std::async: 另起线程异步处理耗时的任务，并返回处理结果。std::packaged_task: 将任务函数打包丢到指定线程处理，并返回处理结果。std::promise: 实现两个线程间的数据传递。

• 刚开始看std::async、std::packaged_task、std::promise，多少有一点抵触。一个同步搞这么多接口，仔细慢看发现每一种都使用都是针对不同的场景。

• std::async在使用时一定主要处理其返回的std::future，否则会引发隐蔽且严重的bug。详见「std::async避坑章节」。

• 针对std::async可能引发的问题，告诫程序员们，一定要规范的使用每一个接口，缺省的参数及返回值不处理的话可能让代码裂成“东非大裂谷”。对于使用到的接口的参数和返回值，不管有没有必要，都要处理。

• 同步并发问题一直是程序中隐蔽而又严重的bug，不到万不得已尽量不要使用。同步并发bug就像是程序里的"蒙娜丽莎"，它在对你笑，你却一脸懵逼。

最后

用心感悟，认真记录，写好每一篇文章，分享每一框干货。

一键四连