多线程服务器分析——Reactor线程模型和性能分析(一)
上接多线程服务器分析——Reactor线程模型和性能分析(序章)
在序章中的最后提到了两个服务器瓶颈的原因：
1、能够扩展的线程数量是有限的
2、阻塞式的等待socket会让线程一直处于空闲状态（当然socket可以换成其他文件描述符）
针对问题1，既然线程的数量是有限的，那我们就尽最大的可能去利用它，榨干它的能力；针对问题2，则可以换一种思路，如果当前的socket没有数据传输过来，那就立即返回一个错误信息，处理下一个建立的连接，然后循环询问所有建立的连接的socket，如果某个socket有数据传输过来就立马进行处理，这就是非阻塞(NIO)的思想。
3.3 非阻塞单线程服务器
在介绍非阻塞的多线程服务器之前，先介绍几个概念，相信读者看过之后会对最终介绍的多线程服务器模型的理解有帮助。这里我们又暂时回到了单线程服务器，只能先委屈一下你的煎饼果子摊了:-)。
3.3.1 非阻塞IO
附近的公司倒闭了，你只得另起炉灶换地方，也雇不起月薪2万的伙计了，不过你有了上次的经验，升级了煎饼果子，可以加鸡排、加鸡蛋，加辣条等等。但渐渐的你发现了一个问题，面对诸多选择顾客们反而犹豫不决，不知道该加啥，站在摊位前选半天才选好，后面有想好买什么的顾客也没法买，等了半天走了。
于是你就自己挨个问顾客，选好的顾客会把煎饼果子的要求告诉你，你再把汇总的信息拿回摊位挨个去做，这样一来效率就高多了。
不等待顾客提出要求，类比到代码上就是不等待文件描述符准备好数据，而是没准备好返回『没准备好』，准备好了就读数据。具象化到socket，当进行read/recv socket的操作时，所谓的『没准备好』的错误码就是EAGAIN 或者 EWOULDBLOCK。
#include 上面的代码就是设置socket非阻塞以及非阻塞读取socket的过程。借助非阻塞IO，就可以通过一个线程监听所有的socket状态，哪个socket准备好就处理哪个socket。
3.3.2 IO多路复用(IO-Multiplexing)
你觉得挨个去收集顾客的要求有点慢，于是你建了一个微信群，顾客直接把需求发到群里，你每过一会儿就去看一眼微信群，把每个微信号对应的需求记下来写在小纸条上，然后做这些小纸条上写好的煎饼果子需求。
有了非阻塞IO，服务器的处理逻辑就可以将所有建立连接的socket放到一个列表中，接着建立一个循环遍历这个列表，每当有socket可读时就去处理；但是当请求的数量越来越多的时候，维护的列表也越来越长，遍历的时间复杂度是
O
(
n
)
O(n)
O(n)，
n
n
n就是列表的长度。Linux则为我们提供了IO多路复用这一机制，可以直接返回监听列表中所有状态发生变化的socket(或者其他file descriptor，以下简称fd)，避免了遍历一遍却只有很少甚至没有fd发生变化的窘境，提高了性能。就像本节开头提到的煎饼果子老板的做法，直接拿到准备好下单顾客的需求，而不是挨个询问。
由于本文不是主要介绍IO多路复用的文章，在这里一步到位使用epoll进行监听，对此不太熟悉的读者可以先了解select、poll到epoll的发展过程以及它们的使用方法。
int main() {
int epoll_fd = epoll_create1(0);
if (epoll_fd == -1) {
...
}
int listen_sock = socket(AF_INET, SOCK_STREAM, 0);
// 设置 socket 为非阻塞模式（可选）
fcntl(listen_sock, F_SETFL, fcntl(listen_sock, F_GETFL, 0) | O_NONBLOCK);
// 绑定和监听socket
...
struct epoll_event ev, events[MAX_EVENTS];
ev.events = EPOLLIN;
ev.data.fd = listen_sock;
// 添加监听 socket 到 epoll 实例
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_sock, &ev) == -1) {
...
}
for (;;) {
// 等待事件
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
...
// 处理事件
for (int n = 0; n < nfds; ++n) {
if (events[n].data.fd == listen_sock) {
// 接受新连接并加入监听
int conn_sock = accept(listen_sock, NULL, NULL);
...
if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, conn_sock, &ev) == -1) {
...
}
} else {
// 处理socket上的可读事件<1>
char buf[512];
ssize_t count = read(events[n].data.fd, buf, sizeof(buf));
...
}
}
}
return 0;
}
3.3.3 单线程Reactor模型
【代码：reactor/single_thread_reactor】
结合3.3.1和3.3.2的内容就是一个最基础的单线程reactor，其本质就是non-blocking和IO multiplexing。在一个死循环中调用epoll_wait，得到可读的socket列表，并依次针对这些socket执行read操作，将read出的数据进行处理。将3.3.2中的代码中的<1>处的read换成handle_request(int fd)，handle_request这个函数可以替换成任意针对客户端发来的数据的处理函数，例如在tcp server的基础上添加一个http的库解析来自客户端的http请求。如果需要对比时间上的差异，也只能与多线程服务器分析——Reactor线程模型和性能分析(序章)中的单线程阻塞服务器进行对比，需要读者额外做的是在client.cc的请求函数中在connect与服务端建立连接之后添加sleep模拟网络阻塞导致数据到达服务端不及时，读者可以自行尝试。
单线程的reactor执行过程则如下图所示：
在loop循环中拿到可读socket并按socket列表顺序执行handle_request进行处理，一句话就可以概括整个的执行过程。在实践中，我们通常不会赤裸裸的直接对socket进行处理，而是将『可读』包装成一个事件(Event)，handle_request则被称为事件回调(Event Callback)，而epoll所在的循环被称为事件循环，以事件驱动(Event Driven)整个模型完成业务逻辑。
3.3.3.1 事件驱动(Event Driven)
在这里提到了事件驱动，这个概念也是困扰了我比较久，主要是它和消息驱动(Message Driven)的区别。在网上搜了很久也没有真正弄懂，其实这两者在实现上是非常接近的，都可以采用生产者-消费者模型，即维护一个队列向里面填充事件/消息，另一边线程池不断从队列中取出事件/消息并处理，因此个人理解这二者的区别更多的是在编程思想或者设计框架时的区别。最终在stackoverflow上找到了两个答案：
A message is an item of data that is sent to a specific destination. An event is a signal emitted by a component upon reaching a given state. In a message-driven system addressable recipients await the arrival of messages and react to them, otherwise lying dormant. In an event-driven system notification listeners are attached to the sources of events such that they are invoked when the event is emitted. This means that an event-driven system focuses on addressable event sources while a message-driven system concentrates on addressable recipients. A message can contain an encoded event as its payload.
这一段主要的意思是消息驱动的系统有一个确定的接收方，等待消息到来并处理、返回，如果没有消息则处于休眠状态；而事件驱动则有一个发送源，针对发送源有多个订阅者，当发送源有事件发生时就会唤醒订阅者。
其实这段话我的个人感觉看着还是有点懵圈，下面有个答案举了个例子：
Let’s say you are building a Payment service for an eCommerce website. When an order is placed, the Order service will ask your Payment service to authorize the customer’s credit card. Only when the credit card has been authorized will the Order service send the order to the warehouse for packing and shipping.
You need to agree with the team working on the Order service on how that request for credit card authorization is sent from their service to yours. There are two options.
Message-driven: When an order is placed, the Order service sends an authorization request to your Payment service. Your service processes the request and returns success/failure to the Order service. The initial request and the result could be sent synchronously or asynchronously.
Event-driven: When an order is placed, the Order service publishes a NewOrder event. Your Payment service subscribes to that type of event so it is triggered. Your service processes the request and either publishes an AuthorizationAccepted or an AuthorizationDeclined event. The Order service subscribes to those event types. All events are asynchronous.
有一个网上电商的付款系统，首先要校验你的信用卡信息才能继续订单操作，现在根据消息驱动和事件驱动有两种处理校验的方式：
- 消息驱动：发送校验消息给付款服务进行校验，付款服务返回给你校验成功或者失败，这个过程可以是同步或者异步的。
- 事件驱动：产生一个校验事件，订阅校验事件的付款服务拿到这个事件，处理之后产生一个校验成功事件或者检验失败事件，由能处理这两种事件的订阅者进行处理，所有的消息传递都是异步的。
读到这里我想了一个挺好玩的例子来说明，就是回转寿司。假设没有线上点单这回事，顾客点单之后，把菜品写到一张纸上并放到传送带上，这就发生了一个『点单事件』，师傅收到点单事件并做好寿司放在传送带上。但是师傅不会是收到一个订单做一个，然后等这个顾客拿到这个订单的寿司再去做下一个，而是一次性做好多订单，然后一起放到传送带上，传送带转到顾客就把自己订单上的寿司拿出来。也可能不同的师傅负责不同的菜品，厨房里还有甜品师傅、凉菜师傅，他们收到订单时间就开始做菜，做完之后产生『军舰寿司事件』、『小布丁事件』、『芥末章鱼事件』并放在传送带上。顾客吃完结账，然后把账单和钱放到传送带上，产生了一个『付款事件』，老板就坐在传送带的某个位置接收这些『付款事件』。
如果是消息驱动，那么点单是你把单子给服务员，服务员给厨师，付款则是你把钱给老板，你是明确知道你的每一个步骤面对的角色是谁，也就是上面所说的消息驱动有一个明确的接收方。
类比到实践中设计系统，这个回转台就相当于event loop，承接各方产生的事件，对应的处理方认领自己的事件。
阅读过《Linux多线程服务端编程》或者在其他地方了解到事件循环(Event Loop)的读者，一定听过one loop per thread，它也是reactor模式的另一种称呼，也就是说一个线程中监听IO事件的循环只有一个。个人认为它的好处是极大的扩展了连接的处理能力，降低响应时间(这个需要多添加一个loop用于单独处理连接事件，在下一节会进行介绍)，配合后端的多线程或者线程池可以轻易的将事件传递给其他线程来做处理，相信读者在之后的介绍中能逐渐体会到这段话的意思。
3.3.4 主从reactor模型
【代码：reactor/master_slave_reactor】
凭借上次的经验，你的煎饼果子摊重新红火了起来，你渐渐又忙不过来了，于是你又雇了一个伙计，同时你想，我都辛苦这么多年了就不能享受享受吗，我不摊煎饼果子了，于是你只负责收集微信群里面的顾客需求，把它们转发给伙计让伙计去做。
在3.3.3节中提到的单线程reactor模型中，处理连接事件和客户端的IO事件都在一个循环中进行：
for (;;) {
// 等待事件
int nfds = epoll_wait(epoll_fd, events, MAX_EVENTS, -1);
if (nfds == -1) {
perror("epoll_wait");
return 1;
}
for (int n = 0; n < nfds; ++n) {
if (events[n].data.fd == listen_sock) {
// 接受新连接，处理连接事件
int conn_sock = accept(listen_sock, NULL, NULL);
...
} else {
// 处理socket上的可读事件，也就是客户端传过来的数据
handle_request(events[n].data.fd);
}
}
}
当客户端请求并发很高的时候，服务端可能来不及处理那么多的连接，因此可以使用一个线程里的loop专门处理连接建立，再将建立好连接的socket传递给其他线程的loop进行监听和处理，实际的处理模型如下图所示：
在上面的图中，main loop中的epoll只监听服务端负责listen的socket，当有请求到达监听队列的时候，服务端listen的socket就会可读，epoll监听到可读事件取消阻塞，服务端就可以执行accept将连接从监听队列中取出，产生的socket就传递给另一个线程中loop中的epoll去监听，剩下的步骤与之前的单线程服务器完全相同，监听到连接的socket可读将其取出并处理连接上的可读事件，即处理客户端发来的请求。值得注意的是，epoll_ctl和epoll_wait是线程安全的，不必考虑fd在线程间传递产生的race condition，不过如果要并发的对socket进行操作，例如读取客户端的数据和close还是要注意的。
按照上面的设计，服务端的代码就变成两个线程，主线程(main loop)负责监听listen_socket并传递给从线程(slave loop)：
void handle_main_loop(int main_epoll_fd, int slave_epoll_fd) {
int listen_sock = socket(AF_INET, SOCK_STREAM, 0);
...// listen_socket处理略，详见代码目录
struct epoll_event ev, events[1];
ev.events = EPOLLIN;
ev.data.fd = listen_sock;
if (epoll_ctl(main_epoll_fd, EPOLL_CTL_ADD, listen_sock, &ev) == -1) { // main loop监听listen socket
perror("epoll_ctl: listen_sock");
std::exit(0);
}
for (;;) {
int nfds = epoll_wait(main_epoll_fd, events, 1, -1);
...
if (events[0].data.fd == listen_sock) {
struct sockaddr_in cli_addr;
socklen_t cli_len = sizeof(cli_addr);
int connfd = accept(listen_sock, (struct sockaddr*)&cli_addr, &cli_len);
...
ev.events = EPOLLIN | EPOLLET;
ev.data.fd = connfd;
if (epoll_ctl(slave_epoll_fd, EPOLL_CTL_ADD, connfd, &ev) == -1) { // 加入到slave loop的epoll监听
perror("epoll_ctl: conn_sock");
std::exit(1);
}
} else {
std::exit(1);
}
}
}
从线程(slave loop)接收来自主线程传递的socket并监听可读事件，产生可读事件后进行处理：
void handle_slave_loop(int slave_epoll_fd) {
struct epoll_event events[MAX_EVENTS];
for(;;) {
int nfds = epoll_wait(slave_epoll_fd, events, MAX_EVENTS, -1);
if (nfds == -1) {
perror("epoll_wait");
std::exit(1);
}
for (int n = 0; n < nfds; ++n) {
handle_request(events[n].data.fd);
}
}
}
然后起两个线程：
int main() {
int global_main_epoll_fd = epoll_create1(0);
if (global_main_epoll_fd == -1) {
perror("epoll_create1");
return 1;
}
int global_slave_epoll_fd = epoll_create1(0);
if (global_slave_epoll_fd == -1) {
perror("epoll_create1");
return 1;
}
auto main_loop = std::bind(handle_main_loop, global_main_epoll_fd, global_slave_epoll_fd);
auto slave_loop = std::bind(handle_slave_loop, global_slave_epoll_fd);
std::thread t_main(main_loop);
std::thread t_slave(slave_loop);
t_main.join(); // unreachable
t_slave.join();
return 0;
}
我们可以简单对比一下，使用了主从loop线程的服务端和所有监听都在一个loop线程的服务端，服务端的处理函数依然与多线程服务器分析——Reactor线程模型和性能分析(序章)中的handle_request相同，都是打印来自客户端的字符串，休眠1毫秒，然后返回"hello world"。
首先把监听队列，也就是TCP的全连接队列设置为1024，方便观察。不过这个队列会受到SOMAXCONN的限制，具体根据不同的操作系统有所不同，如果设置的值它大，会自动截断为SOMAXCONN：
listen(listen_sock, 1024);
当客户端的并发设置为1000时，对比：
访问单reactor服务器耗时：
访问主从reactor服务器耗时：
可以发现二者消耗的时间几乎没什么区别。
当把客户端请求的并发提高至10000，对比：
访问主从reactor服务器耗时：
耗时基本上是1000并发的10倍，这也是符合预期的，目前对于数据的处理，依旧是单线程的。
再来看单reactor的表现：
在执行的过程中出现了Connection reset by peer这个错误，这是由于服务端返回了一个Reset信号重置连接导致的，通常引起这种现象的原因是服务端在一段时间内收到的请求大于它自身的处理能力，实际上就是TCP的全连接队列接收的请求大于SOMAXCONN了，而返回Reset信号也是由内核中的/proc/sys/net/ipv4/tcp_abort_on_overflow参数决定，cat /proc/sys/net/ipv4/tcp_abort_on_overflow看一下这个参数，如果是0表示丢弃，1则表示返回Reset，也就是上面展示的表现。
通过上面的对比，可以发现虽然主从reactor服务器不能提高响应速度，但可以提高处理并发连接的能力。
3.3.5 本章小结
在这一篇我们详细看了非阻塞的单线程服务器，也了解了reactor模型的基本框架和事件驱动的基本概念，为什么把主从reactor也放到这一章来讲解呢？个人觉得广义上的多线程服务器是指在业务的处理逻辑上采用多线程的形式，不过主从reactor称为多线程服务器也完全没有问题，甚至可以采用多个slave loop，main loop接收到socket之后轮询的将socket传递给每个slave loop。
在本文最开头，提了两个问题：
1、能够扩展的线程数量是有限的
2、阻塞式的等待socket会让线程一直处于空闲状态（当然socket可以换成其他文件描述符）
第二个问题已经通过非阻塞相关的内容解答完毕，第一个问题在开头说过需要尽可能的榨取有限线程的能力，这就涉及到了如何使用多线程的问题，是像多线程服务器分析——Reactor线程模型和性能分析(序章)中的一样每个socket占用一个线程吗？接下来，将结合《Linux多线程服务端编程》这本书（没看过没关系，只是会引述书中的一些原文来解释一些概念）以及一个实际的业务场景来和读者一起讨论多线程的一些使用方法。
下一章链接：多线程服务器分析——Reactor线程模型和性能分析(二)
