一：线程池的概念、构成、作用。

        1、概念：

                维持管理固定数量的池式结构。与此结构类似的还有：内存池、对象池、数据库连接池等。他们的共同点是复用资源。

        2、构成：

                其中有我们比较熟悉的生产消费者模型，然而关于生产消费者模型的具体应用其中就有咱们的线程池。线程池包括生产者线程、任务队列、消费者线程我们根据这三个组成部分可以分模块实现不同的功能。

        3、作用：

                1：性能优化：不过度占用核心线程，异步执行耗时任务。

                2：并发执行：充分利用多核，并发执行核心业务。

二：关于线程池结构的设计。

        由于这次使用的是C语言，没有像C++的一些STL容器可以直接使用。因此咱们通过链表来进行实现。对于这三个组成部分来说，这两个线程都需要其中的任务队列，所以我们选择从任务队列着手设计，但是队列中又包含一个一个的结点，所以最开始设计的是其中最底层的结点。

        1：我们开始思考这个任务结点怎么设计，在这个链表队列中，肯定不能缺少指向下一个元素的指针（next指针），然后这个是一个任务节点，因此这个结点还要包含我们要执行的任务（函数指针，回调函数），最后需要一些这个函数所需要的参数（arg）；

//首先要在全局声明这个回调函数的名称
typedef void (*handler_pt)(void * /* ctx */);

//通过链表来实现，这里是一个一个的任务结点
typedef struct task_s {
    void *next;           //指向下一个结点
    handler_pt func;      //任务的回调函数
    void *arg;            //所需要的参数
} task_t;

        2：设置好结点之后，开始思考队列怎么实现？都有结点了，肯定不能少了头指针和尾指针。然后还要有个变量代表阻塞和非阻塞，这个是为了当修改这个值的时候，让所有的线程退出的操作。然后就是关于互斥锁和条件变量了，这两个总是配对出现。当有了互斥锁了还需要自旋锁吗？其实不需要自旋锁了，但是为了了解自旋锁的使用方法，所以这个队列加上了自旋锁。

//任务队列
typedef struct task_queue_s {
    void *head;              //指向头节点
    void **tail;             //指向最后一个元素的next（使用二级指针可以减少代码量）
    int block;               //用来表示是不是阻塞的，默认是阻塞的，当要关闭的时候要改变为非阻塞
    spinlock_t lock;         //自旋锁：关于队列的操作可以单独加锁（不需要这个自旋锁也是可以的）
    pthread_mutex_t mutex;   //互斥锁（对于不使用自旋锁的话可以复用互斥锁）
    pthread_cond_t cond;     //条件变量：要搭配互斥锁共同工作，用来唤醒线程
} task_queue_t;

上面的二级指针可以通过下面这个图来理解。简单来说就是这个二级尾指针就是指向的是最后一个元素的next。

        3：最后设计线程池。这个线程池要使用哪个队列呢？所以要包含这个队列的指针。还要有个标志位，代表着线程的状态，当这个标志位修改的时候，代表全部线程即将退出。既然是个池，肯定不会只有一个线程，所以要包含线程的数量，以及线程指针。

//线程池的结构
struct thrdpool_s {
    task_queue_t *task_queue;       //咱们的阻塞队列
    atomic_int quit;                //原子变量：原子操作，不用单独加锁了。线程安全 
    int thrd_count;                 //线程池中线程的数量
    pthread_t *threads;             //线程池中的线程，需要开辟一块内存出来。
};

        上述三个结构都已经创建好了，那么接下来就是实现各种接口了。那么咱们可以想一下都需要什么接口呢？首先就是队列的创建吧，既然有创建的话，根据对称式设计，那么也就需要一个队列的销毁。当然在队列中还有个阻塞非阻塞的变量，咱们要通过这个变量来判断是否要退出线程是否要退出，因此要通过这个来进行广播退出的消息。对于这个队列来说，肯定有添加和删除的接口吧，然后咱们再给消费者提供一个直接拿去任务的接口。这样关于队列的一些接口就了解了。

        对于线程池的接口，肯定有线程池的创建和销毁，还有一个供生产者进行生产的函数，以及线程创建好之后，这个线程从哪个函数启动，这个函数需要我们根据业务自己去实现。全部的接口都已经说清楚。

三：各个接口的实现。

        对于每个模块的设计，都要遵循回滚式编程，一开始我们假设操作是成功的，然后给出失败的代码

        1：队列的创建。__taskqueue_create

