linux c 线程池

基础概念

线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池可以有效地控制系统中并发线程的数量，避免大量线程之间的切换所带来的开销。

类型

固定大小的线程池：线程池的大小固定，适用于负载较重的服务器。
可缓存的线程池：线程池的大小不固定，可以根据需求自动调整。
单线程的线程池：只有一个线程，适用于需要顺序执行任务的场景。
定时及周期性执行任务的线程池：可以定时或周期性执行任务。

应用场景

Web服务器：处理大量的HTTP请求。
数据库连接池：管理数据库连接，提高数据库访问效率。
并发计算：在科学计算中，利用多线程加速计算过程。

示例代码（Linux C语言）

#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>

typedef struct {
    void (*function)(void *);
    void *argument;
} thread_pool_task_t;

#define MAX_THREADS 10
#define MAX_QUEUE 100

pthread_t threads[MAX_THREADS];
thread_pool_task_t task_queue[MAX_QUEUE];
int queue_head = 0, queue_tail = 0;
int is_running = 1;

void *thread_pool_worker(void *arg) {
    while (is_running) {
        if (queue_head != queue_tail) {
            thread_pool_task_t task = task_queue[queue_head++];
            task.function(task.argument);
        } else {
            sched_yield(); // 让出CPU时间片
        }
    }
    return NULL;
}

void thread_pool_init() {
    for (int i = 0; i < MAX_THREADS; ++i) {
        pthread_create(&threads[i], NULL, thread_pool_worker, NULL);
    }
}

void thread_pool_add_task(void (*function)(void *), void *argument) {
    task_queue[queue_tail].function = function;
    task_queue[queue_tail].argument = argument;
    queue_tail = (queue_tail + 1) % MAX_QUEUE;
}

void thread_pool_destroy() {
    is_running = 0;
    for (int i = 0; i < MAX_THREADS; ++i) {
        pthread_join(threads[i], NULL);
    }
}

// 示例任务函数
void example_task(void *arg) {
    int *num = (int *)arg;
    printf("Task %d is running.\n", *num);
}

int main() {
    thread_pool_init();

    for (int i = 0; i < 20; ++i) {
        int *num = malloc(sizeof(int));
        *num = i;
        thread_pool_add_task(example_task, num);
    }

    sleep(5); // 等待任务执行
    thread_pool_destroy();

    return 0;
}

遇到的问题及解决方法

问题：线程池中的任务执行出现混乱，导致结果不正确。

原因：

任务队列没有正确同步，多个线程同时读写队列。
任务执行过程中出现了竞态条件。

解决方法：

使用互斥锁（mutex）来保护任务队列的访问。
使用条件变量（condition variable）来通知工作线程有新任务到达。

pthread_mutex_t queue_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t queue_cond = PTHREAD_COND_INITIALIZER;

void *thread_pool_worker(void *arg) {
    while (is_running) {
        pthread_mutex_lock(&queue_mutex);
        while (queue_head == queue_tail && is_running) {
            pthread_cond_wait(&queue_cond, &queue_mutex);
        }
        if (!is_running) {
            pthread_mutex_unlock(&queue_mutex);
            break;
        }
        thread_pool_task_t task = task_queue[queue_head++];
        pthread_mutex_unlock(&queue_mutex);

        task.function(task.argument);
    }
    return NULL;
}

void thread_pool_add_task(void (*function)(void *), void *argument) {
    pthread_mutex_lock(&queue_mutex);
    task_queue[queue_tail].function = function;
    task_queue[queue_tail].argument = argument;
    queue_tail = (queue_tail + 1) % MAX_QUEUE;
    pthread_cond_signal(&queue_cond);
    pthread_mutex_unlock(&queue_mutex);
}

通过上述方法，可以有效避免线程池中的任务执行混乱问题。