C++ 多线程学习(3) ---- 条件变量

雨点打透心脏的1/2处 2024-03-17 12:17 51阅读 0赞

### **1. 条件变量简介**  ###

在 C++ 11中，我们可以使用**条件变量（condition variable）**实现多个线程之间的同步操作，当条件不满足时，相关线程一直被阻塞，直到某种条件成立，这些线程才会被唤醒。

条件变量是利用线程间共享的全局变量进行同步的一种机制，主要包含两个动作：

*  一个线程因为等待条件变量的条件成立而挂起，
 *  另外一个线程使条件成立，给出信号，从而唤醒被等待的线程。

为了防止竞争，条件变量总是和一个互斥锁结合在一起，通常情况下这个锁是 std::mutex，并且管理这个锁的只能是 **std::unique\_lock<std::mutex> RAII** 的模板类。

**原子操作的概念：**

所谓原子操作是指**不会被线程调度机制打断的操作**；这种操作一旦开始，就一直运行到结束，中间不会有任何 context switch （切换到另一个线程）。

**临界资源：**

临界资源是一次执行过程仅仅允许一个进程使用的共享资源，各个进程采取互斥的方式实现共享，属于临界资源的硬件有 打印机，磁带机等，软件有消息队列，变量，数组，缓冲区等，各个进程采取互斥的方式实现对这种资源的共享。（可以理解为对资源的一次操作不能被打断，也就是一次操作过程需要完整，不能操作出现中间切走的情况）

**临界区：**

每个进程访问临界资源的那段代码称为临界区（criticalsection），每次只允许一个进程进去临界区，进去后，不允许其他进程进入。不论是硬件临界资源还是软件临界资源，多个进程必须互斥的对它进行访问，使用临界区时，一般不允许其运行时间过长，只要运行在临界区的线程还没有离开，其他所有进入此临界区的线程都会被挂起而进入等待状态，并在一定程度上影响程序的运行性能。

### **2. 条件变量的使用接口** ###

上面提到的两个步骤，分别用下面的方法实现：

等待条件成立使用的是 condition\_variable 类成员函数 **wait，wait\_for 和 wait\_unitl**，

给出信号的使用的是 condition\_variable 类成员函数 **notify\_one 和 notify\_all** 函数

![4ee63b444747432a9183fc81e3ecc02a.png][]

#### **2.1 wait/wait\_for 函数** ####

**wait** 导致当前线程阻塞直至条件变量被**通知**，或**虚假唤醒发生**，可选地循环**直至满足某谓词**。

**wait\_for** 函数导致当前的线程阻塞直到条件变量被通知，或者虚假唤醒发生，或者超时返回。

返回值说明：

1. 如果经过 rel\_time 指定的关联时限则为 **std:cv\_status:timeout**，否则为 **std:cv\_status:no\_timeout**

以上两个类型的 wait 函数都会在阻塞的时候，自动释放锁的权限，即调用 **unique\_lock** 的成员函数 unlock()，以便于其他线程能够有机会获得锁，这就是条件变量只能和 **unique\_**lock 一起使用的原因，否则线程一直占有锁，线程被阻塞。

#### **2.2 notify/notify\_one** ####

**notify/notify\_one** 函数声明如下：

**notify\_one**：任何线程在 **\*this** 上等待，则调用 notify\_one 会解阻塞（唤醒）等待的线程之一

**notify：**唤醒任何在 **\*this** 上等待的线程

![7245cb38e0534bf58cd76fe6732c34d2.png][]

在正常的情况下，wait 类型函数返回时要不是因为被唤醒，要不是因为超时才返回，但是在实际中发现，因为操作系统的原因，wait 类型在不满足条件时，它也会返回，这就导致了虚假唤醒，因此我们一般都是使用带有谓词参数的 wait 函数。

**condition\_variable 的一般用法如下：**

*  如果两个线程**共享的变量**存在其中**一个线程读取，另一个线程写入的情况，或者存在两个线程都要写**的情况，那么它们共享的变量需要在互斥锁的保护之下。
 *  notify 函数本身也需要在互斥锁的保护之下
 *  wait 函数本身就带有一个互斥锁的参数

