C++并发编程的同步介绍

condition_variable：同步

上面的互斥锁只是在共享数据处执行保护操作，但是数据的同步，即线程对数据的操作的先后次序并不确定，当我们还想对线程同步时，必须采取一定的同步操作。条件变量是达到这个目的方法。

C++标准库对条件变量有两套实现：

• std::condition_variable和 std::condition_variable_any 。这两个实现都包含在<condition_variable>头文件的声明中。两者都需要与一个互斥量一起才能工作(互斥量是 为了同步)

-   前者仅限于与`std::mutex`一起工作，
-   而后者可以和任何满足最低标准的互斥量一起工作，从而加上了`_any`的后缀，因此从体积、性能，以及系统资源的使用方面产生额外的开销.
-   所以 `std::condition_variable` 一般作为首选的类型，当对灵活性有硬性要求时，我们才会去考虑 `std::condition_variable_any`

在上面的例子中，10 各线程被同时唤醒，因此打印的时候是乱序的。值得注意的是 while(!ready)，实际上，正常情况下，cv.wait 只会被调用一次，然后等待唤醒，因为线程在调用 wait() 之后就被阻塞了。但是通过一个 while 循环来判断全局标志位是否正确，这样可以防止被误唤醒，这也是条件变量中的常见写法。

接口概览

构造函数

std::condition_variable 的拷贝构造函数被禁用，只提供了默认构造函数。

wait操作

std::condition_variable 提供了两种 wait() 函数，一个是不带条件的，一个是可传入条件，通常为lambda表达式

//无条件等待
void wait (unique_lock<mutex>& lck);
//有条件等待
template <class Predicate>
void wait (unique_lock<mutex>& lck, Predicate pred);

• 当线程调用wait (unique_lock<mutex>& lck)时，当前线程会阻塞并释放当前获得的锁lck,以提醒其他线程可以获得这个自由锁了。因此对于wait (unique_lock<mutex>& lck)只要一调用就会阻塞，那么在外部必须给它价格条件判断，判断线程是否执行wait
• 而wait (unique_lock<mutex>& lck, Predicate pred)是当pred返回false时线程会阻塞，即其自带了条件判断，我们只需传入即可。
• 另外，当阻塞在wait的线程被唤醒时，会再次获得相应的锁。

