mysql线程池源码分析

// 线程池结构体定义
typedef struct {
    pthread_t *threads;          // 线程数组
    int thread_count;            // 线程数量
    pthread_mutex_t lock;        // 互斥锁
    pthread_cond_t cond;         // 条件变量
    Queue *task_queue;           // 任务队列
    int shutdown;                // 关闭标志
} ThreadPool;

// 初始化线程池
ThreadPool* init_thread_pool(int thread_count) {
    ThreadPool *pool = (ThreadPool *)malloc(sizeof(ThreadPool));
    pool->threads = (pthread_t *)malloc(sizeof(pthread_t) * thread_count);
    pool->thread_count = thread_count;
    pthread_mutex_init(&pool->lock, NULL);
    pthread_cond_init(&pool->cond, NULL);
    pool->task_queue = create_queue();
    pool->shutdown = 0;
    for (int i = 0; i < thread_count; i++) {
        pthread_create(&pool->threads[i], NULL, worker_thread, pool);
    }
    return pool;
}

// 工作线程函数
void* worker_thread(void *arg) {
    ThreadPool *pool = (ThreadPool *)arg;
    while (1) {
        pthread_mutex_lock(&pool->lock);
        while (queue_is_empty(pool->task_queue) && !pool->shutdown) {
            pthread_cond_wait(&pool->cond, &pool->lock);
        }
        if (pool->shutdown) {
            pthread_mutex_unlock(&pool->lock);
            break;
        }
        Task *task = queue_pop(pool->task_queue);
        pthread_mutex_unlock(&pool->lock);
        execute_task(task);
        free_task(task);
    }
    return NULL;
}

// 添加任务到线程池
void add_task(ThreadPool *pool, Task *task) {
    pthread_mutex_lock(&pool->lock);
    queue_push(pool->task_queue, task);
    pthread_cond_signal(&pool->cond);
    pthread_mutex_unlock(&pool->lock);
}

// 销毁线程池
void destroy_thread_pool(ThreadPool *pool) {
    pool->shutdown = 1;
    pthread_cond_broadcast(&pool->cond);
    for (int i = 0; i < pool->thread_count; i++) {
        pthread_join(pool->threads[i], NULL);
    }
    free(pool->threads);
    destroy_queue(pool->task_queue);
    free(pool);
}