condition_variable threadqueueDemo::cv;
    mutex threadqueueDemo::mDisplayMutex;
    queue<int> threadqueueDemo::mDisplayQueue;
    
    void threadqueueDemo::threadLoop() {
    	    while (true) {
    		{
    		    unique_lock<mutex> lock(mDisplayMutex);
    			if (mDisplayQueue.empty()) {
    				cout << "thread tid = " << this_thread::get_id() << endl;
    				cv.wait(lock);
    			} else {
    				auto p = mDisplayQueue.front();
    				mDisplayQueue.pop();
    				cout << "get data p =" << p << endl;
    			}
    		}
    	}
    }
    
    threadqueueDemo::threadqueueDemo() {
    	t = new thread(threadLoop);
    	t->detach();
    
    	for (int i = 0; i < 20; i++) {
    		pushdata();
    		this_thread::sleep_for(chrono::milliseconds(500ms));
    	}
    }
    
    void threadqueueDemo::pushdata() {
    	unique_lock<mutex> lock(mDisplayMutex);
    	{
    		++mData;
    		mDisplayQueue.push(mData);
        	cv.notify_all();
    	}
    }
    
    
    void threadqueueDemo::testthreadqueue(int argc, char* argv[]) {
    	threadqueueDemo* t = new threadqueueDemo;
    }

### **3. 生产者消费问题** ###

在这里，我们使用条件变量，解决生产者-消费者问题，该问题的主要描述如下：

**生产者-消费这个问题，也称为有限缓冲问题，是一个进程/线程同步问题的经典案例。该问题描述了共享固定大小缓冲区的两个进程/线程—— 就是所谓的生产者和消费者，在实际运行时会发生的问题。**

生产者的主要作用是生成一定量的数据放到缓冲区，然后重复此过程。与此同时，消费者也在缓冲区中消耗这些数据。该问题的关键就是要保证生产者不会再缓冲区满时加入数据，消费者也不会在缓冲区空时消耗数据。

要解决该问题，就必须让生产者在缓冲区满时休眠，等到下次消费者消耗缓冲区中的数据的时候，生产者才能被唤醒，开始往缓冲区添加数据。

同样，也可以让消费者在缓冲区空时进入休眠，等到生产者往缓冲区添加数据知乎，再唤醒消费者。

示例代码如下：

std::mutex g_cvMutex;
    std::condition_variable g_cv;
    
    
    std::deque<int> g_data_deque;
    const int  MAX_NUM = 30;
    int g_next_index = 0;
    
    const int PRODUCER_THREAD_NUM = 3;
    const int CONSUMER_THREAD_NUM = 3;
    
    void producer_thread(int thread_id)
    {
    	while (true) {
    		std::this_thread::sleep_for(std::chrono::milliseconds(500));
    		std::unique_lock<mutex> lk(g_cvMutex);
    		if (g_data_deque.size() <= MAX_NUM) {
    			g_next_index++;
    			g_data_deque.push_back(g_next_index);
    			std::cout << "producer_thread: " << thread_id << " producer data: " << g_next_index;
    			std::cout << " queue size: " << g_data_deque.size() << std::endl;
    		} else {
    			g_cv.notify_all();
    		}
    	}
    }
    
    
    void consumer_thread(int thread_id) {
    	while (true) {
    		std::this_thread::sleep_for(std::chrono::milliseconds(500));
    		std::unique_lock <std::mutex> lk(g_cvMutex);
    		if (!g_data_deque.empty()) {
    			int data = g_data_deque.front();
    			g_data_deque.pop_front();
    			std::cout << "\tconsumer_thread: " << thread_id << " consumer data: ";
    			std::cout << data << " deque size: " << g_data_deque.size() << std::endl;
    		} else {
    			g_cv.wait(lk);
    		}
    	}
    }
    
    
    producerconsumer::producerconsumer() {
    
    	std::thread *producerthread[PRODUCER_THREAD_NUM];
    	std::thread *consumerthread[CONSUMER_THREAD_NUM];
    
    
    	for (int i = 0; i < PRODUCER_THREAD_NUM; i++) {
    		producerthread[i] = new thread(producer_thread, ref(i));
    	}
    
    	for (int j = 0; j < CONSUMER_THREAD_NUM; j++) {
    		consumerthread[j] = new thread(consumer_thread, ref(j));
    	}
    	
    	for (int i = 0; i < PRODUCER_THREAD_NUM; i++) {
    		producerthread[i]->join();
    	}
    
    	for (int j = 0; j < CONSUMER_THREAD_NUM; j++) {
    		consumerthread[j]->join();
    	}
    }

