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

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

? 条件变量是C++11提供的另外一种用于等待的同步机制，它能阻塞一个或多个线程，直到收到另外一个线程发出的通知或者超时时，才会唤醒当前阻塞的线程。条件变量需要和互斥量配合起来使用，C++11提供了两种条件变量：

• condition_variable：需要配合std::unique_lock<std::mutex>进行wait操作，也就是阻塞线程的操作。
• condition_variable_any：可以和任意带有lock()、unlock()语义的mutex搭配使用，也就是说有四种：
  1. std::mutex：独占的非递归互斥锁
  2. std::timed_mutex：带超时的独占非递归互斥锁
  3. std::recursive_mutex：不带超时功能的递归互斥锁
  4. std::recursive_timed_mutex：带超时的递归互斥锁

条件变量通常用于生产者和消费者模型，大致使用过程如下：

1. 拥有条件变量的线程获取互斥量
2. 循环检查某个条件，如果条件不满足阻塞当前线程，否则线程继续向下执行
   • 产品的数量达到上限，生产者阻塞，否则生产者一直生产。。。
   • 产品的数量为零，消费者阻塞，否则消费者一直消费。。。
3. 条件满足之后，可以调用notify_one()或者notify_all()唤醒一个或者所有被阻塞的线程
   • 由消费者唤醒被阻塞的生产者，生产者解除阻塞继续生产。。。
   • 由生产者唤醒被阻塞的消费者，消费者解除阻塞继续消费。。。

1.condition_variable

1.1成员函数

condition_variable的成员函数主要分为两部分：线程等待（阻塞）函数 和线程通知（唤醒）函数，这些函数被定义于头文件 <condition_variable>。

等待函数

wait() wait_for() wait_until()

调用wait()函数的线程会被阻塞

// ①
void wait (unique_lock<mutex>& lck);
// ②
template <class Predicate>
void wait (unique_lock<mutex>& lck, Predicate pred);

• 函数①：调用该函数的线程直接被阻塞

• 函数②：该函数的第二个参数是一个判断条件，是一个返回值为布尔类型的函数

  • 该参数可以传递一个有名函数的地址，也可以直接指定一个匿名函数

  • 表达式返回false当前线程被阻塞，表达式返回true当前线程不会被阻塞，继续向下执行

• 独占的互斥锁对象不能直接传递给wait()函数，需要通过模板类unique_lock进行二次处理，通过得到的对象仍然可以对独占的互斥锁对象做如下操作，使用起来更灵活。

• 如果线程被该函数阻塞，这个线程会释放占有的互斥锁的所有权，当阻塞解除之后这个线程会重新得到互斥锁的所有权，继续向下执行（这个过程是在函数内部完成的，了解这个过程即可，其目的是为了避免线程的死锁）。

wait_for()函数和wait()的功能是一样的，只不过多了一个阻塞时长，假设阻塞的线程没有被其他线程唤醒，当阻塞时长用完之后，线程就会自动解除阻塞，继续向下执行。

template <class Rep, class Period>
cv_status wait_for (unique_lock<mutex>& lck,
                    const chrono::duration<Rep,Period>& rel_time);
	
template <class Rep, class Period, class Predicate>
bool wait_for(unique_lock<mutex>& lck,
               const chrono::duration<Rep,Period>& rel_time, Predicate pred);

wait_until()函数和wait_for()的功能是一样的，它是指定让线程阻塞到某一个时间点，假设阻塞的线程没有被其他线程唤醒，当到达指定的时间点之后，线程就会自动解除阻塞，继续向下执行。

template <class Clock, class Duration>
cv_status wait_until (unique_lock<mutex>& lck,
                      const chrono::time_point<Clock,Duration>& abs_time);

template <class Clock, class Duration, class Predicate>
bool wait_until (unique_lock<mutex>& lck,
                 const chrono::time_point<Clock,Duration>& abs_time, Predicate pred);

通知函数

void notify_one() noexcept;
void notify_all() noexcept;

• notify_one()：唤醒一个被当前条件变量阻塞的线程
• notify_all()：唤醒全部被当前条件变量阻塞的线程

1.2生产者和消费者模型

#include <iostream>
using namespace std;
#include <condition_variable>
#include <thread>
#include <queue>
#include <mutex>
#include <chrono>

// 锁对象
mutex bufferMutex;
// 条件变量
condition_variable bufferNotFull, bufferNotEmpty;
// 队列
queue<int> buffer;
// 缓冲区大小
const int bufferSize = 5;

