linux高性能服务器-线程池实现

定义

线程池属于生产消费模型，管理维持固定数量线程的池式结构，避免线程频繁的创建和销毁

应用场景

当一类任务耗时，严重影响当前线程处理其他任务，异步执行

任务类型

耗时任务：

• CPU密集型
• IO密集型 （ 网络IO 磁盘IO）

线程数量

n * proc

数据结构设计：

任务设计：

typedef struct task_s {void * next;handler_pt func;void * arg;
} task_t;

生产者线程： 发布任务
消费者线程： 取出任务，执行任务
数据结构为链表

队列设计：

typedef struct task_queue_s {void * head;void **tail;int block; // 是否阻塞 spinlock_t lock; // 自选锁pthread_muxtex_t mutex;pthread_cond_t cond;
}task_queue_t;

队列： 存储任务，调度线程池 ，双端开口，先进先出，在多线程中执行，需要加锁
功能： 调取线程池中的消费者线程， 如果此时队列为空，谁（线程)来取任务，谁阻塞休眠
当允许一个进程进入临界资源（互斥状态）。
自旋锁： 其他线程空转cpu，一直等待进入临界资源
互斥锁：切换cpu, 让出执行权, 线程阻塞住，操作系统调用其他的线程

某个线程持有临界资源的时间 < 线程切换的时间 ， 自旋锁 ，时间复杂度为0(1)
生产者新增任务，消费者取出任务 ，0（1），均为移动指针完成（尾插法，头插法）
故使用自旋锁 spinlock_t lock

线程池设计

struct thredpool_t {atomic_int quit;  // 原子变量int thrd_count;pthread_t * threads;task_queue_t task_queue;
};

原子操作：一个线程在执行过程中，其他线程不能执行这个线程的内部操作，只能看到线程执行前或者执行后
应用场景： 某一个基础类型给的变量

接口设计

static task_queue_t * __taksqueue_create() {task_queue_t  * queue = malloc(sizeof(*queue));int ret;ret = pthrad_mutex_init(&mutex);if(ret == 0) {ret = pthread_cond_init(&cond);if(ret == 0) {queue->head = NULL；queue->tail = &(queue->head);queue->block = 1;return queue;}pthread_cond_destory(&queue);}pthread_muext_destory(&queue->mutex);return NULL;
}static void __add_task(task_queue_t  * queue, void * task) {void **link = (void **)task; // malloc*link = NULL; // task->next = NULL;spinlock_lock(&queue->lock);*queue->tail = link; // 末尾添加新的节点queue->tail = link // tail 指向新的尾节点spinlock_unlock(&qeuue->lock);pthread_cond_signal(&queue->cond); // 有任务，唤醒休眠的线程
}static task_t * __pop__task(task_queue_t * queue) {spinlock_lock(&queue->lock);if(queue->head) {spinlock_unclock(&queue->lock);return NULL;}task_t *task;task = queue->head;queue->head = task->next;if(queue-head == NULL) {queue->tail = &queue->head; // &NULL}spinlock_unlock(&queue->lock);return task;
} static void * __get_task (task_queue_t *queue) {task_t *task;while((task = __pop_task(queue))== NULL) {pthread_mutex_lock(&queue->lock);if(queue->block == 0) {rerurn NULL;}// pthread_cond_wait 执行过程：// 1. 先unlock(&mutex)// 2. cond 休眠// 3, 生产者 发送signal// 4. cond 唤醒// 5. 既然上clock(&mutex)pthead_cond_wait(&queue->cond,&queue->mutex);  //休眠pthread_mutex_unlock(&queue->mutex);} return task;
}static void __taskqueue_destroy(task_queue_t * queue) {task_t *task;while((task) = _pop_task(queue)) {free(ptr:task);}spinlock_destroy(&queue->lock);pthread_cond_destory(&queue->cond);pthread_mutex_destory(&queue->mutex);free(ptr:queue);
}// 消费者线程 ，取出任务，执行任务
static void * __thrdpoll_worker(void *arg) {thrdpool_t *pool = (thrdpool *)arg;task_t *task;void *ctx;while(atomic_load(&pool->quit) == 0) {  //原子读task = (task *) __get_task(poll->task_queue);if(!task) {break;}handler_pt func = task->func;ctx = taks->arg;free(task);func(ctx);}return NULL;
}
// 设置队列为非阻塞，并唤醒所有的线程
static void __nonblock(task_queue_t *queue) {pthread->mutex_lock =&queue->lock;queue->block = 0;pthread_mutex_unclock(&queue->mutex);pthread_cond_broadcast(queue->cond); 
}
// 创建线程，回滚式创建对象
static int __threads_create(thrdpool * pool, size_t thrd_count) {pthread_attr_t attr;int ret;ret = pthread_attr_init(&attr); //初始化线程参数if (ret == 0) {pool_>threads = (pthread_t *)malloc(sizeof(pthread_t) * thrd_count);if(pool_threads) {int i = 0;for(;i < thrd_count; i++) {if(pthread_create(&pool->threads[i),&attr,start_routine(),NULL);break; // 创建线程失败，返回}pool->thrd_count  = i; pthread_attr_destory(&attr);if( i == thrd_count)reurn 0;__threads_terminante(pool); // 如果创建的线程数量不等于thrd_count,把创建的线程全部销毁free(pool->threads); // 释放堆空间}}return ret;
}// 创建线程池
static thrdpool_t *  thrdpool_create(int thrd_count) {thrdpool_t * pool;poll = (thrdpool_t *)malloc(sizeof(poll);if(!pool) return NULL;task_queue_t  *queue = __taskqueue_create();if(queue) {pool->task_queue = queue;atomic_init(&pool->quit, 0);if(__threads_create(pool,thrd_count) == 0 ) {return pool;}__taskqueue_destory(pool->taks_queue);}free(pool);return NULL;
}static void __threads_terminate(thrdpool_t * pool) {atomic_store(&queue->quit,1); //原子写__nonblock(pool->task_queue); // 设置非阻塞队列，唤醒所有的线程int i;for(i=0; i<pool->thrd_count;i++) {pthread_join(pool->thread[i],NULL);}
}// 生产者创建任务
static int thrdpool_post(thrdpool_t  * pool, handler_pt func, void *arg) {if (atomic_load(&pool->quit) == 1 ) {  //判断线程池是否标记退出return -1;}task * task = (task_t *)malloc(sizeof(task_t));if(!task)  return -1;task->func = func; // 初始化task->arg = arg;__add_task(pool->task_queue,task); // 添加任务return 0;
}//等待所有线程结束，释放资源
static thrdpool_wait(thrdpool_t *pool) {int i;for(i=0; i<poll->thrd_count;i++) {pthread_join(pool->thread[i],NULL);}__taskqueue_destory(pool->taks_queue);free(pool->threads);free(pool);
}