三.线程池

        熟悉QT的朋友应该知道线程池的概念QThreadPool，它是QT自定义的线程池类，用于管理多线程并发业务场景。

        在前文中也提到，线程池也可以用于进行异步派发事件，触发多条线程的消息响应函数。

        在应用中经常会遇到这样一种场景，我不需要进行复杂的操作，也不需要搞复杂的订阅发布，我只是单纯的把一个简单操作重复很多次。

        这个时候就需要用到线程池来对并发操作进行管理，理论上你当然也可以不用线程池，每当需要进行一次操作的时候新建一条线程，代价是功耗飙升CPU占用率+++++。

        所以还是需要这样一个pool，来帮助快速降低和减少性能损耗。

        那么，一个典型的线程池应该是怎么样的呢。

1.定义

        线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小，以默认的优先级运行，并处于多线程单元中。   

        如果某个线程在托管代码中空闲（如正在等待某个事件）, 则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙，但队列中包含挂起的工作，则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队，但他们要等到其他线程完成后才启动。

       简单解析这段话，线程池应该是由这样几个东西组成：

• 任务队列

        需要进行的任务将被添加到队列，线程被创建后会启动并执行任务。

        队列的长度有限，超出任务队列处理能力的任务会被挂机排队，等到之前执行任务的线程完成任务释放资源后才能进行新任务的处理。同理，已执行的任务会从任务队列里删除。

• 工作线程

        就像食堂打饭一样，任务队列里充满了任务，而工作的线程们不断读取任务（打饭）进行执行（吃饭），执行完毕后释放资源，然后重新获取任务（打饭）。

        没有任务待执行怎么办？工作线程会被阻塞挂起。（等阿姨上菜）

        当任务队列里添加任务之后，工作线程会解除阻塞，重新开始工作（打饭吃饭）。

• 管理线程

        相当于食堂主管，线程池创建时，创建工作线程组合。当线程工作时，管理调度线程与任务。当挂起线程数目过多时，删除一部分线程以释放资源。

// 任务
typedef struct Task
{void (*function)(void* arg);void* arg;
}Task;                                                                                // 线程池结构体
struct ThreadPool
{// 任务队列Task* myTask;int queueCapacity;  // 容量int queueSize;      // 当前任务个数int queueFront;     // 队头 -> 取数据int queueRear;      // 队尾 -> 放数据pthread_t managerID;    // 管理者线程IDpthread_t *threadIDs;   // 工作的线程IDint minNum;             // 最小线程数量int maxNum;             // 最大线程数量int busyNum;            // 忙的线程的个数int liveNum;            // 存活的线程的个数int exitNum;            // 要销毁的线程个数pthread_mutex_t mutexPool;  // 锁整个的线程池pthread_mutex_t mutexBusy;  // 锁busyNum变量pthread_cond_t notFull;     // 任务队列是不是满了pthread_cond_t notEmpty;    // 任务队列是不是空了bull shutdown;           // 是不是要销毁线程池, 销毁为true, 不销毁为false
};

2.实现

        首先，我们需要写一个自定义线程池的头文件。

#ifndef _THREADPOOL_H
#define _THREADPOOL_H
//刚刚已经定义了线程池结构体
typedef struct ThreadPool ThreadPool;
// 创建线程池并初始化
ThreadPool *threadPoolCreate(int min, int max, int queueSize);// 销毁线程池
int threadPoolDestroy(ThreadPool* pool);// 给线程池添加任务
void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg);// 获取线程池中工作的线程的个数
int threadPoolBusyNum(ThreadPool* pool);// 获取线程池中活着的线程的个数
int threadPoolAliveNum(ThreadPool* pool);//
// 工作的线程(消费者线程)任务函数
void* worker(void* arg);
// 管理者线程任务函数
void* manager(void* arg);
// 单个线程退出
void threadExit(ThreadPool* pool);
#endif  // _THREADPOOL_H

        然后在CPP里进行实现。

