条件变量存在的陷阱、生产者与消费者模型

一、条件变量的使用

Linux下条件变量相关的API如下：
pthread_cond_init：用于初始化条件变量。
pthread_cond_destory：销毁条件变量。
pthread_cond_broadcast：以广播的方式唤醒所有等待目标条件变量的线程。
pthread_cond_wait：用于等待目标条件变量。该函数调用时需要传入mutex参数(加锁的互斥锁)，函数执行时，先把调用线程放入条件变量的请求队列，然后将互斥锁mutex解锁，当函数成功返回为0时，表示重新抢到了互斥锁，互斥锁会再次被锁上，也就是说函数内部会有一次解锁和加锁操作。

条件变量的基本使用方法如下：

// 1. 生产者使用条件变量的方法
......
pthread_mutex_lock(&mutex);
// 这里将产生任务，供消费者线程进行处理
pthread_mutex_unlock(&mutex);
pthread_cond_signal(&cond); // 唤醒一个消费者线程处理
......// 2. 消费者线程使用条件变量的方法
......
pthread_mutex_lock(&mutex);
while(线程执行的目标条件是否满足){pthread_cond_wait(&cond, &mutex);
}
pthread_mutex_unlock(&mutex);
......

二、条件变量深究与陷阱

2.1 条件变量的加锁问题

1.pthread_cond_wait(&cond, &mutex)做了哪些事情？
答：该函数执行时会将线程放在条件变量的请求队列后，内部解锁线程等待被pthread_cond_broadcast信号唤醒或者pthread_cond_signal信号唤醒，唤醒后去竞争锁，若竞争到互斥锁，内部再次加锁。

2.消费者线程使用前要加锁的原因？
答：因为多线程访问，为了避免资源竞争，所以要加锁，使得每个线程互斥的访问公有资源。

3.pthread_cond_wait内部为什么要解锁？
答：使用while判断的时候，满足执行条件，线程便会调pthread_cond_wait阻塞自己，此时它还在持有锁，如果他不解锁，那么其他线程将会无法访问公有资源。具体到pthread_cond_wait的内部实现，当pthread_cond_wait被调用线程阻塞的时候，pthread_cond_wait会自动释放互斥锁。

4.为什么要把调用线程放入条件变量的请求队列后再解锁？
答：线程是并发执行的，如果在把调用线程A放在等待队列之前，就释放了互斥锁，这就意味着其他线程比如线程B可以获得互斥锁去访问公有资源，这时候线程A所等待的条件改变了，但是它没有被放在等待队列上，导致A忽略了等待条件被满足的信号。倘若在线程A调用pthread_cond_wait开始，到把A放在等待队列的过程中，都持有互斥锁，其他线程无法得到互斥锁，就不能改变公有资源。

5.为什么pthread_cond_wait函数最后还要加锁？
答：将线程放在条件变量的请求队列后，将其解锁，此时等待被唤醒，若成功竞争到互斥锁，再次加锁，则该消费者线程获得公共资源的消费权。

2.2 消费线程的虚假唤醒

为什么判断线程执行的条件用while而不是if？

答：一般来说，在多线程资源竞争的时候，消费者线程判断资源是否可用，不可用，便调用pthread_cond_wait，生产者线程如果判断资源可用的话，则调用pthread_cond_signal发送一个资源可用信号。

然而，在pthread_cond_wait成功之后，消费者线程就一定可以消费资源么？答案是否定的，如果同时有两个或者两个以上的线程正在等待此资源，wait返回后，资源可能已经被使用了，于是被唤醒但却没有获得消费权利的情况就叫做虚假唤醒。

再具体点，有可能多个线程都在等待这个资源可用的信号，信号发出后只有一个资源可用，但是有A，B两个消费者线程都在等待，B线程的比较速度快，获得互斥锁，然后加锁，消耗资源，然后解锁，之后A获得互斥锁，但A回去发现资源已经被使用了，它便有两个选择，一个是去访问不存在的资源，另一个就是继续等待，那么继续等待下去的条件就是使用while，要不然使用if的话pthread_cond_wait返回后，就会顺序执行下去。

所以，在这种多消费者情况下，应该使用while而不是if，如果只有一个消费者，那么使用if是可以的。

三、基于条件变量实现的生产者与消费者模型

文章写到这里，相信大家对条件变量的理解已经到位了，接下来咱直接使用条件变量来实现一个基本的生产者与消费者模型，c++代码如下：

#include <iostream>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <list>using namespace std;// 任务
class task {public:task(int id){this->taskId = id;}void doTask(){cout << "  dotask id = " << taskId << endl;}private:int taskId;
};list<task*> taskList;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;// 生产者线程
void *producerThreadFunc(void *param){int taskId = 0;task *t = nullptr;while (true){t = new task(taskId);pthread_mutex_lock(&mutex);taskList.emplace_back(t);cout << "Produce task, id = " << taskId << endl;pthread_mutex_unlock(&mutex);pthread_cond_signal(&cond); // 唤醒一个消费者线程处理taskId++;sleep(1);}return nullptr;
}// 消费者线程
void *consumerThreadFunc(void *param){task *t = nullptr;while (true){pthread_mutex_lock(&mutex);while(taskList.empty()){pthread_cond_wait(&cond, &mutex);}t = taskList.front();taskList.pop_front();pthread_mutex_unlock(&mutex);cout << "ThreadID = " << pthread_self();t->doTask();delete t;}return nullptr;
}int main(){pthread_t producerThread;pthread_t consumerThread[10]; // 创建10个消费线程pthread_create(&producerThread, NULL, producerThreadFunc, NULL);for(int i = 0; i < 10; i++){pthread_create(&consumerThread[i], NULL, consumerThreadFunc, NULL);}pthread_join(producerThread, NULL);for(int i = 0; i < 10; ++i){pthread_join(consumerThread[i], NULL);}pthread_cond_destroy(&cond);pthread_mutex_destroy(&mutex);return 0;
}

执行结果如下：

参考博客如下：
最新版Web服务器项目详解 - 09 日志系统（上）：https://mp.weixin.qq.com/s/IWAlPzVDkR2ZRI5iirEfCg