epoll+threadpool高并发网络IO模型的实现

网络模型的选择

当多个任务到来时需要对其进行及时响应，并将任务下发给特定的处理线程，完成对应的任务。如果只用传统的服务器模型：同步阻塞、单线程进行listen轮询监听的话，那效率和并发往往达不到需求。而我们借助 epoll 的话就会很完美的解决这个问题
关于epoll请阅读：Linux下的I/O复用与epoll详解
常用的网络IO模型 ：五种网络IO模型

ThreadPool

当实现高并发服务器时，当然需要实现多线程并发工作来处理多个客户端的连接。那么方法有如下

方案：

1. 每当客户端发送一个请求，服务器创建一个线程来处理客户端的连接请求。

2. 先预先创建若干个线程，并让线程先阻塞。每当一个请求到来唤醒一个线程，对其进行处理，处理完毕后继续阻塞。

对比：

1. 是动态创建，当一个任务到达，就需要创建线程，当任务处理完毕后对线程进行销毁。
2. 而是先预先创建，线程处理完任务后无需销毁，可以继续阻塞等待下一个任务。
   很明显方案1的开销会远大于方案2.从这一点来看需要选择方案2。

但是1也有优点。那就是他可以动态创建，保证每个任务在内核承载能力允许的情况下，可以及时处理。所以怎么样才可以将二者的优点结合呢？
那就是在方案2的基础上再创建一个线程，一直轮询监听任务数，来动态增加或者删减任务数。这样也就是线程池。

实现

生产者消费者模型

详细介绍和实现可参考：生产者与消费者问题C语言实现
首先，我们是利用生产者消费者模型来实现这个服务器模型。谁来充当消费者和生产者的模型呢，下面介绍一下生产者和消费者在本模块的大致功能。

生产者功能：

1. 对客户端的连接请求进行监听。
2. 将客户端的连接请求存储到一个容器中。
3. 当任务不为空时发送不为空信号来通知消费者来接收。

消费者功能：

1. 监听生产者发送的不为空信号。
2. 当收到信号时，从容器取走任务并对其处理。
3. 向生产者发送容器未满信号。(因为任务如果过快的话容器可能已经放满，生产者无法接着接受客户端请求)

所以，由上可得
生产者就是epoll模型，消费者就是线程池。

线程池

1.线程池（一个结构体）。
既然是线程池，首先需要一个标识符来表明这个线程池是否已经开始使用或者是否已经被关闭。其次我们需要一个记录每个线程的tid的数组，和存放管理线程的id的mtid变量，来方便管理线程对普通i线程的回收。那么就需要以下结构体成员：

int thread_shutdown;//1为关闭线程池，0为打开线程池
pthread_t *tids;//tid数组
pthread_t mtid;//管理者线程的id
int thread_max;//最大的线程数
int thread_min;//最小的线程数（销毁时需要）
int thread_alive;//存活的线程数
int thread_busy;//正在处理任务的线程数

生产者需要将任务存放到一个容器中，如果我们将容器单独考虑，那么就会很麻烦（如何对其进行加锁，函数传参等问题）。所以不如直接将它扔到这个结构体中(我这里使用的是循环队列，当然其他的容器也可以)。既然我们是多线程访问这个临界资源，哪么也少不了一把锁了。所以还需要以下结构体成员：

task_t *task_queue;
int queue_cur;//当前存放的任务数
int queue_max;//队列允许存放的最大容器数
int queue_front;//队列头部指针（类似于栈顶）
int queue_rear;//队列的尾部指针

pthread_mutex_t plock;//锁

我们在上文中提到生产者需要将来任务了告诉消费者，而且消费者需要阻塞等待这个信号。那么我们该如何实现呢?，这时候我们使用条件变量就可以很好地满足。
因为条件变量一方面可以让消费者阻塞。二来可以让生产者通过pthread_cond_signal（）函数来通知消费者停止挂起
（小心惊群问题 后文会详细解释这个问题的避免）。
所以我们就需要两个条件变量来实现：

1. 生产者告诉消费者容器有任务了可以去处理了。
2. 消费者告诉生产者容器没有满可以继续放任务了。

pthread_cond_t pnot_full;//消费者告诉生产者容器没有满可以继续放任务了
pthread_cond_t pnot_empty;//生产者告诉消费者可以去处理了

最后因为管理者线程需要普通线程进行销毁工作（只有少量的线程在工作，而大部分都被挂起），那我们只需要设置一个 int变量表明需要销毁的线程数

