C语言第2章：线程池实现

分享未结  0  780

李延 LV6 2022-05-11

悬赏：20积分

# 1 说明

通过自己写一个简单的线程池，来练习c语言的多线程。

需要用到上一章节中的queue实现。

# 2 实现代码

## 2.1 execute_pool.h

头文件 execute_pool.h

```c
#ifndef EXECUTE_POOL
#define EXECUTE_POOL

#include <pthread.h>
#include "queue.h"

typedef struct execute_work {
    void *(*work_function)(void *); //线程执行方法
    void *args;                   //参数
} execute_work_t;

typedef struct execute_pool {
    size_t shutdown;       // 线程池运行状态, 0 运行 1停止
    size_t core_num_thread; // 线程池核心运行数
    size_t run_size;     // 当前执行任务数量
    pthread_t *thread_id;     // threadsId 数组
    queue *queue_work;     // execute_work 队列
    pthread_mutex_t pool_lock;     // 线程池锁 lock
} execute_pool_t;

/**
 * 创建线程池.
 * @param core_num_thread 核心线程数
 * @param max_num_thread 队列最大等待数量
 * @return
 */
execute_pool_t *execute_create_tpool(size_t core_num_thread,size_t max_num_thread);

/**
 * 销毁线程池
 *
 * @param pool
 */
void execute_destroy_tpool(execute_pool_t *pool);

/**
 *  向线程池添加 任务
 * @param pool 线程池对象
 * @param function 需要执行的函数
 * @param args 函数参数
 * @return
 */
int execute_add_task(execute_pool_t *pool, void *(*function)(void *), void *args);

#endif
```

## 2.2 execute_pool.c

线程池的实现

```c
#include "execute_pool.h"
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>

/**
 * 执行任务
 * @param args
 * @return
 */
static void *work_run(void *args) {
    execute_pool_t *pool = (execute_pool_t *) args;

while (1) {
        execute_work_t *work = (execute_work_t *) queue_poll(pool->queue_work);
        //队列中没有任务,关闭当前线程
        if (work == NULL) {
            pthread_mutex_t lock = pool->pool_lock;
            pthread_mutex_lock(&lock);
            pool->run_size = pool->run_size - 1;
            pthread_mutex_unlock(&lock);
            pthread_exit(NULL);
            break;
        }

work->work_function(work->args);

free(work);

}
    return NULL;
}

int execute_add_task(execute_pool_t *pool, void *(*function)(void *), void *args) {

execute_work_t *work = (execute_work_t *) malloc(sizeof(execute_work_t));
    if (!work) {
        printf("in %s,malloc work error!,errno = %d,explain:%s\n", __func__, errno, strerror(errno));
        return -1;
    }

work->work_function = function;
    work->args = args;
    queue *queue = pool->queue_work;
    int ref = queue_peek(queue, work);
    if (ref) {
        return ref;
    }

pthread_mutex_t lock = pool->pool_lock;
    pthread_mutex_lock(&lock);
    size_t run_size = pool->run_size;
    size_t core_num_thread = pool->core_num_thread;

if (core_num_thread > run_size) {
        pthread_t run = pool->thread_id[run_size];
        if (pthread_create(&run, NULL, work_run, pool)) {
            printf("pthread_create failed!\n");
            exit(-1);
        }
        pool->run_size = pool->run_size + 1;
    }

pthread_mutex_unlock(&lock);

return ref;
}

execute_pool_t *execute_create_tpool(size_t core_num_thread, size_t max_num_thread) {
    execute_pool_t *pool = (execute_pool_t *) malloc(sizeof(execute_pool_t));
    if (pool == NULL) {
        printf("in %s,malloc execute_pool_t failed!,errno = %d,explain:%s\n", __func__, errno, strerror(errno));
        exit(-1);
    }
    pool->shutdown = 0;
    pool->run_size = 0;
    pool->core_num_thread = core_num_thread;

pool->queue_work = queue_create(max_num_thread);

pool->thread_id = (pthread_t *) malloc(sizeof(pthread_t) * core_num_thread);
    if (pool->thread_id == NULL) {
        exit(-1);
    }

if (pthread_mutex_init(&pool->pool_lock, NULL) != 0) {
        exit(-1);
    }
    return pool;
}

void execute_destroy_tpool(execute_pool_t *pool) {

if (pool->shutdown) {
        return;
    }
    pool->shutdown = 1;

for (int i = 0; i < pool->core_num_thread; i++) {
        pthread_join(pool->thread_id[i], NULL);
    }
    free(pool->thread_id);
    queue *queue_t = pool->queue_work;
    execute_work_t *work = (execute_work_t *) queue_poll(queue_t);
    while (work != NULL) {
        free(work);
        work = (execute_work_t *) queue_poll(queue_t);
    }

pthread_mutex_destroy(&pool->pool_lock);
    free(pool);
    queue_destroy(queue_t);
}
```

# 3 测试用例

```c
#include "execute_pool.h"
#include <unistd.h>
#include <stdio.h>

void *fun(void *args) {
    int thread = (int) args;
    printf("running the thread of %d\n", thread);
    return NULL;
}

int main(int argc, char *args[]) {
    execute_pool_t *pool = execute_create_tpool(30, 10000);

for (int i = 0; i < 10000; i++) {
        execute_add_task(pool, fun, (void *) i);
    }
    sleep(2);
    execute_destroy_tpool(pool);
    return 0;
}
```

回帖

消灭零回复