本文以营业厅为例子，实现简单的线程池
一、线程池介绍
现在的企业客户端数以百万，如果某一时刻同时向服务器发消息，那么服务器要处理这些消息是同时开百万个线程吗？？当然不行！！
根据posix标准，一个线程大概占8M空间，那么对于16G的内存，最多也就是开2048个线程。因此此时需要线程池。
线程池是一种常见的并发编程模型，它可以有效地管理和复用线程资源，提高程序运行效率和响应速度。通常情况下，线程池由一个任务队列和若干个工作线程组成。
线程池的优点包括：
- 提高性能：线程池可以重复使用现有的线程，避免了频繁创建和销毁线程的开销，从而提高程序的性能。
- 提高响应速度：线程池中已经创建好的空闲线程可以立即响应任务请求，减少了任务等待时间，提高了程序的响应速度。
- 控制并发数量：线程池可以限制同时执行的任务数量，有效地控制了系统资源的消耗。
- 管理线程：线程池可以统一管理所有线程，并且可以设置线程池参数来满足业务需求。
- 提供更好的可扩展性：通过使用线程池，我们可以灵活地调整系统中任务队列长度和工作线程数目，从而实现更好的可扩展性。
二、实现一个简单的线程池
以银行营业厅进行类比：
#include<stdio.h> #include <string.h> #include <stdlib.h> #include<pthread.h> #include<errno.h> //'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' //''''''''''''''''''''属性'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' #define LIST_INSERT(item ,list) do{\ item->prev=NULL; \ item->next=list; \ if ((list) != NULL) list->prev=item; \ (list)=item; \ }while(0) #define LIST_REMOVE(item,list) do{\ if(item->prev != NULL) item->prev->next=item->next; \ if(item->next != NULL) item->next->prev=item->prev; \ if(list==item) list=item->next; \ item->prev=item->next=NULL; \ }while(0) //顾客 struct nTask { void (*task_fun)(struct nTask*); void *user_data; struct nTask *prev; struct nTask *next; }; //业务员 struct nWorker{ pthread_t threadid; int terminate; //判断线程是否终止 struct nManager *manager; //业务员中应当有管理权限，如查看任务队列是否为空 struct nWorker *prev; struct nWorker *next; }; //线程池 typedef struct nManager{ struct nTask *tasks; //任务队列 struct nWorker *workers; //执行队列 pthread_mutex_t mutex; pthread_cond_t cond; }ThreadPool; //'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' //''''''''''''''''''''接口层'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' //线程的回调 //判断任务队列是否内是否有任务，若有取出执行，若无等待任务 static void *nThreadPoolCallback(void *arg){ struct nWorker *worker=(struct nWorker *)arg; printf("nThreadPoolCallback\n"); while (1){ //加锁 pthread_mutex_lock(&worker->manager->mutex); //若任务队列为空，等待 while(worker->manager->tasks == NULL){ if (worker->terminate) break; //此时该线程上的任务队列为空，因此不解锁退出也不会影响其他线程，即不会造成死锁 pthread_cond_wait(&worker->manager->cond,&worker->manager->mutex); } if(worker->terminate){ //需要先解锁后退出，否则会死锁 pthread_mutex_unlock(&worker->manager->mutex); break; } //取出第一个任务，而后从任务队列删除该任务 struct nTask *task=worker->manager->tasks; //指向任务队列（链表）的第一个任务 LIST_REMOVE(task,worker->manager->tasks); //解锁 pthread_mutex_unlock(&worker->manager->mutex); //线程执行该任务 task->task_fun(task); } free(worker); } //创建线程池 int nThreadPoolCreat(ThreadPool *pool,int numWorkers){ if (pool == NULL) return -1; if (numWorkers < 1) numWorkers=1; memset(pool,0,sizeof(ThreadPool)); //初始化 pthread_mutex_init(&pool->mutex,NULL); pthread_cond_init(&pool->cond,NULL); //创建线程(业务员)，加入到执行队列中 int i=0; for(i=0;i<numWorkers;i++){ //分配内存，并初始化（堆区创建的数据一定要初始化） struct nWorker *worker=(struct nWorker *)malloc(sizeof(struct nWorker )); if (worker == NULL){ perror("malloc"); return -2; } memset(worker,0,sizeof(struct nWorker)); worker->manager=pool; //创建线程(业务员) int ret = pthread_create(&worker->threadid, NULL, nThreadPoolCallback, worker); if (ret){ perror("pthread_creat"); free(worker); return -3; } //将线程（业务员）加入到任务队列中 printf("LIST_INSERT\n"); LIST_INSERT(worker,pool->workers); } //sucess return 0; } //销毁线程池 int nThreadPoolDestory(ThreadPool *pool){ //将线程池内的所有线程状态terminate设为真，表示退出 struct nWorker *worker=NULL; for (worker=pool->workers;worker!=NULL;worker=worker->next){ worker->terminate; } //获取与条件变量相关联的互斥锁-->向所有等待在该条件变量上的线程发送信号以唤醒它们-->释放线程池的互斥锁 //pthread_cond_broadcast配合pthread_cond_wait使用，获取的是同一把锁 pthread_mutex_lock(&pool->mutex); pthread_cond_broadcast(&pool->cond); pthread_mutex_unlock(&pool->mutex); //将线程池中的workers和tasks指针分别设置为NULL，释放内存空间 pool->tasks=NULL; pool->workers=NULL; return 0; } //往线程池加入任务 int nThreadPoolPushTask(ThreadPool *pool,struct nTask *task){ /*pthread_cond_signal通知等待在条件变量上的某个线程去获取锁并继续执行。如果此时没有加锁， 那么可能会出现竞态条件（Race Condition），也就是多个线程同时进入临界区域并修改共享数据，从而导致程序逻辑错误或者崩溃。 */ pthread_mutex_lock(&pool->mutex); LIST_INSERT(task,pool->tasks); pthread_cond_signal(&pool->cond); pthread_mutex_unlock(&pool->mutex); } //'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' //''''''''''''''''''''调试接口'''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''''' #if 1 #define THREADPOOL_INIT_COUNT 2 #define TASK_INIT_COUNT 20 void task_entry(struct nTask *task){ int idx=*(int *)task->user_data; printf("idx: %d\n",idx); free(task->user_data); free(task); } int main(){ ThreadPool pool={0}; //创建线程 nThreadPoolCreat(&pool,THREADPOOL_INIT_COUNT); printf("nThreadPoolCreat -- finish\n"); //创建任务 int i=0; for (i=0;i<TASK_INIT_COUNT;i++){ struct nTask *task=(struct nTask *)malloc(sizeof(struct nTask)); if (task==NULL){ perror("malloc"); exit(1); } memset(task,0,sizeof(struct nTask)); task->task_fun=task_entry; task->user_data=malloc(sizeof(int)); *(int *)task->user_data=i; //加入任务队列 nThreadPoolPushTask(&pool,task); } getchar(); } #endif
