线程池原理详解及如何用C语言实现线程池( 二 ) _线程池

pool->queue_max_size = queue_max_size;
pool->shutdown = false;
/* 根据最大线程数，给工作线程数组开空间，清0 */
pool->threads = (pthread_t *)malloc(sizeof(pthread_t)*max_thr_num);
if (pool->threads == NULL)
{
printf("malloc threads false;n");
break;
}
memset(pool->threads, 0, sizeof(pthread_t)*max_thr_num);
/* 队列开空间 */
pool->task_queue =
(threadpool_task_t *)malloc(sizeof(threadpool_task_t)*queue_max_size);
if (pool->task_queue == NULL)
{
printf("malloc task queue false;n");
break;
}
/* 初始化互斥锁和条件变量 */
if ( pthread_mutex_init(&(pool->lock), NULL) != 0 ||
pthread_mutex_init(&(pool->thread_counter), NULL) !=0 ||
pthread_cond_init(&(pool->queue_not_empty), NULL) !=0 ||
pthread_cond_init(&(pool->queue_not_full), NULL) !=0)
{
printf("init lock or cond false;n");
break;
}
/* 启动min_thr_num个工作线程 */
for (i=0; i<min_thr_num; i++)
{
/* pool指向当前线程池 threadpool_thread函数在后面讲解 */
pthread_create(&(pool->threads[i]), NULL, threadpool_thread, (void *)pool);
printf("start thread 0x%x... n", (unsigned int)pool->threads[i]);
}
/* 管理者线程 admin_thread函数在后面讲解 */
pthread_create(&(pool->admin_tid), NULL, admin_thread, (void *)pool);
return pool;
} while(0);
/* 释放pool的空间 */
threadpool_free(pool);
return NULL;
}
二、线程数组线程数组实际上是在线程池初始化时开辟的一段存放一堆线程tid的空间，在逻辑上形成一个池，里面放置着提前创建的线程；这段空间中包含了正在工作的线程，等待工作的线程(空闲线程) ，等待被销毁的线程，申明但没有初始化的线程空间；

文章插图

/*工作线程*/
void *
threadpool_thread(void *threadpool)
{
threadpool_t *pool = (threadpool_t *)threadpool;
threadpool_task_t task;
while (true)
{
pthread_mutex_lock(&(pool->lock));
/* 无任务则阻塞在 “任务队列不为空” 上，有任务则跳出 */
while ((pool->queue_size == 0) && (!pool->shutdown))
{
printf("thread 0x%x is waiting n", (unsigned int)pthread_self());
pthread_cond_wait(&(pool->queue_not_empty), &(pool->lock));
/* 判断是否需要清除线程,自杀功能 */
if (pool->wait_exit_thr_num > 0)
{
pool->wait_exit_thr_num--;
/* 判断线程池中的线程数是否大于最小线程数，是则结束当前线程 */
if (pool->live_thr_num > pool->min_thr_num)
{
printf("thread 0x%x is exiting n", (unsigned int)pthread_self());
pool->live_thr_num--;
pthread_mutex_unlock(&(pool->lock));
pthread_exit(NULL);//结束线程
}
}
}
/* 线程池开关状态 */
if (pool->shutdown) //关闭线程池
{
pthread_mutex_unlock(&(pool->lock));
printf("thread 0x%x is exiting n", (unsigned int)pthread_self());
pthread_exit(NULL); //线程自己结束自己
}
//否则该线程可以拿出任务
task.function = pool->task_queue[pool->queue_front].function; //出队操作
task.arg = pool->task_queue[pool->queue_front].arg;
pool->queue_front = (pool->queue_front + 1) % pool->queue_max_size; //环型结构
pool->queue_size--;
//通知可以添加新任务
pthread_cond_broadcast(&(pool->queue_not_full));
//释放线程锁
pthread_mutex_unlock(&(pool->lock));
//执行刚才取出的任务
printf("thread 0x%x start working n", (unsigned int)pthread_self());
pthread_mutex_lock(&(pool->thread_counter)); //锁住忙线程变量
pool->busy_thr_num++;
pthread_mutex_unlock(&(pool->thread_counter));
(*(task.function))(task.arg); //执行任务
//任务结束处理
printf("thread 0x%x end working n", (unsigned int)pthread_self());
pthread_mutex_lock(&(pool->thread_counter));
pool->busy_thr_num--;
pthread_mutex_unlock(&(pool->thread_counter));
}
pthread_exit(NULL);
}