线程池(C/C++版)

#include "threadpool.h" void threadpool_init(threadpool_t *pool, int threads) { pool->first = pool->tail = NULL; pool->max_threads = threads; pool->counter = 0; pool->idle = 0; pool->quit = false; } // 这其实就相当于一个消费者线程, 不断的消费任务(执行任务) void *thread_routine(void *args) { printf("thread 0x%lx is starting...\n", (unsigned long)pthread_self()); threadpool_t *pool = (threadpool_t *)args; //等待任务的到来, 然后执行任务 while (true) { bool timeout = false; pool->ready.lock(); //当处于等待的时候, 则说明空闲的线程多了一个 ++ pool->idle; //等待任务队列中有任务到来 //或者等待线程池销毁通知 while (pool->first == NULL && pool->quit == false) { printf("thread 0x%lx is waiting...\n", (unsigned long)pthread_self()); //等待两秒钟 if (0 != pool->ready.timedwait(2)) { printf("thread 0x%lx is wait timeout ...\n", (unsigned long)pthread_self()); timeout = true; //break出循环, 继续乡下执行, 会执行到 //if (timeout == true && pool->first == NULL)处 //让该语句处理等待超时的情况 break; } } //条件成熟(当等待结束), 线程开始执行任务或者是线程销毁, 则说明空闲线程又少了一个 -- pool->idle; //如果是有任务了, 则执行任务, //则重新进入循环, 进行等待 if (pool->first != NULL) { //从队头取出任务进行处理 task_t *t = pool->first; pool->first = pool->first->next; //执行任务需要一定的时间 //解锁以便于其他的生产者可以继续生产任务, 其他的消费者也可以消费任务 pool->ready.unlock(); //处理任务 t->run(t->args); free(t); //当执行完任务时, 再重新加锁 pool->ready.lock(); } //如果是等待到了线程的销毁通知, 且任务都执行完毕了 if (pool->quit == true && pool->first == NULL) { -- pool->counter; //如果没有线程了, 则给线程池发送通知 //告诉线程池, 池中已经没有线程了 if (pool->counter == 0) pool->ready.signal(); //解锁然后跳出循环 pool->ready.unlock(); break; } //如果等待超时(一般此时任务队列已经空了) if (timeout == true && pool->first == NULL) { -- pool->counter; //解锁然后跳出循环 pool->ready.unlock(); break; } //解锁 pool->ready.unlock(); } //跳出循环之后, 打印退出信息, 然后销毁线程 printf("thread 0x%lx is exiting...\n", (unsigned long)pthread_self()); pthread_exit(NULL); } //添加任务函数, 类似于一个生产者, 不断的将任务生成, 挂接到任务队列上, 等待消费者线程进行消费 void threadpool_add_task(threadpool_t *pool, void *(*run)(void *args), void *args) { // 生成任务 task_t *newTask = (task_t *)malloc(sizeof(task_t)); newTask->run = run; newTask->args = args; newTask->next = NULL; pool->ready.lock(); //将任务添加到队尾 if (pool->first == NULL) pool->first = newTask; else pool->tail->next = newTask; pool->tail = newTask; /**********************************************************/ // 如果有等待线程, 则唤醒其中一个, 让它来执行任务 if (pool->idle > 0) pool->ready.signal(); // 没有等待线程, 而且当前先线程总数尚未达到阈值, 我们就需要创建一个新的线程 else if (pool->counter < pool->max_threads) { pthread_t tid; pthread_create(&tid, NULL, thread_routine, pool); ++ pool->counter; } /**********************************************************/ pool->ready.unlock(); } void threadpool_destroy(threadpool_t *pool) { //如果已经调用过了, 则直接返回 if (pool->quit == true) return; pool->ready.lock(); pool->quit = true; if (pool->counter > 0) { //对于处于等待状态, 则给他们发送通知, //这些处于等待状态的线程, 则会接收到通知, //然后直接退出 if (pool->idle > 0) pool->ready.broadcast(); //对于正处于执行任务的线程, 他们接收不到这些通知, //则需要等待他们执行完任务 while (pool->counter > 0) pool->ready.wait(); } pool->ready.unlock(); }