void producer(int id) {
    for (int i = 0; i < 10; ++i) {
        unique_lock<mutex> lock(bufferMutex);
        // 检查缓冲区是否已满，如果满则等待
        bufferNotFull.wait(lock, []() { return buffer.size() < bufferSize; });

        // 生产数据并放入缓冲区
        int data = i;
        buffer.push(data);
        cout << "Producer " << id << " produced data: " << data << endl;

        // 通知消费者线程缓冲区非空
        bufferNotEmpty.notify_one();
        lock.unlock();
        this_thread::sleep_for(chrono::milliseconds(10));
    }
}

void consumer(int id) {
    for (int i = 0; i < 10; i++) {
        unique_lock<mutex> lock(bufferMutex);
        // 检查缓冲区是否为空，如果空则等待
        bufferNotEmpty.wait(lock, []() { return buffer.size() > 0; });
        int data = buffer.front();
        buffer.pop();
        cout << "Consumer " << id << " consumed data: " << data << endl;

        // 通知生产者缓冲区不满
        bufferNotFull.notify_one();
        lock.unlock();
        this_thread::sleep_for(chrono::milliseconds(10));
    }

}

int main() {
    thread producerThread1(producer, 1);
    thread producerThread2(producer, 2);
    thread consumerThread1(consumer, 1);
    thread consumerThread2(consumer, 2);

    producerThread1.join();
    producerThread2.join();
    consumerThread1.join();
    consumerThread2.join();
    return 0;
}

程序输出结果:

Producer 2 produced data: 0
Producer 1 produced data: 0
Consumer 1 consumed data: 0
Consumer 2 consumed data: 0
Producer 2 produced data: 1
Consumer 2 consumed data: 1
Producer 1 produced data: 1
Consumer 1 consumed data: 1
Producer 1 produced data: 2
Consumer 2 consumed data: 2
Producer 2 produced data: 2
Consumer 1 consumed data: 2
Producer 1 produced data: 3
Consumer 2 consumed data: 3
Producer 2 produced data: 3
Consumer 1 consumed data: 3
Producer 1 produced data: 4
Consumer 2 consumed data: 4
Producer 2 produced data: 4
Consumer 1 consumed data: 4
Producer 1 produced data: 5
Consumer 1 consumed data: 5
Producer 2 produced data: 5
Consumer 2 consumed data: 5
Producer 2 produced data: 6
Producer 1 produced data: 6
Consumer 2 consumed data: 6
Consumer 1 consumed data: 6
Producer 2 produced data: 7
Consumer 2 consumed data: 7
Producer 1 produced data: 7
Consumer 1 consumed data: 7
Producer 1 produced data: 8
Producer 2 produced data: 8
Consumer 2 consumed data: 8
Consumer 1 consumed data: 8
Producer 1 produced data: 9
Producer 2 produced data: 9
Consumer 1 consumed data: 9
Consumer 2 consumed data: 9

在这个例子中：

• bufferMutex 是一个互斥量，用于保护对缓冲区的并发访问。
• bufferNotEmpty 和 bufferNotFull 是条件变量，用于在缓冲区非空和非满的情况下进行线程同步。
• 生产者线程使用 bufferNotFull.wait 来等待缓冲区非满，消费者线程使用 bufferNotEmpty.wait 来等待缓冲区非空。
• 在生产者放入数据后，会通过 bufferNotEmpty.notify_one() 通知等待的消费者线程。
• 在消费者取出数据后，会通过 bufferNotFull.notify_one() 通知等待的生产者线程。

这个模型使用了条件变量来实现生产者和消费者线程之间的同步和通信，确保生产者在缓冲区非满时生产数据，而消费者在缓冲区非空时消费数据。这有助于避免竞态条件和提高程序的效率。

2.condition_variable_any

2.1成员函数

condition_variable_any的成员函数也是分为两部分：线程等待（阻塞）函数 和线程通知（唤醒）函数，这些函数被定义于头文件 <condition_variable>。

等待函数

// ①
template <class Lock> void wait (Lock& lck);
// ②
template <class Lock, class Predicate>
void wait (Lock& lck, Predicate pred);

• 函数①：调用该函数的线程直接被阻塞
• 函数②：该函数的第二个参数是一个判断条件，是一个返回值为布尔类型的函数
  • 该参数可以传递一个有名函数的地址，也可以直接指定一个匿名函数
  • 表达式返回false当前线程被阻塞，表达式返回true当前线程不会被阻塞，继续向下执行
• 可以直接传递给wait()函数的互斥锁类型有四种，分别是：
  • std::mutex、std::timed_mutex、std::recursive_mutex、std::recursive_timed_mutex
• 如果线程被该函数阻塞，这个线程会释放占有的互斥锁的所有权，当阻塞解除之后这个线程会重新得到互斥锁的所有权，继续向下执行（这个过程是在函数内部完成的，了解这个过程即可，其目的是为了避免线程的死锁）。

