#include #include #include #include #include #include #include #include #include #define MAX_THREAD 24 #define LL

您的代码片段定义了一个简单的线程池和工作队列结构。下面我将为您提供一些关于这个线程池的初始化和实现的建议，以及如何使用它。

代码分析

数据结构:
- worker_t: 表示一个工作线程，包含线程标识符、终止标志、工作队列指针以及双向链表节点指针。
- job_t: 表示一个待处理的作业，包含作业函数指针、用户数据以及链表节点指针。
- workqueue_t: 包含工人列表（线程）、等待的作业列表以及用于保护作业队列和条件变量的锁。
宏定义:
- LL_ADD 和 LL_REMOVE 用于操作双向链表，以便在工作者和作业之间添加或删除元素。
初始化函数: 您声明了一个 threadpool_init 函数，但没有提供具体实现。

实现初始化和其他功能

接下来是一些需要实现的关键功能，包括初始化工作队列、添加任务、启动工作线程等。

1. 初始化工作队列

int workqueue_init(workqueue_t *workqueue, int numWorkers) {
    if (pthread_mutex_init(&workqueue->jobs_mutex, NULL) != 0) {
        return -1;
    }
    
    if (pthread_cond_init(&workqueue->jobs_cond, NULL) != 0) {
        pthread_mutex_destroy(&workqueue->jobs_mutex);
        return -1;
    }

    workqueue->workers = NULL;
    workqueue->waiting_jobs = NULL;

    for (int i = 0; i < numWorkers; ++i) {
        worker_t *worker = malloc(sizeof(worker_t));
        if (!worker) {
            // 清理已分配资源
            workqueue_shutdown(workqueue);
            return -1;
        }

        worker->terminate = 0;
        worker->workqueue = workqueue;

        pthread_create(&worker->thread, NULL, worker_function, (void *)worker);
        
        LL_ADD(worker, workqueue->workers); // 将新创建的工人加入链表
    }

    return 0;
}

2. 工作函数

这是每个工人在自己的线程中执行的函数：

static void *worker_function(void *ptr) {
    worker_t *worker = (worker_t *)ptr;

    while (1) {
        job_t *job;

        // 锁定并检查等待任务
        pthread_mutex_lock(&worker->workqueue->jobs_mutex);

        while (!worker->workqueue->waiting_jobs && !worker->terminate) {
            pthread_cond_wait(&worker->workqueue->jobs_cond, &worker->workqueue->jobs_mutex);
        }

        if (worker->terminate) {
            pthread_mutex_unlock(&worker->workqueue->jobs_mutex);
            break; // 如果接到终止信号则退出循环
        }

        // 从等待队列中取出一个作业
        job = worker->workqueue->waiting_jobs;
        
        LL_REMOVE(job, worker->workqueue->waiting_jobs); // 从任务列表中移除
        
        pthread_mutex_unlock(&worker->workqueue->jobs_mutex);

        job_function(job); // 执行作业函数（请确保已定义 job_function）
        
        free(job); // 清理分配给作业的数据结构
    }
    
    return NULL;
}

3. 初始化函数

extern "C"{
void threadpool_init(void)
{
    memset(&workqueue, 0, sizeof(workqueue)); // 清空整个结构体内存
    
    if (workqueue_init(&workqueue, MAX_THREAD) != 0) {
       fprintf(stderr, "Failed to initialize the thread pool.\n");
       exit(EXIT_FAILURE);
   }
}
}