后台C++开发你一定要知道的条件变量

后台C++开发你一定要知道的条件变量

今天因为工作需要，需要帮同事用C语言（不是C++）写一个生产者消费者的任务队列工具库，考虑到不能使用任何第三库和C++的任何特性，所以我将任务队列做成一个链表，生产者在队列尾部加入任务，消费者在队列头部取出任务。很快就写好了，代码如下： /** * 线程池工具, ctrip_thread_pool.h * zhangyl 2018.03.23 */ #ifndef __CTRIP_THREAD_POOL_H__ #define __CTRIP_THREAD_POOL_H__ #include <pthread.h> #ifndef NULL #define NULL 0 #endif #define PUBLIC PUBLIC struct ctrip_task { struct ctrip_task* pNext; int value; }; struct ctrip_thread_info { //线程退出标志 int thread_running; int thread_num; int tasknum; struct ctrip_task* tasks; pthread_t* threadid; pthread_mutex_t mutex; pthread_cond_t cond; }; /* 初始化线程池线程数目 * @param thread_num 线程数目, 默认为8个 */ PUBLIC void ctrip_init_thread_pool(int thread_num); /* 销毁线程池 */ PUBLIC void ctrip_destroy_thread_pool(); /**向任务池中增加一个任务 * @param t 需要增加的任务 */ PUBLIC void ctrip_thread_pool_add_task(struct ctrip_task* t); /**从任务池中取出一个任务 * @return 返回得到的任务 */ struct ctrip_task* ctrip_thread_pool_retrieve_task(); /**执行任务池中的任务 * @param t 需要执行的任务 */ PUBLIC void ctrip_thread_pool_do_task(struct ctrip_task* t); /**线程函数 * @param thread_param 线程参数 */ void* ctrip_thread_routine(void* thread_param); #endif //!__CTRIP_THREAD_POOL_H__ /** * 线程池工具, ctrip_thread_pool.c * zhangyl 2018.03.23 */ #include "ctrip_thread_pool.h" #include <stdio.h> #include <stdlib.h> struct ctrip_thread_info g_threadinfo; int thread_running = 0; void ctrip_init_thread_pool(int thread_num) { if (thread_num <= 0) thread_num = 5; pthread_mutex_init(&g_threadinfo.mutex, NULL); pthread_cond_init(&g_threadinfo.cond, NULL); g_threadinfo.thread_num = thread_num; g_threadinfo.thread_running = 1; g_threadinfo.tasknum = 0; g_threadinfo.tasks = NULL; thread_running = 1; g_threadinfo.threadid = (pthread_t*)malloc(sizeof(pthread_t) * thread_num); int i; for (i = 0; i < thread_num; ++i) { pthread_create(&g_threadinfo.threadid[i], NULL, ctrip_thread_routine, NULL); } } void ctrip_destroy_thread_pool() { g_threadinfo.thread_running = 0; thread_running = 0; pthread_cond_broadcast(&g_threadinfo.cond); int i; for (i = 0; i < g_threadinfo.thread_num; ++i) { pthread_join(g_threadinfo.threadid[i], NULL); } free(g_threadinfo.threadid); pthread_mutex_destroy(&g_threadinfo.mutex); pthread_cond_destroy(&g_threadinfo.cond); } void ctrip_thread_pool_add_task(struct ctrip_task* t) { if (t == NULL) return; pthread_mutex_lock(&g_threadinfo.mutex); struct ctrip_task* head = g_threadinfo.tasks; if (head == NULL) g_threadinfo.tasks = t; else { while (head->pNext != NULL) { head = head->pNext; } head->pNext = t; } ++g_threadinfo.tasknum; //当有变化后，使用signal通知wait函数 pthread_cond_signal(&g_threadinfo.cond); pthread_mutex_unlock(&g_threadinfo.mutex); } struct ctrip_task* ctrip_thread_pool_retrieve_task() { struct ctrip_task* head = g_threadinfo.tasks; if (head != NULL) { g_threadinfo.tasks = head->pNext; --g_threadinfo.tasknum; printf("retrieve a task, task value is [%d]\n", head->value); return head; } printf("no task\n"); return NULL; } void* ctrip_thread_routine(void* thread_param) { printf("thread NO.%d start.\n", (int)pthread_self()); while (thread_running/*g_threadinfo.thread_running*/) { struct ctrip_task* current = NULL; pthread_mutex_lock(&g_threadinfo.mutex); while (g_threadinfo.tasknum <= 0) { //如果获得了互斥锁，但是条件不合适的话，wait会释放锁，不往下执行。 //当变化后，条件合适，将直接获得锁。 pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); if (!g_threadinfo.thread_running) break; current = ctrip_thread_pool_retrieve_task(); if (current != NULL) break; }// end inner-while-loop pthread_mutex_unlock(&g_threadinfo.mutex); ctrip_thread_pool_do_task(current); }// end outer-while-loop printf("thread NO.%d exit.\n", (int)pthread_self()); } void ctrip_thread_pool_do_task(struct ctrip_task* t) { if (t == NULL) return; //TODO: do your work here printf("task value is [%d]\n", t->value); //TODO：如果t需要释放，记得在这里释放 } 测试代码如下： // ctrip_thread_pool.cpp : Defines the entry point for the console application. // //#include "stdafx.h" #include "ctrip_thread_pool.h" #include <stdlib.h> #include <unistd.h> int main(int argc, char* argv[]) { ctrip_init_thread_pool(5); struct ctrip_task* task = NULL; int i; for (i = 0; i < 100; ++i) { task = (struct ctrip_task*)malloc(sizeof(struct ctrip_task)); task->value = i + 1; task->pNext = NULL; printf("add task, task value [%d]\n", task->value); ctrip_thread_pool_add_task(task); } sleep(10); ctrip_destroy_thread_pool(); return 0; } 代码很快就写好了，但是每次程序只能执行前几个加到任务池子里面的任务，导致池子有不少任务积累在池子里面。甚是奇怪，我也看了半天才看出结果。聪明的你，能看出上述代码为啥只能执行加到池子里面的前几个任务？先不要看答案，自己想一会儿。 linux条件变量是做后台开发必须熟练掌握的基础知识，而条件变量使用存在以下几个非常让人迷惑的地方，讲解如下 第一、必须要结合一个互斥体一起使用。使用流程如下： pthread_mutex_lock(&g_threadinfo.mutex) pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); pthread_mutex_unlock(&g_threadinfo.mutex); 上面的代码，我们分为一二三步，当条件不满足是pthread_cond_wait会挂起线程，但是不知道你有没有注意到，如果在第二步挂起线程的话，第一步的mutex已经被上锁，谁来解锁？mutex的使用原则是谁上锁谁解锁，所以不可能在其他线程来给这个mutex解锁，但是这个线程已经挂起了，这就死锁了。所以pthread_cond_wait在挂起之前，额外做的一个事情就是给绑定的mutex解锁。反过来，如果条件满足，pthread_cond_wait不挂起线程，pthread_cond_wait将什么也不做，这样就接着走pthread_mutex_unlock解锁的流程。而在这个加锁和解锁之间的代码就是我们操作受保护资源的地方。 第二，不知道你有没有注意到pthread_cond_wait是放在一个while循环里面的： pthread_mutex_lock(&g_threadinfo.mutex); while (g_threadinfo.tasknum <= 0) { //如果获得了互斥锁，但是条件不合适的话，wait会释放锁，不往下执行。 //当变化后，条件合适，将直接获得锁。 pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); if (!g_threadinfo.thread_running) break; current = ctrip_thread_pool_retrieve_task(); if (current != NULL) break; }// end inner-while-loop pthread_mutex_unlock(&g_threadinfo.mutex); 注意，我说的是内层的while循环，不是外层的。pthread_cond_wait一定要放在一个while循环里面吗？一定要的。这里有一个非常重要的关于条件变量的基础知识，叫条件变量的虚假唤醒（spurious wakeup），那啥叫条件变量的虚假唤醒呢？假设pthread_cond_wait不放在这个while循环里面，正常情况下，pthread_cond_wait因为条件不满足，挂起线程。然后，外部条件满足以后，调用pthread_cond_signal或pthread_cond_broadcast来唤醒挂起的线程。这没啥问题。但是条件变量可能在某些情况下也被唤醒，这个时候pthread_cond_wait处继续往下执行，但是这个时候，条件并不满足（比如任务队列中仍然为空）。这种唤醒我们叫“虚假唤醒”。为了避免虚假唤醒时，做无意义的动作，我们将pthread_cond_wait放到while循环条件中，这样即使被虚假唤醒了，由于while条件（比如任务队列是否为空，资源数量是否大于0）仍然为true，导致线程进行继续挂起。有人说条件变量是最不可能用错的线程之间同步技术，我却觉得这是最容易使用错误的线程之间同步技术。 上述代码存在的问题是，只考虑了任务队列开始为空，生产者后来添加了任务，条件变量被唤醒，然后消费者取任务执行的逻辑。假如一开始池中就有任务呢？这个原因导致，只有开始的几个添加到任务队列中任务被执行。因为一旦任务队列不为空。内层while循环条件将不再满足，导致消费者线程不再从任务队列中取任务消费。正确的代码如下： /** * 线程池工具, ctrip_thread_pool.c（修正后的代码） * zhangyl 2018.03.23 */ #include "ctrip_thread_pool.h" #include <stdio.h> #include <stdlib.h> struct ctrip_thread_info g_threadinfo; void ctrip_init_thread_pool(int thread_num) { if (thread_num <= 0) thread_num = 5; pthread_mutex_init(&g_threadinfo.mutex, NULL); pthread_cond_init(&g_threadinfo.cond, NULL); g_threadinfo.thread_num = thread_num; g_threadinfo.thread_running = 1; g_threadinfo.tasknum = 0; g_threadinfo.tasks = NULL; g_threadinfo.threadid = (pthread_t*)malloc(sizeof(pthread_t) * thread_num); int i; for (i = 0; i < thread_num; ++i) { pthread_create(&g_threadinfo.threadid[i], NULL, ctrip_thread_routine, NULL); } } void ctrip_destroy_thread_pool() { g_threadinfo.thread_running = 0; pthread_cond_broadcast(&g_threadinfo.cond); int i; for (i = 0; i < g_threadinfo.thread_num; ++i) { pthread_join(g_threadinfo.threadid[i], NULL); } free(g_threadinfo.threadid); pthread_mutex_destroy(&g_threadinfo.mutex); pthread_cond_destroy(&g_threadinfo.cond); } void ctrip_thread_pool_add_task(struct ctrip_task* t) { if (t == NULL) return; pthread_mutex_lock(&g_threadinfo.mutex); struct ctrip_task* head = g_threadinfo.tasks; if (head == NULL) g_threadinfo.tasks = t; else { while (head->pNext != NULL) { head = head->pNext; } head->pNext = t; } ++g_threadinfo.tasknum; //当有变化后，使用signal通知wait函数 pthread_cond_signal(&g_threadinfo.cond); pthread_mutex_unlock(&g_threadinfo.mutex); } struct ctrip_task* ctrip_thread_pool_retrieve_task() { struct ctrip_task* head = g_threadinfo.tasks; if (head != NULL) { g_threadinfo.tasks = head->pNext; --g_threadinfo.tasknum; printf("retrieve a task, task value is [%d]\n", head->value); return head; } printf("no task\n"); return NULL; } void* ctrip_thread_routine(void* thread_param) { printf("thread NO.%d start.\n", (int)pthread_self()); while (g_threadinfo.thread_running) { struct ctrip_task* current = NULL; pthread_mutex_lock(&g_threadinfo.mutex); while (g_threadinfo.tasknum <= 0) { //如果获得了互斥锁，但是条件不合适的话，wait会释放锁，不往下执行。 //当变化后，条件合适，将直接获得锁。 pthread_cond_wait(&g_threadinfo.cond, &g_threadinfo.mutex); if (!g_threadinfo.thread_running) break; }// end inner-while-loop current = ctrip_thread_pool_retrieve_task(); pthread_mutex_unlock(&g_threadinfo.mutex); ctrip_thread_pool_do_task(current); }// end outer-while-loop printf("thread NO.%d exit.\n", (int)pthread_self()); } void ctrip_thread_pool_do_task(struct ctrip_task* t) { if (t == NULL) return; //TODO: do your work here printf("task value is [%d]\n", t->value); //TODO：如果t需要释放，记得在这里释放 } ok，不知道你有没有看明白呀？