注意wait()函数一定要搭配unique_lock类模板使用，而不是lock_guard。这是因为lock_guard在线程调用wait阻塞时，不会自动释放当前线程所获的的锁，这样就会导致死锁的发生。unique_lock`是一个灵活性的锁机制

和mutex 的 lock类似，std::condition_variable 也提供了相应的两种（带 Predicate 和不带Predicate） wait_for() 函数，与 std::condition_variable::wait() 类似，不过 wait_for 可以指定一个时间段，在当前线程收到通知或者指定的时间超时之前，该线程都会处于阻塞状态。而一旦超时或者收到了其他线程的通知，wait_for 返回，剩下的处理步骤和 wait() 类似。还有 wait_util()，用法也类似

notify 操作

• std::condition_variable::notify_one()

唤醒某个等待（`wait`）线程。如果当前没有等待线程，则该函数什么也不做，如果同时存在多个等待线程，则唤醒某个线程是不确定的（`unspecified`）。

• std::condition_variable::notify_all() 唤醒所有的等待（wait）线程。如果当前没有等待线程，则该函数什么也不做。

示例

std::mutex mut;
std::queue<data_chunk> data_queue; // 1
std::condition_variable data_cond;
void data_preparation_thread()
{
	while(more_data_to_prepare())
	{
		data_chunk const data=prepare_data();
		std::lock_guard<std::mutex> lk(mut);
		data_queue.push(data); // 2
		data_cond.notify_one(); // 3
	}
}
void data_processing_thread()
{
	while(true)
	{
		std::unique_lock<std::mutex> lk(mut); // 4
		data_cond.wait(
			lk,[]{return !data_queue.empty();}); // 5
		data_chunk data=data_queue.front();
		data_queue.pop();
		lk.unlock(); // 6
		process(data);
		if(is_last_chunk(data))
			break;
	}
}

代码解释

1. std::queue<data_chunk> data_queue;：定义一个队列用于存放生产者准备好的数据块。
2. data_queue.push(data);：将生产者准备好的数据块放入队列中。
3. data_cond.notify_one();：通过条件变量通知消费者队列中有数据可用。
4. std::unique_lock<std::mutex> lk(mut);：定义一个独占互斥锁，保证在修改队列时线程安全。
5. data_cond.wait(lk,[]{return !data_queue.empty();});：消费者线程等待条件变量，等待生产者线程通知有数据可用，同时检查队列是否为空。如果队列不为空，则唤醒消费者线程继续处理数据。
6. lk.unlock();：释放独占互斥锁，确保其他线程可以访问队列。

整个过程中，生产者和消费者通过条件变量和互斥锁来保证线程同步和线程安全。生产者通过条件变量通知消费者队列中有数据可用，消费者通过条件变量等待生产者通知并检查队列是否为空，从而避免了忙等待，节省了资源。同时，互斥锁确保了生产者和消费者对队列的操作是线程安全的，避免了数据竞争和死锁的发生。

 condition_variable_any 介绍

与 std::condition_variable 类似，只不过 std::condition_variable_any 的 wait 函数可以接受任何 lockable 参数，而 std::condition_variable 只能接受 std::unique_lock<std::mutex> 类型的参数，除此以外，和 std::condition_variable 几乎完全一样。

生成者消费者模型

一般来说，生产者消费者模型可以通过 queue， mutex 和 condition_variable 来实现。下面是一个简单实现：

#include <iostream>
#include <mutex>
#include <queue>
#include <condition_cariable>
#include <chrono>
#include <thread>
#include <atomic>

int main()
{
  std::queue<int> production;
  std::mutex mtx;
  std::condition_variable cv;
  bool ready = false;  // 是否有产品可供消费
  bool done = false;   // 生产结束

  std::thread producer(
    [&] () -> void {
      for (int i = 1; i < 10; ++i)
      {
        // 模拟实际生产过程
        std::this_thread ::sleep_for(std::chrono::milliseconds(10));
        std::cout << "producing " << i << std::endl;

        std::unique_lock<std::mutex> lock(mtx);
        production.push(i);

        // 有产品可以消费了
        ready = true;
        cv.notify_one();
      }
      // 生产结束了
      done = true;
    }
  );

  std::thread consumer(
    [&] () -> void {
      std::unique_lock<std::mutex> lock(mtx);
      // 如果生成没有结束或者队列中还有产品没有消费，则继续消费，否则结束消费
      while(!done || !production.empty())
      {
        // 防止误唤醒
        while(!ready)
        {
          cv.wait(lock);
        }

        while(!production.empty())
        {
          // 模拟消费过程
          std::cout << "consuming " << production.front() << std::endl;
          production.pop();
        }

        // 没有产品了
        ready = false;
      }
    }
  );

  producer.join();
  consumer.join();

  return 0;
}

上面的这段代码使用了 C++11 中的标准库 mutex、condition_variable 和 atomic 等工具来实现线程同步和互斥。

主函数中首先定义了一个 std::queue<int> 对象 production，表示生产者生产的产品队列，同时定义了一个 std::mutex 对象 mtx，表示生产者和消费者之间的互斥锁，以及一个 std::condition_variable 对象 cv，表示生产者和消费者之间的条件变量，用于线程间的同步。

接下来定义了两个 bool 类型的变量 ready 和 done，分别表示是否有产品可供消费和生产是否结束。ready 变量在生产者线程中被设置为 true，表示生产者已经将产品放入了队列中，可以供消费者消费了。而 done 变量则在生产者线程中被设置为 true，表示生产者已经生产完毕，队列中已经没有产品了。

然后创建了两个线程：生产者线程和消费者线程。生产者线程使用 lambda 表达式定义，实现了一个简单的生产过程，每次生产一个数字并将其放入队列中。在生产完成后，将 done 变量设置为 true，表示生产结束。同时，将 ready 变量设置为 true，并使用 cv.notify_one() 通知消费者线程有产品可供消费了。

消费者线程同样使用 lambda 表达式定义，实现了一个简单的消费过程。在消费过程中，首先使用 std::unique_lockstd::mutex lock(mtx) 获得了互斥锁 mtx，并使用 while 循环判断是否有产品可以消费。在 while 循环中，首先调用 cv.wait(lock) 等待生产者线程的通知，直到有产品可供消费为止。然后，使用 while 循环从队列中取出产品进行消费，并在消费完成后将 ready 变量设置为 false，表示队列中没有产品可供消费了。

最后，使用 producer.join() 和 consumer.join() 等待生产者和消费者线程完成，然后返回 0 表示程序运行正常结束。

我正在参与2023腾讯技术创作特训营第三期有奖征文，组队打卡瓜分大奖！