线程池设计原理（线程池的设计一篇就够）

小君 2023-01-01 17:35:37 925

线程池设计原理（线程池的设计一篇就够）一下所有的代码设计适用于单生产者多消费者模式3）生产者之间的同步方式，消费者之间的同步方式？OK，问题来了任务由谁产生（生产者），如何丢给线程池的某个线程（消费者）？这个问题的回答需从以下几方面：1）生产者采用什么方式与消费者同步？2）任务如何保存？

为什么需要线程池

在那些情况下我们会使用到多线程：

阻塞调用（阻塞IO调用、等待资源）
耗时的计算（读写文件、复杂的计算）
高密度任务（高并发低延时的网络IO请求）

面临以上情况时都去临时创建线程会带来什么问题：

创建了太多的线程，系统资源就会被浪费，而且会浪费时间去创建和销毁线程。
创建线程太慢，导致执行任务结果返回过慢。
销毁线程太慢，可能会影响别的进程使用资源。

所以：创建多个线程，放在池子里不销毁，要用的时候就把任务丢给池子里的线程去执行，这就是线程池。

OK，问题来了任务由谁产生（生产者），如何丢给线程池的某个线程（消费者）？这个问题的回答需从以下几方面：

1）生产者采用什么方式与消费者同步？

2）任务如何保存？

3）生产者之间的同步方式，消费者之间的同步方式？

一下所有的代码设计适用于单生产者多消费者模式

条件变量结合互斥锁任务队列

设计如何：

线程池设计原理（线程池的设计一篇就够）(1)

代码如下：

typedef struct queue_task { void* (*run)(void *); void* argv; }task_t; typedef struct queue { int head; int tail; int size; int capcity; task_t* tasks; } queue_t; typedef struct async_queue { pthread_mutex_t mutex; pthread_cond_t cond; int waiting_threads; queue_t* queue; int quit; // 0 表示不退出 1 表示退出 /* 调试变量 */ long long tasked; // 已经处理完的任务数量 } async_queue_t;

取任务的代码设计如下：

task_t* async_cond_queue_pop_head(async_queue_t* q int timeout) { task_t *task = NULL; struct timeval now; struct timespec outtime; pthread_mutex_lock(&(q->mutex)); if (queue_is_empty(q->queue)) { q->waiting_threads ; while (queue_is_empty(q->queue) && (q->quit == 0)) { gettimeofday(&now NULL); if (now.tv_usec timeout > 1000) { outtime.tv_sec = now.tv_sec 1; outtime.tv_nsec = ((now.tv_usec timeout) % 1000) * 1000; } else { outtime.tv_sec = now.tv_sec; outtime.tv_nsec = (now.tv_usec timeout) * 1000; } pthread_cond_timedwait(&(q->cond) &(q->mutex) &outtime); } q->waiting_threads--; } task = queue_pop_head(q->queue); /* 调试代码 */ if (task) { q->tasked ; static long long precision = 10; if ((q->tasked % precision ) == 0) { time_t current_stm = get_current_timestamp(); precision *= 10; } } pthread_mutex_unlock(&(q->mutex)); return task; }

不足：

因为Mutex引起线程挂起和唤醒的操作，在IO密集型的服务器上不是特别高效（实测过）；
条件变量必须和互斥锁相结合使用，使用起来较麻烦；
条件变量不能像eventfd一样为I/O事件驱动。
管道可以和I/O复用很好的融合，但是管道比eventfd多用了一个文件描述符，而且管道内核还得给其管理的缓冲区，eventfd则不需要，因此eventfd比起管道要高效。

eventfd epoll

队列的设计：

typedef struct async_queue { queue_t* queue; int quit; // 0 表示不退出 1 表示退出 int efd; //event fd int epollfd; // epoll fd /* 调试变量 */ long long tasked; // 已经处理完的任务数量 } async_queue_t;

插入任务：

BOOL async_eventfd_queue_push_tail(async_queue_t* q task_t *task) { unsigned long long i = 0xffffffff; if (!queue_is_full(q->queue)) { queue_push_tail(q->queue task); struct epoll_event ev; int efd = eventfd(0 EFD_CLOEXEC | EFD_NONBLOCK); if (efd == -1) printf("eventfd create: %s" strerror(errno)); ev.events = EPOLLIN ;// | EPOLLLT; ev.data.fd = efd; if (epoll_ctl(q->epollfd EPOLL_CTL_ADD efd &ev) == -1) { return NULL; } write(efd &i sizeof (i)); return TRUE; } return FALSE; }

取任务：

task_t* async_eventfd_queue_pop_head(async_queue_t* q int timeout) { unsigned long long i = 0; struct epoll_event events[MAX_EVENTS]; int nfds = epoll_wait(q->epollfd events MAX_EVENTS -1); if (nfds == -1) { return NULL; } else { read(events[0].data.fd &i sizeof (i)); close(events[0].data.fd); // NOTE: need to close here task_t* task = queue_pop_head(q->queue); /* 调试代码 */ if (task) { q->tasked ; static long long precision = 10; if ((q->tasked % precision ) == 0) { time_t current_stm = get_current_timestamp(); printf("%d tasks cost : %d\n" precision current_stm - start_stm); precision *= 10; } } return task; } return NULL; }

因为eventfd每次写数据后，只会唤醒一个epoll_wait所在的线程，so，确保了同一时刻仅有一个线程取任务。

不足：

上面两种方案，所有的线程共用同一个队列，所以消费者线程之间取任务时需要做同步，生产者和消费者也需要做同步。用一个形象的图可以表示如下：

线程池设计原理（线程池的设计一篇就够）(2)

eventfd epoll 多队列的设计

设计思想如下图：

线程池设计原理（线程池的设计一篇就够）(3)

Oh my god huge project where can i find thre thread pool source. 说来话长，这个代码是EasyDarwin的源码，但是因为某种原因，EasyDarwin的源码不再共享，取而代之的打赏的二维码，so，我把他们的源码做了局部的修改，然后重新提交，且命名为StreamingServer，里面的设计是采用条件变量做同步的，但是多队列的思想是可以沿袭的，打算加班用eventfd实现。

这样一种设计是不是让我们能够想到下面这幅图呢？

线程池设计原理（线程池的设计一篇就够）(4)