#ifndef _THREADPOOL_H
#define _THREADPOOL_H
//刚刚已经定义了线程池结构体
typedef struct ThreadPool ThreadPool;
// 创建线程池并初始化
ThreadPool *threadPoolCreate(int min, int max, int queueSize);// 销毁线程池
int threadPoolDestroy(ThreadPool* pool);// 给线程池添加任务
void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg);// 获取线程池中工作的线程的个数
int threadPoolBusyNum(ThreadPool* pool);// 获取线程池中活着的线程的个数
int threadPoolAliveNum(ThreadPool* pool);//
// 工作的线程(消费者线程)任务函数
void* worker(void* arg);
// 管理者线程任务函数
void* manager(void* arg);
// 单个线程退出
void threadExit(ThreadPool* pool);
#endif  // _THREADPOOL_HThreadPool* threadPoolCreate(int min, int max, int queueSize)
{//创建一个线程池并分配空间ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));do{if (pool == NULL){//分配内存失败则直接退出printf("malloc threadpool fail...\n");break;}pool->threadIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);if (pool->threadIDs == NULL){//分配线程内存失败则直接退出printf("malloc threadIDs fail...\n");break;}memset(pool->threadIDs, 0, sizeof(pthread_t) * max);pool->minNum = min;pool->maxNum = max;pool->busyNum = 0;pool->liveNum = min;    // 和最小个数相等pool->exitNum = 0;if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||pthread_cond_init(&pool->notEmpty, NULL) != 0 ||pthread_cond_init(&pool->notFull, NULL) != 0){//获取线程权限失败则退出printf("mutex or condition init fail...\n");break;}// 任务队列pool->myTask = (Task*)malloc(sizeof(Task) * queueSize);pool->queueCapacity = queueSize;pool->queueSize = 0;pool->queueFront = 0;pool->queueRear = 0;pool->shutdown = false;// 创建线程pthread_create(&pool->managerID, NULL, manager, pool);for (int i = 0; i < min; ++i){pthread_create(&pool->threadIDs[i], NULL, worker, pool);}return pool;} while (0);// 释放资源if (pool && pool->threadIDs) free(pool->threadIDs);if (pool && pool->taskQ) free(pool->taskQ);if (pool) free(pool);return NULL;
}int threadPoolDestroy(ThreadPool* pool)
{if (pool == NULL){return -1;}// 关闭线程池pool->shutdown = 1;// 阻塞回收管理者线程pthread_join(pool->managerID, NULL);// 唤醒阻塞的消费者线程for (int i = 0; i < pool->liveNum; ++i){pthread_cond_signal(&pool->notEmpty);}// 释放堆内存if (pool->taskQ){free(pool->taskQ);}if (pool->threadIDs){free(pool->threadIDs);}pthread_mutex_destroy(&pool->mutexPool);pthread_mutex_destroy(&pool->mutexBusy);pthread_cond_destroy(&pool->notEmpty);pthread_cond_destroy(&pool->notFull);free(pool);pool = NULL;return 0;
}void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg)
{pthread_mutex_lock(&pool->mutexPool);while (pool->queueSize == pool->queueCapacity && !pool->shutdown){// 阻塞生产者线程pthread_cond_wait(&pool->notFull, &pool->mutexPool);}if (pool->shutdown){pthread_mutex_unlock(&pool->mutexPool);return;}// 添加任务pool->taskQ[pool->queueRear].function = func;pool->taskQ[pool->queueRear].arg = arg;pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity;pool->queueSize++;pthread_cond_signal(&pool->notEmpty);pthread_mutex_unlock(&pool->mutexPool);
}int threadPoolBusyNum(ThreadPool* pool)
{pthread_mutex_lock(&pool->mutexBusy);int busyNum = pool->busyNum;pthread_mutex_unlock(&pool->mutexBusy);return busyNum;
}int threadPoolAliveNum(ThreadPool* pool)
{pthread_mutex_lock(&pool->mutexPool);int aliveNum = pool->liveNum;pthread_mutex_unlock(&pool->mutexPool);return aliveNum;
}void* worker(void* arg)
{ThreadPool* pool = (ThreadPool*)arg;while (1){pthread_mutex_lock(&pool->mutexPool);// 当前任务队列是否为空while (pool->queueSize == 0 && !pool->shutdown){// 阻塞工作线程pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);// 判断是不是要销毁线程if (pool->exitNum > 0){pool->exitNum--;if (pool->liveNum > pool->minNum){pool->liveNum--;pthread_mutex_unlock(&pool->mutexPool);threadExit(pool);}}}// 判断线程池是否被关闭了if (pool->shutdown){pthread_mutex_unlock(&pool->mutexPool);threadExit(pool);}// 从任务队列中取出一个任务Task task;task.function = pool->taskQ[pool->queueFront].function;task.arg = pool->taskQ[pool->queueFront].arg;// 移动头结点pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity;pool->queueSize--;// 解锁pthread_cond_signal(&pool->notFull);pthread_mutex_unlock(&pool->mutexPool);printf("thread %ld start working...\n", pthread_self());pthread_mutex_lock(&pool->mutexBusy);pool->busyNum++;pthread_mutex_unlock(&pool->mutexBusy);task.function(task.arg);free(task.arg);task.arg = NULL;printf("thread %ld end working...\n", pthread_self());pthread_mutex_lock(&pool->mutexBusy);pool->busyNum--;pthread_mutex_unlock(&pool->mutexBusy);}return NULL;
}void* manager(void* arg)
{ThreadPool* pool = (ThreadPool*)arg;while (!pool->shutdown){// 每隔3s检测一次sleep(3);// 取出线程池中任务的数量和当前线程的数量pthread_mutex_lock(&pool->mutexPool);int queueSize = pool->queueSize;int liveNum = pool->liveNum;pthread_mutex_unlock(&pool->mutexPool);// 取出忙的线程的数量pthread_mutex_lock(&pool->mutexBusy);int busyNum = pool->busyNum;pthread_mutex_unlock(&pool->mutexBusy);// 添加线程// 任务的个数>存活的线程个数 && 存活的线程数<最大线程数if (queueSize > liveNum && liveNum < pool->maxNum){pthread_mutex_lock(&pool->mutexPool);int counter = 0;for (int i = 0; i < pool->maxNum && counter < NUMBER&& pool->liveNum < pool->maxNum; ++i){if (pool->threadIDs[i] == 0){pthread_create(&pool->threadIDs[i], NULL, worker, pool);counter++;pool->liveNum++;}}pthread_mutex_unlock(&pool->mutexPool);}// 销毁线程// 忙的线程*2 < 存活的线程数 && 存活的线程>最小线程数if (busyNum * 2 < liveNum && liveNum > pool->minNum){pthread_mutex_lock(&pool->mutexPool);pool->exitNum = NUMBER;pthread_mutex_unlock(&pool->mutexPool);// 让工作的线程自杀for (int i = 0; i < NUMBER; ++i){pthread_cond_signal(&pool->notEmpty);}}}return NULL;
}void threadExit(ThreadPool* pool)
{pthread_t tid = pthread_self();for (int i = 0; i < pool->maxNum; ++i){if (pool->threadIDs[i] == tid){pool->threadIDs[i] = 0;printf("threadExit() called, %ld exiting...\n", tid);break;}}pthread_exit(NULL);
}

总结

        思绪比较发散，因为是两天写出来的东西。

        总之，本文主要是两个方向的内容，一块介绍了观察者模式和常见的多线程同步异步处理设计架构，一块介绍了线程池的底层原理及自我实现。

        在实际的生产过程中，根据自己的业务需求，对于多线程的操作可以继承语言本身提供的多线程管理类EventBus或者ThreadPool，也可以自己手写去进行实现。在了解了底层原理后，对于实际应用中的常见场景和问题，会处理得更得心应手。

        多线程是操作系统里的核心部分，想要学好操作系统，手撕源码是必经之路。