C++线程同步之条件变量

C++线程同步之条件变量

什么是条件变量（Condition Variable）？

条件变量（condition_variable）是一种线程同步机制，用于在多线程程序中协调线程之间的执行顺序。它通过允许线程在某些条件成立时被唤醒，或者当某些条件不满足时进入等待状态，从而实现线程的协调和同步。与互斥锁（mutex）配合使用，条件变量允许线程在某个条件发生变化时继续执行。

条件变量的主要用途

条件变量的核心作用是：

1. 线程等待某个条件：当一个线程需要等待某个特定条件发生时，它可以通过条件变量进入等待状态。比如等待队列中有元素时才能进行消费操作。
2. 通知其他线程：当条件发生变化时，其他线程可以通知正在等待的线程继续执行。比如生产者线程生产了新的数据，通知消费者线程可以开始处理数据。

在并发编程中，条件变量通常与互斥锁一起使用，来确保数据访问的一致性。

常见的应用场景

1. 生产者-消费者问题：多个生产者线程和多个消费者线程共享一个缓冲区，生产者将数据放入缓冲区，消费者从缓冲区取数据。在缓冲区为空时，消费者线程等待；在缓冲区已满时，生产者线程等待。条件变量可以用来协调生产者和消费者的执行顺序。
2. 线程池：当线程池中的线程正在处理任务时，如果没有任务，线程会进入等待状态，直到任务被提交，线程才会被唤醒并执行任务。
3. 线程之间的事件通知：比如一个线程执行一些初始化操作，另一个线程依赖于该操作完成后才能继续执行。通过条件变量，一个线程可以在另一个线程完成初始化后收到通知。
4. 共享资源的访问：多个线程需要访问一个共享资源，在某些情况下资源可能不可用（比如池中没有空闲资源）。线程可以等待资源可用并在条件满足时继续执行。

C++11中的条件变量

在C++11中，标准库引入了条件变量类，提供了两种主要的条件变量：

1. std::condition_variable：这是最常用的条件变量，适用于需要在多个线程间协调的场景。它基于std::mutex或std::unique_lock<std::mutex>来进行线程同步。
2. std::condition_variable_any：这是一个通用的条件变量，它可以与任意类型的锁一起使用（不仅限于std::mutex），但它的使用场景相对较少。

condition_variable的使用方法

std::condition_variable的使用步骤

1. 创建一个条件变量。
2. 使用std::mutex或std::unique_lock来保护共享数据。
3. 使用wait函数使线程进入等待状态，直到接收到通知（notify_one或notify_all）。
4. 使用notify_one或notify_all通知一个或所有等待的线程。

典型的使用示例：生产者-消费者模型

#include <cstddef>
#include <cstdio>
#include <cstdlib>
#include <iostream>
#include <queue>
#include <thread>
#include <condition_variable>
#include <mutex>using namespace  std;std::queue<int> buffer;
std::mutex mtx;
condition_variable m_QueueEmpty;   // 队列为空的条件变量
condition_variable m_QueueFull;    // 队列已满的条件变量const size_t max_buffer_size = 5;void producer() {std::unique_lock<std::mutex> lock(mtx);// 队列已满while (buffer.size() >= max_buffer_size) {cout << "buffer is full, producer wait..." << endl;m_QueueFull.wait(lock);}// 生产数据 模拟生产数据 随机获取1~100的随机数int data = rand() % 100 + 1;buffer.push(data);auto threadid = this_thread::get_id();printf("produce data: %d , thread id: %lld \n",data, threadid);// 通知消费者可以消费数据m_QueueEmpty.notify_one();
}void consumer() {unique_lock<mutex> lock(mtx);while (buffer.empty()) {cout << "buffer is empty, consumer wait..." << endl;m_QueueEmpty.wait(lock);}int data = buffer.front();buffer.pop();auto threadid = this_thread::get_id();printf("consume data: %d , thread id: %lld \n",data, threadid);// 通知生产者可以生产数据m_QueueFull.notify_one();
}int main() {thread t1[15];thread t2[15];for (int i = 0; i < 15; ++i){t1[i] = thread(producer);t2[i] = thread(consumer);}for (int i = 0; i < 15; ++i){t1[i].join();t2[i].join();}return 0;
}

条件变量condition_variable类的wait()还有一个重载的方法，可以接受一个条件，这个条件也可以是一个返回值为布尔类型的函数，条件变量会先检查判断这个条件是否满足，如果满足条件（布尔值为true），则当前线程重新获得互斥锁的所有权，结束阻塞，继续向下执行；如果不满足条件（布尔值为false），当前线程会释放互斥锁（解锁）同时被阻塞，等待被唤醒。

这样代码就可以优化一下:

