Linux条件变量线程池详解

一、条件变量

        【互斥量】解决了线程间同步的问题，避免了多线程对同一块临界资源访问产生的冲突，但同一时刻对临界资源的访问，不论是生产者还是消费者，都需要竞争互斥锁，由此也带来了竞争的问题。即生产者和消费者、消费者和消费者之间时刻都在竞争这把锁，而临界资源是有限的，当临界资源为空的时候，消费者之间的竞争便没有意义，反而降低了运行效率。

        有没有什么办法可以等生产者线程生产出资源，消费者线程再去竞争锁消费呢？这就是条件变量的作用。

        正如互斥量保护了【临界资源】，条件变量也保护【条件】资源，当条件不符合时即【条件】资源空缺，消费者线程休眠等待；当【条件】资源产生，消费者线程被唤醒然后去消费资源。这样便解决了线程间等待的问题，提高了运行效率。

pthread_cond_t cond  //定义条件变量
pthread_cond_init(&cond) //动态初始化pthread_cond_t cond = PTHREAD_COND_INITIALIZER //静态创建并初始化/*消费者线程休眠等待生产者线程通知*/
int pthread_cond_wait(pthread_cond_t *restrict cond,pthread_mutex_t *restrict mutex)/*生产者线程生产完资源后发送通知*/
int pthread_cond_signal(pthread_cond_t *cond)    //唤醒一个休眠的线程
int pthread_cond_broadcast(pthread_cond_t *cond) //广播唤醒所有休眠的线程

• 条件变量的使用要绑定互斥锁

  • 因为条件变量的使用过程中，对于生产者线程，需要产生资源当然要上锁；对于消费者线程，需要消费资源甚至还要判断资源是否为空，也要上锁

• 【wait】函数的具体动作

  • 进入等待的线程列表中休眠并释放锁

  • 被唤醒后返回，同时上锁

wait函数的每个动作都是【原子操作】，动作连续并且不会被别的程序干扰，意味着CPU调度一定能保证动作粒一气呵成

         消费线程不判断资源是否为空直接等待的话容易丢失signal信号（假设生产线程提前生产出资源也signal了）

        唤醒也可以用broadcast，这种情况消费线程一定要判断资源是否为空，否则链表形式的资源容易出现【段错误】（因为多个线程去争抢资源时，总有抢不到的）

二、线程池 

若干线程的集合，可用结构体来构造

必要性：当会出现大量执行时间短的任务时，甚至这个时间比线程创建+销毁的时间还短，这时候再创建多个线程就不划算了，可以未雨绸缪，构建线程池，提前创建多个线程来节省开销

 线程池的实现过程

1.创建任务队列、线程池的结构体、定义一个线程池

#define pool_num (10)typedef struct Task{struct Task *next;void *(*func)(void *);void *arg;}Task;typedef struct{pthread_mutex_t mutex;pthread_cond_t cond;Task *task_head;int busywork;pthread_t tid[pool_num];}ThreadPool;ThreadPool *pool;

2.初始化线程池、线程的工作

