赞
踩
- #include <stdio.h>
- #include <stdlib.h>
- #include <pthread.h>
-
- // 线程池中最大线程数
- #define MAX_THREADS 10
-
- // 任务结构体,包含任务执行函数和参数
- typedef struct {
- void *(*task)(void *);
- void *arg;
- } task_t;
-
- // 线程池结构体,包含互斥锁、条件变量、任务队列等
- typedef struct {
- pthread_mutex_t lock; // 线程池互斥锁
- pthread_cond_t notify; // 条件变量,用于通知空闲线程有新任务可做
- pthread_t threads[MAX_THREADS]; // 线程池中的线程
- task_t *tasks; // 任务队列
- int num_threads; // 当前线程池中的线程数
- int max_tasks; // 任务队列中最多允许的任务数
- int head; // 队列头指针
- int tail; // 队列尾指针
- int count; // 当前队列中的任务数量
- int shutdown; // 线程池是否关闭标志
- } thread_pool_t;
-
- // 初始化线程池
- int thread_pool_init(thread_pool_t *pool, int num_threads, int max_tasks) {
- pool->num_threads = 0;
- pool->max_tasks = max_tasks;
- pool->head = 0;
- pool->tail = 0;
- pool->count = 0;
- pool->shutdown = 0;
-
- // 初始化互斥锁和条件变量
- if (pthread_mutex_init(&(pool->lock), NULL) != 0 ||
- pthread_cond_init(&(pool->notify), NULL) != 0) {
- return -1;
- }
-
- // 创建指定数量的线程
- for (int i = 0; i < num_threads; i++) {
- if (pthread_create(&(pool->threads[i]), NULL, thread_pool_worker, (void *)pool) != 0) {
- return -1;
- }
- pool->num_threads++;
- }
-
- // 创建任务队列
- pool->tasks = (task_t *)malloc(max_tasks * sizeof(task_t));
- if (pool->tasks == NULL) {
- return -1;
- }
-
- return 0;
- }
-
- // 向线程池中添加新任务
- int thread_pool_add_task(thread_pool_t *pool, void *(*task)(void *), void *arg) {
- // 加锁,保证线程安全
- pthread_mutex_lock(&(pool->lock));
-
- // 检查任务队列是否已满
- if (pool->count == pool->max_tasks || pool->shutdown) {
- pthread_mutex_unlock(&(pool->lock));
- return -1;
- }
-
- // 将任务加入队列并更新计数器
- task_t new_task = {task, arg};
- pool->tasks[pool->tail] = new_task;
- pool->tail = (pool->tail + 1) % pool->max_tasks;
- pool->count++;
-
- // 通知空闲线程有新任务可做
- pthread_cond_signal(&(pool->notify));
-
- // 解锁
- pthread_mutex_unlock(&(pool->lock));
-
- return 0;
- }
-
- // 销毁线程池
- int thread_pool_destroy(thread_pool_t *pool) {
- pool->shutdown = 1;
-
- // 等待所有任务执行完毕
- while (pool->count > 0) {
- sleep(1);
- }
-
- // 终止所有线程并释放资源
- for (int i = 0; i < pool->num_threads; i++) {
- pthread_cancel(pool->threads[i]);
- pthread_join(pool->threads[i], NULL);
- }
- free(pool->tasks);
- pthread_mutex_destroy(&(pool->lock));
- pthread_cond_destroy(&(pool->notify));
-
- return 0;
- }
-
- // 线程池的工作线程,从任务队列中取出一个任务并执行
- void *thread_pool_worker(void *arg) {
- thread_pool_t *pool = (thread_pool_t *)arg;
-
- while (1) {
- // 加锁,保证线程安全
- pthread_mutex_lock(&(pool->lock));
-
- // 如果任务队列为空,则等待通知有新任务可做
- while (pool->count == 0 && !pool->shutdown) {
- pthread_cond_wait(&(pool->notify), &(pool->lock));
- }
-
- // 如果线程池已关闭,则退出循环结束线程
- if (pool->shutdown) {
- pthread_mutex_unlock(&(pool->lock));
- pthread_exit(NULL);
- }
-
- // 取出一个任务并更新计数器
- task_t task = pool->tasks[pool->head];
- pool->head = (pool->head + 1) % pool->max_tasks;
- pool->count--;
-
- // 解锁
- pthread_mutex_unlock(&(pool->lock));
-
- // 执行任务
- (*(task.task))(task.arg);
- }
-
- return NULL;
- }
-
- // 示例任务执行函数
- void *example_task(void *arg) {
- int num = *((int *)arg);
- printf("Task #%d started.\n", num);
- sleep(1); // 模拟任务执行过程
- printf("Task #%d ended.\n", num);
- return NULL;
- }
-
- // 测试代码
- int main() {
- thread_pool_t pool;
-
- // 初始化线程池
- if (thread_pool_init(&pool, MAX_THREADS, 10) != 0) {
- printf("Error initializing thread pool.\n");
- return -1;
- }
-
- // 向线程池中添加多个任务
- for (int i = 0; i < 15; i++) {
- int *num = (int *)malloc(sizeof(int));
- *num = i;
- thread_pool_add_task(&pool, example_task, num);
- }
-
- // 等待任务执行完毕并销毁线程池
- thread_pool_destroy(&pool);
-
- return 0;
- }
上述代码中,先定义了一个任务结构体和一个线程池结构体,分别用于存储任务的执行函数和参数,以及线程池中的相关信息。在初始化线程池时,会创建指定数量的线程,并将其加入到线程池中,并创建一个任务队列。添加任务时,需要使用加锁操作保证多线程安全,并将任务加入到队列中,并唤醒空闲线程开始执行任务。每个工作线程从任务队列中取出一个任务并执行,如果队列为空则等待通知。在销毁线程池时,会等待所有任务执行完毕后终止所有线程并释放占用的资源。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。