*  程序中创建三个生产者线程，每个线程都会向 **g\_data\_deque 这个双端队列中 push** 数据
 *  三个生产者线程push 的 int 类型使用的是全局变量，是在已有的基础上累加的 int 类型变量
 *  **程序中创建了三个消费者线程，只要双端队列不为空，就会将数据从双端队列中 pop 出来**
 *  生产者消费者会有抢占的操作，就是消费者并没有完全将队列消费完，让队列为空的时候，生产者再次将生产的内容放入队列，相当于有6个线程，在g\_cvMutex 的保护下，生产和消费队列里面的内容

![665f2edb8d184f9ba3839b2f0d7f4d4c.png][]

执行结果如下：

![2065bae7d495483b87b302b9e8a1fff5.png][]

### **4. 为什么条件变量要和互斥锁一起使用** ###

首先要说明为什么要引入条件变量，如果某个线程需要等待某个条件成立，而这个条件又是其他线程给出的，其中的一种解决方案是：

threadA() {
    
    lock
    if(条件满足)
    unlock
        excute code;
    } else {
        sleep(n);
    }
    }

*  如果一直判断条件是否满足，线程没有阻塞等待，是非常浪费CPU资源的，所以引入了 sleep 函数，相当于 sleep 一段时间后再做判断
 *  注意此时的**判断条件是临界资源**，需要锁的保护

引入条件变量后，若条件不满足，**则相应线程被阻塞直至条件发生变化被唤醒，再去查询**条件是否满足，避免了上述条件变化发生之前的无用查询包括加解锁

**条件变量的引入，使多个线程以一种无竞争的方式等待条件的改变**。

在使用 condition\_variable 的 wait 函数之前，通常要进行条件判断，**而此条件属于临界资源（在读取此条件的时候不能被打断，比如去更新这个条件**），需要在访问之前加锁，这是为了保护临界资源的需要。

unique_lock<mutex> lock(mDisplayMutex);
    if (mDisplayQueue.empty()) {
    
    	cv.wait(lock);
    }

比如上面的代码中如果不加锁保护，在判断 mDisplayQueue.empty() 的条件成立后，此时线程被挂起，调度了另一个线程，另外的线程用于唤醒这个 condition\_variable，此时当前线程因为还没有处于 wait 状态上（没有处于调度器的等待队列上），所以会丢失掉这个唤醒操作，如果这个唤醒操作只有一次，那么当前线程很可能永远处于阻塞等待的状态上。

所以 condtion\_variable 加互斥锁的作用，就是保证全局条件和wait 的操作是原子操作

> [C++面试问题：为什么条件变量要和互斥锁一起使用？\_条件变量为什么要和锁一起用\_August8757的博客-CSDN博客][C_August8757_-CSDN]

[4ee63b444747432a9183fc81e3ecc02a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b7851e342d71491ab8d92429a30608f7.png
[7245cb38e0534bf58cd76fe6732c34d2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/354e18c77af8484ab14fa46a5ed13f0d.png
[665f2edb8d184f9ba3839b2f0d7f4d4c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/b485108f91674eb9b8a8ad1e479d10a1.png
[2065bae7d495483b87b302b9e8a1fff5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/15/065dd95e1e1b4f51bfd46dbe4376e7ee.png
[C_August8757_-CSDN]: https://blog.csdn.net/weixin_37822792/article/details/112430570