void pool_init()
{pool = (ThreadPool *)malloc(sizeof(ThreadPool));pthread_mutex_init(&pool->mutex,NULL);pthread_cond_init(&pool->cond,NULL);pool->task_head = NULL;pool->busywork = 0;for(i=0;i<pool_num;i++){pthread_create(&pool->tid[i],NULL,workthread,NULL);}
}void *workthread(void *arg)
{while(1){	//	usleep(10000);  //avoid other process couldn't rob the lock and several process repeatedly rob the lockpthread_mutex_lock(&pool->mutex);while(pool->task_head == NULL){pthread_cond_wait(&pool->cond,&pool->mutex);}Task *ptask = pool->task_head;//任务取走线程池的头个线程pool->task_head = ptask->next;//重置线程池的头个线程为下一个pool->busywork--;//pthread_mutex_unlock(&pool->mutex);printf("%ld start task %d\n",pthread_self(),(int)ptask->arg);ptask->func(ptask->arg);}
}

3.添加任务

void *realwork(void *arg)
{usleep(100000);printf("finish task %d\n",(int)arg);}void add_task(int arg)
{pthread_mutex_lock(&pool->mutex);while(pool->busywork >= pool_num){pthread_mutex_unlock(&pool->mutex);usleep(10000);pthread_mutex_lock(&pool->mutex);}pthread_mutex_unlock(&pool->mutex);//以上为判断任务数量是否超过线程池数量，超过则不再允许添加任务Task *newtask;newtask = (Task *)malloc(sizeof(Task));newtask->func = realwork;newtask->arg = (void *)arg;pthread_mutex_lock(&pool->mutex);Task *p = pool->task_head;if(p == NULL){pool->task_head = newtask;}else{while(p->next != NULL){p = p->next;}p->next = newtask;pthread_cond_signal(&pool->cond);pool->busywork++;}pthread_mutex_unlock(&pool->mutex);
}	

4.销毁线程池、任务队列、互斥锁、条件变量

void pool_destroy()
{Task *p;while(pool->task_head != NULL){p = pool->task_head;pool->task_head = p->next;free(p);}pthread_mutex_destroy(&pool->mutex);pthread_cond_destroy(&pool->cond);free(pool);
}

示例

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <unistd.h>
#include <stdlib.h>#define pool_num (10)
int i = 0;typedef struct Task{struct Task *next;void *(*func)(void *);void *arg;}Task;typedef struct{pthread_mutex_t mutex;pthread_cond_t cond;Task *task_head;int busywork;pthread_t tid[pool_num];}ThreadPool;ThreadPool *pool;void *workthread(void *arg)
{while(1){	//	usleep(10000);  //avoid other process couldn't rob the lock and several process repeatedly rob the lockpthread_mutex_lock(&pool->mutex);while(pool->task_head == NULL){pthread_cond_wait(&pool->cond,&pool->mutex);}Task *ptask = pool->task_head;//任务取走线程池的头个线程pool->task_head = ptask->next;//重置线程池的头个线程为下一个pool->busywork--;//pthread_mutex_unlock(&pool->mutex);printf("%ld start task %d\n",pthread_self(),(int)ptask->arg);ptask->func(ptask->arg);}
}void *realwork(void *arg)
{usleep(100000);printf("finish task %d\n",(int)arg);}void add_task(int arg)
{pthread_mutex_lock(&pool->mutex);while(pool->busywork >= pool_num){pthread_mutex_unlock(&pool->mutex);usleep(10000);pthread_mutex_lock(&pool->mutex);}pthread_mutex_unlock(&pool->mutex);Task *newtask;newtask = (Task *)malloc(sizeof(Task));newtask->func = realwork;newtask->arg = (void *)arg;pthread_mutex_lock(&pool->mutex);Task *p = pool->task_head;if(p == NULL){pool->task_head = newtask;}else{while(p->next != NULL){p = p->next;}p->next = newtask;pthread_cond_signal(&pool->cond);pool->busywork++;}pthread_mutex_unlock(&pool->mutex);
}	void pool_init()
{pool = (ThreadPool *)malloc(sizeof(ThreadPool));pthread_mutex_init(&pool->mutex,NULL);pthread_cond_init(&pool->cond,NULL);pool->task_head = NULL;pool->busywork = 0;for(i=0;i<pool_num;i++){pthread_create(&pool->tid[i],NULL,workthread,NULL);}}void pool_destroy()
{Task *p;while(pool->task_head != NULL){p = pool->task_head;pool->task_head = p->next;free(p);}pthread_mutex_destroy(&pool->mutex);pthread_cond_destroy(&pool->cond);free(pool);
}int main()
{printf("mypid is %d\n",getpid());pool_init();sleep(5);for(i=1;i<=40;i++){add_task(i);}sleep(5);pool_destroy();while(1){printf("now im alone\n");sleep(2);}	}