C语言线程解池解读和实现01
在说到并发，池式组件的时候，最先想到的肯定是线程池。那线程池的原理是什么呢，又是如何工作的呢？这篇文章告诉你答案。
知识梳理
- 什么是线程池
线程池就是维护和管理一定数量线程的池式组件。有提高CPU工 作效率的作用 - 为什么需要线程池
通俗来说，如果我们有一个IO十分耗时但是我们又是单线程的那么我们的线程将会阻塞，等待这个IO执行完之后才会继续执行。这样就会十分耗时。 - 线程池有哪些组件
- 任务队列
- 一定数量的线程
- 锁（保证线程安全）
- 线程池是如歌管理线程的
- 有任务：执行任务
- 没有任务：CPU休眠
头文件解读
我们先来看一下头文件吧：
#ifndef THRDPOOL_H_
#define THRDPOOL_H_
typedef struct thrdpool_s thrdpool_t;
typedef void (*handler_pt)(void*);
#ifdef __cplusplus
extern "C"
{
#endif
thrdpool_t *thrdpool_create(int thrd_count);
void thrdpool_treminater(thrdpool_t *pool);
int thrdpool_post(thrdpool_t *pool, handler_pt func, void *arg);
void thrdpool_waitdone(thrdpool_t *pool);
#ifdef __cplusplus
}
#endif
#endif
代码解读：
作为一个组件，或者说一个库，我们并不希望别人可以看到我们内部是如何实现的，所以我们告诉用户如何使用即可。thrdpool_create用来初始化一个线程池，参数是线程数量。thrdpool_terminater用来停止线程池。thrdpool_post用来抛出任务，即在哪个线程池，执行哪个函数，参数是什么。thrdpool_waitdone检测线程是不是都执行完。
注意，虽然我们不希望用户看到我们的实现，但是要告诉用户我们库的使用规范，就是最前面的两个typedef。他告诉用户我们线程池的对象类型和传入任务的规范。
由于我们支持C++使用我们库，所以我们加上#ifdef __cplusplus extern “C”。就是说如果是C++，那我们就采用C规则编译这个文件。
数据结构解读
队列
typedef struct task_s
{
void *next;
handler_pt func;//对应函数
void *arg;//参数
} task_t;
typedef struct task_queue_s
{
void *head;//头指针
void **tail;//尾指针
int block;//标志
spinlock_t lock;//回旋锁
pthread_mutex_t mutex;//互斥锁
pthread_cond_t cond;//条件
} task_queue_t;
代码解读：
这里我们先看一下我们队列结构的示意图：
我们使用一个链式结构。将所有task连接在一起，然后有一个manager来管理这些task。看了这张图结合上面的代码就可以很清楚的了解到我们队列的结构。
池
typedef struct thrdpool_s
{
task_queue_t *task_queue;//任务队列
atomic_int quit;//标志
uint32_t thrd_count;//池内线程数量
pthread_t *threads;//线程数组
} thrdpool_t;
代码解读：
这里没有什么好说的。其中标志的意思就是：如果为0，正常运行，如果为1阻塞。它是一个原子变量，所以具有线程安全。由于就这一个变量，没有设计复杂的操作，所以没有使用锁的必要，我们用原子变量即可解决。
代码实现
初始化队列
static task_queue_t* __taskqueue_create()
{
int ret;
task_queue_t *queue = (task_queue_t*)malloc(sizeof(*queue));
if(queue)
{
ret = pthread_mutex_init(&queue->mutex);
if(ret == 0)
{
ret = pthread_cond_init(&queue->cond);
if(ret == 0)
{
spinlock_init(&queue->lock);
queue->head = NULL;
queue->tail = &queue->head;
queue->block = 1;
return queue;
}
pthread_mutex_destroy(&queue->mutex);
}
free(queue);
}
return NULL;
}
代码解读：
没有什么特别好说的，这里就是申请一块内存，然后对其中的结构体成员进行初始化。如果全部初始化成功就返回指针。主要注意的是，我们的block要初始化为0，就是默认在阻塞状态，因为刚刚初始化，队列里面没有任何任务。
移除阻塞
static int __nonblock(task_queue_t *queue)
{
pthread_mutex_lock(&queue->mutex);
queue->block = 0;
pthread_mutex_unlock(&queue->mutex);
pthread_cond_broadcast(&queue->cond);
}
代码解读：
在对task_queue内部变量操作的时候，必须使用锁来保证线程安全。将标志为阻塞的block置0，解除阻塞。然后广播，唤醒其他线程。
插入任务
static inline __add_task(task_queue_t *queue, void *task)
{
void **link = (void**)task;
*link = NULL;
spinlock_lock(&queue->lock);
*queue->tail = link;
queue->tail = link;
spinlock_unlock(&queue->lock);
pthread_cond_signal(&queue->cond);
}
代码解读：
首先，我们让task->next指向NULL。然后，在加锁的情况下，对队列进行尾插。最后唤醒一个线程获取任务。这里需要解释一下我们的写法，为什么可以这样写。事实上，在内核中，队列都是这样的一级指针头+二级指针尾的写法，原理请看图：
从指正管理长度的角度来看void*就是管理了整个node，但是**void就是管理指向内存的后面8个字节（64位操作系统）即*next；
所以*queue->tail 就是 queue->tail->next；同理*link 就是 *next。
删除任务
static task_t* __pop_task(task_queue_t *queue)
{
spinlock_lock(&queue->lock);
if(queue->head == NULL)
return NULL;
task_t *task;
void **link = (void**)queue->head;
queue->head = *link;
if(queue->head == NULL)
queue->tail = &queue->head;
spinlock_unlock(&queue->lock);
}
代码讲解：
我们将最前面的task取出，然后head指针指向后面一个。
获取任务
static void* __get_task(task_queue_t *queue)
{
task_t *task;
pthread_mutex_lock(&queue->mutex);
while(task = __pop_task(queue) == NULL)
{
if(queue->block == 0)
{
pthread_mutex_unlock(&queue->mutex);
return NULL;
}
pthread_cond_wait(&queue->cond, &queue->mutex);
pthread_mutex_unlock(&queue->mutex);
}
return task;
}
代码解读：
我们调用上面写的__pop_task来获取队列最前面的task。如果队列是非阻塞的，那就直接返回。如果队列是阻塞的（刚刚初始化完），那我们就要进入休眠，等待__add_task里面的唤醒。然后返回获取到的task。
队列的销毁
static void __destroy_task_queue(task_queue_t *queue)
{
task_t *task;
while(task = __pop_task(queue))
{
free(task);
}
pthread_mutex_destroy(&queue->mutex);
pthread_cond_destroy(&queue->cond);
spinlock_destroy(&queue->lock);
free(queue);
}
代码解读：
释放全部资源销毁即可。
线程池工作
static void* __thrdpool_work(void *arg)
{
thrdpool_t *thrdpool = (thrdpool_t*)arg;
task_t *task;
void *cxt;
if(atomic_load(&thrdpool->quit) == 0)
{
task = (task_t*)__get_task(thrdpool->task_queue);
handler_pt func = task->func;
cxt = task->arg;
free(task);
func(cxt);
}
return NULL;
}
代码解读：
线程停止
static void __thrdpool_terminater(thrdpool_t *pool)
{
atomic_store(&pool->quit, 1);
__nonblock(pool->task_queue);
int i = 0;
for(i; i < pool->thrd_count; i++)
{
pthread_join(pool->threads[i], NULL);
}
}
代码解读：
首先让线程池阻塞，然后执行完当前所有线程的任务。
创建线程池
static int __thrdpool_create(thrdpool_t *pool, int thrd_num)
{
int ret;
pthread_attr_t attr;
ret = pthread_attr_init(&attr);
if(ret == 0)
{
pool->threads = (pthread_t*)malloc(sizeof(pthread_t) * thrd_num);
if(pool->threads)
{
int i = 0;
for(i; i < thrd_num; i++)
{
if(pthread_create(&pool->threads[i], &attr, __thrdpool_work, pool) != 0)
{
break;
}
}
pool->thrd_count = i;
if(thrd_num == i)
return 0;
__thrdpool_terminater(pool);
free(pool->threads);
}
ret = -1;
}
return ret;
}
代码解读：
主要就是在堆上开辟空间，然后使用循环批量创建线程。就是要注意一点一点来，如果资源创建失败需要及时销毁资源并且返回。
接口
void thrdpool_treminater(thrdpool_t *pool)
{
atomic_store(&pool->quit, 1);
__nonblock(pool->task_queue);
}
thrdpool_t *thrdpool_create(int thrd_count)
{
thrdpool_t *pool = (thrdpool_t*)malloc(sizeof(thrdpool_t));
if(pool)
{
task_queue_t *task = __taskqueue_create();
if(task)
{
pool->task_queue = task;
int ret = __thrdpool_create(pool, thrd_count);
if(ret == 0)
{
return pool;
}
__destroy_task_queue(pool->task_queue);
}
free(pool);
}
return NULL;
}
int thrdpool_post(thrdpool_t *pool, handler_pt func, void *arg)
{
task_t *task = (task_t*)malloc(sizeof(task_t));
if(atomic_load(&pool->quit) == 1)
{
return -1;
}
task->arg = arg;
task->func = func;
__add_task(pool->task_queue, task);
return 0;
}
void thrdpool_waitdone(thrdpool_t *pool)
{
int i = 0;
for(i; i < pool->thrd_count; i++)
{
pthread_join(pool->threads[i], NULL);
}
__destroy_task_queue(pool->task_queue);
free(pool->threads);
free(pool);
}
代码解读：
这里的接口是提供给客户使用的。这里需要注意waitdone和terminater的区别。一个是完全停止线程池（带有销毁功能），一个是不销毁，只是暂停。
