宿州大型网站建设公司,网站网站设计公司,wordpress获取所有分类,深圳摇号申请注册点击蓝字关注我们1、线程池原理我们使用线程的时候就去创建一个线程#xff0c;这样实现起来非常简便#xff0c;但是就会有一个问题#xff1a;如果并发的线程数量很多#xff0c;并且每个线程都是执行一个时间很短的任务就结束了#xff0c;这样频繁创建线程就会大大降低…点击蓝字关注我们1、线程池原理我们使用线程的时候就去创建一个线程这样实现起来非常简便但是就会有一个问题如果并发的线程数量很多并且每个线程都是执行一个时间很短的任务就结束了这样频繁创建线程就会大大降低系统的效率因为频繁创建线程和销毁线程需要时间。那么有没有一种办法使得线程可以复用就是执行完一个任务并不被销毁而是可以继续执行其他的任务呢线程池是一种多线程处理形式处理过程中将任务添加到队列然后在创建线程后自动启动这些任务。线程池线程都是后台线程。每个线程都使用默认的堆栈大小以默认的优先级运行并处于多线程单元中。如果某个线程在托管代码中空闲如正在等待某个事件, 则线程池将插入另一个辅助线程来使所有处理器保持繁忙。如果所有线程池线程都始终保持繁忙但队列中包含挂起的工作则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值。超过最大值的线程可以排队但他们要等到其他线程完成后才启动。在各个编程语言的语种中都有线程池的概念并且很多语言中直接提供了线程池作为程序猿直接使用就可以了下面给大家介绍一下线程池的实现原理线程池的组成主要分为 3 个部分这三部分配合工作就可以得到一个完整的线程池1任务队列存储需要处理的任务由工作的线程来处理这些任务通过线程池提供的 API 函数将一个待处理的任务添加到任务队列或者从任务队列中删除已处理的任务会被从任务队列中删除线程池的使用者也就是调用线程池函数往任务队列中添加任务的线程就是生产者线程2工作的线程任务队列任务的消费者 N个线程池中维护了一定数量的工作线程他们的作用是是不停的读任务队列从里边取出任务并处理工作的线程相当于是任务队列的消费者角色如果任务队列为空工作的线程将会被阻塞 (使用条件变量 / 信号量阻塞)如果阻塞之后有了新的任务由生产者将阻塞解除工作线程开始工作3管理者线程不处理任务队列中的任务1个它的任务是周期性的对任务队列中的任务数量以及处于忙状态的工作线程个数进行检测当任务过多的时候可以适当的创建一些新的工作线程当任务过少的时候可以适当的销毁一些工作的线程2、任务队列// 任务结构体
typedef struct Task
{void (*function)(void* arg);void* arg;
}Task;3. 线程池定义// 线程池结构体
struct ThreadPool
{// 任务队列Task* taskQ;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; // 任务队列是不是空了int shutdown; // 是不是要销毁线程池, 销毁为1, 不销毁为0
};4、头文件声明#ifndef _THREADPOOL_H
#define _THREADPOOL_Htypedef 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_H5、源文件定义ThreadPool* 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-taskQ (Task*)malloc(sizeof(Task) * queueSize);pool-queueCapacity queueSize;pool-queueSize 0;pool-queueFront 0;pool-queueRear 0;pool-shutdown 0;// 创建线程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);
}6. 测试代码void taskFunc(void* arg)
{int num *(int*)arg;printf(thread %ld is working, number %d\n,pthread_self(), num);sleep(1);
}int main()
{// 创建线程池ThreadPool* pool threadPoolCreate(3, 10, 100);for (int i 0; i 100; i){int* num (int*)malloc(sizeof(int));*num i 100;threadPoolAdd(pool, taskFunc, num);}sleep(30);threadPoolDestroy(pool);return 0;
}*声明本文于网络整理版权归原作者所有如来源信息有误或侵犯权益请联系我们删除或授权事宜。戳“阅读原文”我们一起进步