void producer() {std::unique_lock<std::mutex> lock(mtx);// 队列已满m_QueueFull.wait(lock, []{ return buffer.size() < max_buffer_size; });// 生产数据 模拟生产数据 随机获取1~100的随机数int data = rand() % 100 + 1;buffer.push(data);auto threadid = this_thread::get_id();printf("produce data: %d , thread id: %lld \n",data, threadid);// 通知消费者可以消费数据m_QueueEmpty.notify_one();
}void consumer() {unique_lock<mutex> lock(mtx);m_QueueEmpty.wait(lock, []{ return !buffer.empty(); });int data = buffer.front();buffer.pop();auto threadid = this_thread::get_id();printf("consume data: %d , thread id: %lld \n",data, threadid);// 通知生产者可以生产数据m_QueueFull.notify_one();
}

修改之后可以发现，程序变得更加精简了，而且执行效率更高了，因为在这两个函数中的while循环被删掉了，但是最终的效果是一样的，推荐使用这种方式的wait()进行线程的阻塞。

condition_variable_any的使用方法

condition_variable_any 是 C++ 标准库中用于同步的一个条件变量，它允许线程在某些条件满足时继续执行。与标准的 std::condition_variable 相比，condition_variable_any 在底层实现上更加灵活，能够支持更多类型的锁，比如 std::mutex 和 std::shared_mutex，而 std::condition_variable 只能与 std::unique_lock<std::mutex> 配合使用。

std::condition_variable_any的使用步骤

std::condition_variable_any 的基本使用方式与 std::condition_variable 类似：

1. 创建一个 condition_variable_any 对象。
2. 在某些条件下等待（wait），当条件满足时，通知其他线程（notify_one 或 notify_all）。
3. 使用合适的锁来保护共享数据。

典型的使用示例：生产者-消费者模型

#include <iostream>
#include <thread>
#include <condition_variable>
#include <mutex>
#include <queue>std::mutex mtx;                          // 互斥锁
std::condition_variable_any cv;           // 条件变量
std::queue<int> buffer;                   // 模拟缓冲区
const int BUFFER_SIZE = 5;                 // 缓冲区最大容量// 生产者线程
void producer() {int product = 0;while (true) {std::this_thread::sleep_for(std::chrono::milliseconds(500)); // 模拟生产过程{std::lock_guard<std::mutex> lock(mtx);if (buffer.size() < BUFFER_SIZE) {buffer.push(product++);std::cout << "Produced: " << product << std::endl;cv.notify_one();  // 通知消费者有新的产品} else {std::cout << "Buffer is full. Producer is waiting." << std::endl;cv.wait(mtx, []{ return buffer.size() < BUFFER_SIZE; }); // 等待缓冲区有空间}}}
}// 消费者线程
void consumer() {while (true) {std::this_thread::sleep_for(std::chrono::milliseconds(1000)); // 模拟消费过程{std::lock_guard<std::mutex> lock(mtx);if (buffer.empty()) {std::cout << "Buffer is empty. Consumer is waiting." << std::endl;cv.wait(mtx, []{ return !buffer.empty(); });  // 等待缓冲区有数据} else {int product = buffer.front();buffer.pop();std::cout << "Consumed: " << product << std::endl;cv.notify_one();  // 通知生产者可以继续生产}}}
}int main() {std::thread producerThread(producer);std::thread consumerThread(consumer);producerThread.join();consumerThread.join();return 0;
}

1. 生产者线程：生产者线程首先会检查缓冲区是否有空间，如果缓冲区已满，生产者将等待，直到缓冲区有空间。使用 cv.wait(mtx, predicate) 来进行等待，其中 predicate 是一个 lambda 表达式，它会在缓冲区有空间时返回 true，通知线程继续执行。
2. 消费者线程：消费者线程首先会检查缓冲区是否为空。如果为空，消费者将等待，直到缓冲区有数据。使用同样的 cv.wait(mtx, predicate) 来进行等待。
3. 互斥锁 (mtx)：我们使用 std::mutex 来保证对缓冲区的互斥访问，确保线程安全。
4. 条件变量的通知：当生产者生产了一个新产品或消费者消费了一个产品时，会通过 cv.notify_one() 来通知另一个线程继续执行。

condition_variable_any vs std::condition_variable

• std::condition_variable_any 的主要优势在于，它不局限于与 std::mutex 一起使用，还可以与其他类型的锁配合使用，比如 std::shared_mutex。
• 在大多数情况下，std::condition_variable_any 和 std::condition_variable 用法类似。如果你不需要特别灵活的锁类型，std::condition_variable 更常见。

注意事项

1. 避免虚假唤醒：cv.wait() 会在被唤醒后重新检查条件，因此要使用 cv.wait(mtx, predicate) 这种带条件的等待方式，而不是简单地调用 cv.wait(mtx)，以防止虚假唤醒。
2. 使用 notify_one() 或 notify_all()：通常我们使用 notify_one() 来唤醒一个等待线程，或者使用 notify_all() 来唤醒所有等待线程。在生产者-消费者模型中，notify_one() 是常用的选择，因为每次只有一个线程被唤醒。

小结

• condition_variable_any 提供了一个灵活的方式来实现线程间的同步，尤其是对于更复杂的锁类型（如 std::shared_mutex）。
• 在生产者-消费者模型中，它有助于实现线程间的协调，使得生产者在缓冲区满时等待，消费者在缓冲区空时等待，通过适时的通知机制有效地协调生产和消费过程。