static task_queue_t*
__taskqueue_create() {
    int ret;                                        
    task_queue_t* queue = (task_queue_t*)malloc(sizeof(task_queue_t));//开辟空间
    if (queue) {
        ret = pthread_mutex_init(&queue->mutex, NULL);      //初始化互斥锁和条件变量
        if (ret == 0) {
            ret = pthread_cond_init(&queue->cond, NULL);
            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;
}

        2：队列的销毁。__taskqueue_destroy

//队列的销毁
static void
__taskqueue_destroy(task_queue_t* queue) {
    task_t* task;
    while ((task = __pop_task(queue))) {    //将队列中全部的任务取出来并进行释放
        free(task);
    }
    spinlock_destroy(&queue->lock);          //将队列中别的资源也进行销毁
    pthread_cond_destroy(&queue->cond);
    pthread_mutex_destroy(&queue->mutex);
    free(queue);
}

        3：队列添加的操作。__add_task

static inline void
__add_task(task_queue_t* queue, void* task) {
    // 不限定任务类型，只要该任务的结构起始内存是一个用于链接下一个节点的指针
    void** link = (void**)task;     //如果你不知道这个指向下一个结点的指针叫什么（next），那就用二级指针(link)
    *link = NULL;                   //二级指针直接指向这个指向下一个节点的地址，也就是不需要这个指针的名称了

    spinlock_lock(&queue->lock);    //给队列加锁，避免多个任务同时进入队列
    *queue->tail /* 等价于 queue->tail->next */ = link;     //这里是让最后一个的指针next指向新进来的next
    queue->tail = link;                                 //然后这个是修改尾指针，指向新进来的next的指针（跟链表的插入相似）
    spinlock_unlock(&queue->lock);
    pthread_cond_signal(&queue->cond);  //唤醒线程,告诉线程我现在有任务了
}

        4：队列的删除操作。__pop_task

//这个模块没啥好说的，就单纯的取出头节点，然后删除
static inline void*
__pop_task(task_queue_t* queue) {
    spinlock_lock(&queue->lock);
    if (queue->head == NULL) {
        spinlock_unlock(&queue->lock);
        return NULL;
    }
    task_t* task;
    task = queue->head;

    void** link = (void**)task;
    queue->head = *link;

    if (queue->head == NULL) {
        queue->tail = &queue->head;
    }
    spinlock_unlock(&queue->lock);
    return task;
}

        5：消费者使用的get_task函数

        如果单纯使用上面的队列删除操作来获取任务的话，，是不安全的，因为有多个线程都在取任务，因此为了线程安全，咱们还要封装一个消费者可以安全的取出任务的操作。

//消费者使用的接口
static inline void*
__get_task(task_queue_t* queue) {
    task_t* task;
    // 虚假唤醒，这里用while循环是为了防止虚假唤醒，当被唤醒之后，要再次检查
    //为什么有虚假唤醒：可能是内部的信号导致
    while ((task = __pop_task(queue)) == NULL) {
        pthread_mutex_lock(&queue->mutex);
        if (queue->block == 0) {    //这里是当线程是非阻塞的时候，代表我们要退出线程了，不在取任务了
            pthread_mutex_unlock(&queue->mutex);
            return NULL;
        }
        // 1. 先 unlock(&mtx)
        // 2. 在 cond 休眠
        // --- __add_task 时唤醒
        // 3. 在 cond 唤醒
        // 4. 加上 lock(&mtx);                                       //一开始，队列都为空，所以会阻塞在这里，当有任务添加
        pthread_cond_wait(&queue->cond, &queue->mutex);  //这个get函数会被唤醒，因此会再走一遍这个程序，也就能取出这个任务
        pthread_mutex_unlock(&queue->mutex);

        //在pthread_cond_wait 内部是
        //1：先pthread_mutex_unlock 解锁，然后当前线程休眠.2：当被内个signal唤醒的时候，会pthread_mutex_lock 加锁
        //有人会说这里while循环那里都没有锁，怎么能线程安全呢？其实当线程全部创建以后，线程都会去判断队列有没有任务
        //刚创建的队列肯定没任务，因此都休眠在了pthread_cond_wait 这里，然后它内部最后是加锁，但是外面还有个解锁呢
    }
    return task;        //这里是当上面返回的不为NULL时。
}

        6：配合线程退出的队列操作。__nonblock

//将队列的状态修改为非阻塞状态，代表着线程和队列即将退出，不会再接收任务
static void
__nonblock(task_queue_t* queue) {
    pthread_mutex_lock(&queue->mutex);
    queue->block = 0;                            //修改为非阻塞
    pthread_mutex_unlock(&queue->mutex);
    pthread_cond_broadcast(&queue->cond);       //广播，将全部线程全部唤醒
}

        7：线程池的创建。threads_create

//复杂资源创建：对称式的接口设计
//回滚式设计
//下面是线程池的创建：创建哪个线程池，创建其中的几个线程
static int
__threads_create(thrdpool_t* pool, size_t thrd_count) {
    pthread_attr_t attr;        //线程的属性，使用之前要先初始化，用完后要去除初始化
    int ret;

    ret = pthread_attr_init(&attr);     //初始化

    if (ret == 0) {
        pool->threads = (pthread_t*)malloc(sizeof(pthread_t) * thrd_count);        //创建其中的线程
        if (pool->threads) {
            int i = 0;
            for (; i < thrd_count; i++) {
                if (pthread_create(&pool->threads[i], &attr, __thrdpool_worker, pool) != 0) {
                    break;  //其中的__thrdpool_worker  这个函数，是我们用户态要去写的函数，线程创建好之后，要走这个函数
                }
            }
            pool->thrd_count = i;
            pthread_attr_destroy(&attr);    //线程用完之后要去除初始化
            if (i == thrd_count)
                return 0;

            //如果线程未全部创建，那就全部销毁掉
            __threads_terminate(pool);
            free(pool->threads);
        }
        ret = -1;
    }
    return ret;
}


thrdpool_t*
thrdpool_create(int thrd_count) {
    thrdpool_t* pool;

    pool = (thrdpool_t*)malloc(sizeof(*pool));
    if (pool) {
        task_queue_t* queue = __taskqueue_create();    //先创建任务队列，也就是说这个任务队列属于这个线程池
        if (queue) {
            pool->task_queue = queue;
            atomic_init(&pool->quit, 0);
            if (__threads_create(pool, thrd_count) == 0)    //然后创建多个线程
                return pool;
            __taskqueue_destroy(queue);
        }
        free(pool);
    }
    return NULL;
}

        8：线程池的销毁。thrdpool_terminate

//线程的销毁（这个是运行后，告诉线程要进行退出）
void
thrdpool_terminate(thrdpool_t* pool) {
    atomic_store(&pool->quit, 1);
    __nonblock(pool->task_queue);        //修改任务队列的阻塞为非阻塞
}

        9：生产者使用的任务添加。thrdpool_post

//这里传入消费者线程池，就可以将任务添加到任务队列里面了
int
thrdpool_post(thrdpool_t* pool, handler_pt func, void* arg) {
    if (atomic_load(&pool->quit) == 1)
        return -1;
    task_t* task = (task_t*)malloc(sizeof(task_t));
    if (!task) return -1;
    task->func = func;
    task->arg = arg;
    __add_task(pool->task_queue, task);
    return 0;
}

        10：用户态函数。__thrdpool_worker

static void*
__thrdpool_worker(void* arg) {
    thrdpool_t* pool = (thrdpool_t*)arg;        //这里的arg是参数信息，在这里是将线程池的信息传递进来，在下面可以用到线程池的东西。
    task_t* task;
    void* ctx;

    while (atomic_load(&pool->quit) == 0) {
        task = (task_t*)__get_task(pool->task_queue);    //取任务
        if (!task) break;
        handler_pt func = task->func;                    //函数指针
        ctx = task->arg;
        free(task);
        func(ctx);                                        //给函数传递参数，运行这个函数
    }

    return NULL;
}

        欸？到这里我有个疑问，当线程全部退出的时候，进行广播然后线程将队列中剩余的任务全部取出来，到这个时候，我发现没有锁，管理这个任务，会发生线程不安全，后来发现在删除操作里面有自旋锁保护着呢，所以线程很安全。但还有个小疑问，上面写了好多关于消费者的函数，而关于生产者只有一个添加的操作，如果只看这些的话，我可能还是不会使用这个线程池，因此查看了例子。

constexpr int64_t n = 1000000;

void producer(thrdpool_t *pool) {
    for(int64_t i=0; i < n; ++i) {
        thrdpool_post(pool, JustTask, NULL);
    }
}

void test_thrdpool(int nproducer, int nconsumer) {
    auto pool = thrdpool_create(nconsumer);        //创建线程池并返回线程池
    for (int i=0; i<nproducer; ++i) {                //创建生产者线程，并赋予任务函数
        std::thread(&producer, pool).detach();
    }

    time_t t1 = GetTick();
    // wait for all producer done
    while (g_count.load() != n*nproducer) {
        usleep(100000);
    }

    time_t t2 = GetTick();

    std::cout << t2 << " " << t1 << " " << "used:" << t2-t1 << " exec per sec:"
        << (double)g_count.load()*1000 / (t2-t1) << std::endl;

    thrdpool_terminate(pool);
    thrdpool_waitdone(pool);
}

int main() {
    // test_thrdpool(1, 8);
    test_thrdpool(4, 4);        //四个生产者，四个消费者
    return 0;
}

到这里C语言的线程池已经讲解完毕，第一次写博客写的很垃圾，请大家见谅。https://xxetb.xetslk.com/s/2D96kH