面试题：C语言线程池实现基础

Question

请简述在Linux环境下，使用C语言实现线程池的基本步骤，包括线程池的数据结构设计、线程的创建与管理等方面。

Answer 1

1. 数据结构设计

1. 任务结构体：
   • 定义一个结构体来表示任务。每个任务通常包含一个函数指针（指向具体要执行的任务函数）和该函数所需的参数。

   typedef struct Task {
       void (*func)(void*);
       void* arg;
       struct Task* next;
   } Task;
   

2. 线程池结构体：
   • 包含任务队列，用于存储待执行的任务。
   • 线程数组，存放线程ID。
   • 线程数量、任务队列最大容量、当前任务数量等信息。
   • 还需要一些同步机制相关的变量，如互斥锁用于保护任务队列的访问，条件变量用于线程间的同步。

   typedef struct ThreadPool {
       Task* head;
       Task* tail;
       pthread_t* threads;
       int thread_count;
       int queue_max_size;
       int queue_size;
       pthread_mutex_t mutex;
       pthread_cond_t cond;
       int shutdown;
   } ThreadPool;
   

2. 线程池初始化

1. 分配内存：
   • 为线程池结构体分配内存。

   ThreadPool* createThreadPool(int thread_count, int queue_max_size) {
       ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
       if (!pool) {
           return NULL;
       }
       pool->thread_count = thread_count;
       pool->queue_max_size = queue_max_size;
       pool->queue_size = 0;
       pool->head = pool->tail = NULL;
       pool->shutdown = 0;
   

2. 初始化同步机制：
   • 初始化互斥锁和条件变量。

       if (pthread_mutex_init(&pool->mutex, NULL) != 0 ||
           pthread_cond_init(&pool->cond, NULL) != 0) {
           free(pool);
           return NULL;
       }
   

3. 创建线程：
   • 为线程数组分配内存，并创建指定数量的线程。每个线程将执行线程池的工作函数。

       pool->threads = (pthread_t*)malloc(thread_count * sizeof(pthread_t));
       if (!pool->threads) {
           pthread_mutex_destroy(&pool->mutex);
           pthread_cond_destroy(&pool->cond);
           free(pool);
           return NULL;
       }
       for (int i = 0; i < thread_count; ++i) {
           if (pthread_create(&pool->threads[i], NULL, worker, (void*)pool) != 0) {
               for (int j = 0; j < i; ++j) {
                   pthread_cancel(pool->threads[j]);
               }
               pthread_mutex_destroy(&pool->mutex);
               pthread_cond_destroy(&pool->cond);
               free(pool->threads);
               free(pool);
               return NULL;
           }
       }
       return pool;
   }
   

3. 线程工作函数

1. 获取任务并执行：
   • 线程从任务队列中获取任务，执行任务函数，然后等待新的任务。

   void* worker(void* arg) {
       ThreadPool* pool = (ThreadPool*)arg;
       while (1) {
           pthread_mutex_lock(&pool->mutex);
           while (pool->queue_size == 0 &&!pool->shutdown) {
               pthread_cond_wait(&pool->cond, &pool->mutex);
           }
           if (pool->shutdown && pool->queue_size == 0) {
               pthread_mutex_unlock(&pool->mutex);
               pthread_exit(NULL);
           }
           Task* task = pool->head;
           pool->head = task->next;
           if (pool->queue_size == 1) {
               pool->tail = NULL;
           }
           pool->queue_size--;
           pthread_mutex_unlock(&pool->mutex);
           (*task->func)(task->arg);
           free(task);
       }
       return NULL;
   }
   

4. 添加任务到线程池

1. 分配任务内存并插入队列：
   • 创建新的任务结构体，将任务插入任务队列。

   int addTask(ThreadPool* pool, void (*func)(void*), void* arg) {
       pthread_mutex_lock(&pool->mutex);
       if (pool->shutdown || pool->queue_size >= pool->queue_max_size) {
           pthread_mutex_unlock(&pool->mutex);
           return -1;
       }
       Task* task = (Task*)malloc(sizeof(Task));
       if (!task) {
           pthread_mutex_unlock(&pool->mutex);
           return -1;
       }
       task->func = func;
       task->arg = arg;
       task->next = NULL;
       if (pool->tail == NULL) {
           pool->head = pool->tail = task;
       } else {
           pool->tail->next = task;
           pool->tail = task;
       }
       pool->queue_size++;
       pthread_cond_signal(&pool->cond);
       pthread_mutex_unlock(&pool->mutex);
       return 0;
   }
   

5. 销毁线程池

1. 设置关闭标志并唤醒所有线程：
   • 设置线程池的关闭标志，唤醒所有等待的线程。

   void destroyThreadPool(ThreadPool* pool) {
       pthread_mutex_lock(&pool->mutex);
       pool->shutdown = 1;
       pthread_cond_broadcast(&pool->cond);
       pthread_mutex_unlock(&pool->mutex);
       for (int i = 0; i < pool->thread_count; ++i) {
           pthread_join(pool->threads[i], NULL);
       }
       while (pool->head != NULL) {
           Task* task = pool->head;
           pool->head = task->next;
           free(task);
       }
       free(pool->threads);
       pthread_mutex_destroy(&pool->mutex);
       pthread_cond_destroy(&pool->cond);
       free(pool);
   }