wait_for()函数和wait()的功能是一样的，只不过多了一个阻塞时长，假设阻塞的线程没有被其他线程唤醒，当阻塞时长用完之后，线程就会自动解除阻塞，继续向下执行。

template <class Lock, class Rep, class Period>
cv_status wait_for (Lock& lck, const chrono::duration<Rep,Period>& rel_time);
	
template <class Lock, class Rep, class Period, class Predicate>
bool wait_for (Lock& lck, const chrono::duration<Rep,Period>& rel_time, Predicate pred);

wait_until()函数和wait_for()的功能是一样的，它是指定让线程阻塞到某一个时间点，假设阻塞的线程没有被其他线程唤醒，当到达指定的时间点之后，线程就会自动解除阻塞，继续向下执行。

template <class Lock, class Clock, class Duration>
cv_status wait_until (Lock& lck, const chrono::time_point<Clock,Duration>& abs_time);

template <class Lock, class Clock, class Duration, class Predicate>
bool wait_until (Lock& lck, 
                 const chrono::time_point<Clock,Duration>& abs_time, 
                 Predicate pred);

通知函数

void notify_one() noexcept;
void notify_all() noexcept;

• notify_one()：唤醒一个被当前条件变量阻塞的线程
• notify_all()：唤醒全部被当前条件变量阻塞的线程

2.2生产者和消费者模型

只需要改一下代码即可

#include <iostream>
using namespace std;
#include <condition_variable>
#include <thread>
#include <queue>
#include <mutex>
#include <chrono>

// 锁对象
mutex bufferMutex;
// 条件变量
condition_variable_any bufferNotFull, bufferNotEmpty;
// 队列
queue<int> buffer;
// 缓冲区大小
const int bufferSize = 5;

void producer(int id) {
    for (int i = 0; i < 10; ++i) {
        // unique_lock<mutex> lock(bufferMutex);
        lock_guard<mutex> lock(bufferMutex);
        // 检查缓冲区是否已满，如果满则等待
        bufferNotFull.wait(bufferMutex, []() { return buffer.size() < bufferSize; });

        // 生产数据并放入缓冲区
        int data = i;
        buffer.push(data);
        cout << "Producer " << id << " produced data: " << data << endl;

        // 通知消费者线程缓冲区非空
        bufferNotEmpty.notify_one();
        // lock.unlock();
    }
}

void consumer(int id) {
    for (int i = 0; i < 10; i++) {
        // unique_lock<mutex> lock(bufferMutex);
        lock_guard<mutex> lock(bufferMutex);
        // 检查缓冲区是否为空，如果空则等待
        bufferNotEmpty.wait(bufferMutex, []() { return buffer.size() > 0; });
        int data = buffer.front();
        buffer.pop();
        cout << "Consumer " << id << " consumed data: " << data << endl;

        // 通知生产者缓冲区不满
        bufferNotFull.notify_one();
        // lock.unlock();
    }

}

int main() {
    thread producerThread1(producer, 1);
    thread producerThread2(producer, 2);
    thread consumerThread1(consumer, 1);
    thread consumerThread2(consumer, 2);

    producerThread1.join();
    producerThread2.join();
    consumerThread1.join();
    consumerThread2.join();
    return 0;
}

程序运行结果:

Producer 1 produced data: 0
Producer 1 produced data: 1
Producer 1 produced data: 2
Producer 1 produced data: 3
Producer 1 produced data: 4
Consumer 1 consumed data: 0
Consumer 1 consumed data: 1
Consumer 1 consumed data: 2
Consumer 1 consumed data: 3
Consumer 1 consumed data: 4
Producer 1 produced data: 5
Producer 1 produced data: 6
Producer 1 produced data: 7
Producer 1 produced data: 8
Producer 1 produced data: 9
Consumer 2 consumed data: 5
Consumer 2 consumed data: 6
Consumer 2 consumed data: 7
Consumer 2 consumed data: 8
Consumer 2 consumed data: 9
Producer 2 produced data: 0
Producer 2 produced data: 1
Producer 2 produced data: 2
Producer 2 produced data: 3
Producer 2 produced data: 4
Consumer 1 consumed data: 0
Consumer 1 consumed data: 1
Consumer 1 consumed data: 2
Consumer 1 consumed data: 3
Consumer 1 consumed data: 4
Producer 2 produced data: 5
Producer 2 produced data: 6
Producer 2 produced data: 7
Producer 2 produced data: 8
Producer 2 produced data: 9
Consumer 2 consumed data: 5
Consumer 2 consumed data: 6
Consumer 2 consumed data: 7
Consumer 2 consumed data: 8
Consumer 2 consumed data: 9