int thread_exitcode;

任务

这里的任务就是上面的task_t，也是一个结构体。
因为我们需要将这个模型实现普及化，那么任务必然可以改变的。所以我们就可以在这个结构体封装两个成员即可，一个函数指针来表示对应的任务。还有一个void*指针来存放该函数需要的参数。

typedef struct{
    
void * (*job)(void*);
void *arg;
} task_t;

生产者epoll

epoll主要就只有一个epfd文件描述符来标明红黑树的根。我们在主函数声明一个即可（要是c++实现的话就可以将其封装成一个类的成员变量）。

typedef struct {
    
	void * (*job)(void *);
	void *arg;
}task_t;

typedef struct {
    
	int thread_shutdown;
	int thread_max;
	int thread_min;
	int thread_alive;
	int thread_busy;
	int thread_exitcode;

	task_t *task_queue;
	int queue_max;
	int queue_cur;
	int queue_front;
	int queue_rear;

	pthread_t *tids;
	pthread_t mtid;

	pthread_mutex_t plock;
	pthread_cond_t pnot_full;
	pthread_cond_t pnot_empty;
}pool_t;

基本实现逻辑

main函数：
1.0. 对pool的网络端进行初始化(thread_pool_netinit)
1.0.1. thread_pool_netinit:这里就是基本的socket通信的server端，将初始化好的serverfd返回给主函数
1.1. 对task进行初始化，对于task的函数博主这里就是一个简单的小写字母转大写（thread_user_job）,参数为一个serverfd。
1.1.1. thread_user_job：由于epoll已经监听到客户端的连接请求，那么只需要对serverfd进行监听即可，剩余的就是处理主逻辑（博主的是一个小写转大写逻辑）
1.2 . 其次对pool进行初始化(thread_pool_init)
1.2.1. thread_pool_init：这里主要是为线程池分配空间(比如：pool结构体，tids数组，task_queue循环队列进行malloc分配空间)将各个成员初始化（thread_shutdown这个成员一定要初始化为0,表述打开状态）,最后按照参数建立指定个数的消费者线程(thread_customer_job)，以及一个管理者线程（thread_manger_job）。最后结构体的地址返回给main函数。
1.2.1.1. thread_customer_job：首先判断线程池是否打开，然后先获得锁（因为条件变量本身也是临界资源）然后pthread_cond_wait(pnot_empty,&plock)。阻塞等待pnot_empty条件变量的到来。注意：上面的惊群问题将在这里进行描述。因为pthread_cond_waitAPI手册中描述的是，将会对若干个（注意这个若干个）阻塞等待该条件变量的线程进行唤醒。所以每当一个条件变量到达，原理上只需要一个线程即可，而这里会唤醒若干个线程，所以会导致惊群问题，那么我们这里可以加一个while循环在此判断数据thread_cur（当前任务数是否为空），只有第一个获得信号量的会跳出while循环然后对数据thread_cur–避免其他的线程跳出循环，（但博主认为，在跳出while循环和执行cur–期间也可能会有其他线程跳出循环，这里留个坑先）。处理完逻辑是注意将线程数和循环队列对应的数据进行更新即可。最后会继续阻塞等待该条件变量。
1.2.1.2. ：thread_manger_job:先判断线程池是否打开。抢占锁并对数据进行更新保留。再判断逻辑结构（博主这里是通过存活线程数thread_alive，当前任务数thread_cur和忙碌线程数，thread_busy，进行简单数学比例运算来判断是否销毁线程）。
1.3. ：对epoll进行初始化（thread_epoll_init）
1.3.1. :thread_epoll_init 主要就是epoll_create 以及将serverfd通过epoll_ctl放到红黑树上进行监听。最后将epfd返回（值得注意的是，需要将epoll改成ET模式 因为我添加任务后一定会解决，不必阻塞等待解决）
1.4. :epoll开始监听（thread_epll_start）
1.4.1. ：thread_epoll_start:首先判断线程池是否打开，然后epoll_wait阻塞监听serverfd，如果返回值>0那么有连接请求(thread_epoll_addtask)。
1.4.1.1. ：thread_epoll_addtask。首先判断线程池是否关闭，再判断容器是否已经满了，没满的话添加任务到队列中。满了的话pthread_cond_wait阻塞等待 pnot_full。
1.5.: 销毁掉那个全局锁