网站动态图片如何做,外贸营销邮件主题,饮料招商网站大全,厦门网站流量优化价格文章目录 定义应用场景任务类型线程数量数据结构设计#xff1a;任务设计#xff1a;队列设计#xff1a;线程池设计 接口设计 定义
线程池属于生产消费模型#xff0c;管理维持固定数量线程的池式结构#xff0c;避免线程频繁的创建和销毁
应用场景
当一类任务耗时任务设计队列设计线程池设计 接口设计 定义
线程池属于生产消费模型管理维持固定数量线程的池式结构避免线程频繁的创建和销毁
应用场景
当一类任务耗时严重影响当前线程处理其他任务异步执行
任务类型
耗时任务
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) 生产者新增任务消费者取出任务 01均为移动指针完成尾插法头插法 故使用自旋锁 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 NULLqueue-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(i0; ipool-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(i0; ipoll-thrd_count;i) {pthread_join(pool-thread[i],NULL);}__taskqueue_destory(pool-taks_queue);free(pool-threads);free(pool);
}
