20种不同并发模型示例，带你深入理解并发模型

作者：lionleeli

文末抽奖送书，欢迎参与~

导语

曾看过很多并发模型相关的文章，但是这些文章大部分只讲了并发模型的实现原理，并没有给出具体的示例代码，看完总觉得对并发模型这个知识点是一知半解，不得要领。为了掌握高并发模型，我这里抛砖引玉，实现了20种常见的并发模型，并对每种并发模型进行了性能压测和分析。由于本人水平有限，文章中难免有一些不准确或者纰漏的地方，欢迎大家批评指正。

1.缘起

最近看了好友推荐的一本新书 ，该书RPC框架和微服务集群的部分甚是不错，其中的“第10章-I/O模型与并发”中介绍了 17 种不同的并发模型，看完之后更是感觉受益匪浅。

但美中不足的是这 17 种并发模型只支持短连接，配套的 BenchMark 工具不支持发起指定的请求负载，给出的性能指标也不够丰富。

受到该内容的启发，我在该内容的基础上实现了 20 种常见的支持长连接的并发模型，完善了协议解析效率和BenchMark 工具。

2.前置说明

因为这 20 种并发模型的代码量已经达到了 1万2千多行，是不适合在一篇文章中全部展示的。所以我把相关的代码开源在github上，方便大家查看。

在文中介绍到相关代码时都会给出具体的代码位置，但只会在文章中贴出关键的代码，即便如此，本文的代码量依然不少，强烈建议收藏后阅读。

github上的项目是MyEchoServer，项目链接为：

2.1项目的目录结构

MyEchoServer 项目的目录结构如下所示。

.
├── BenchMark
│   ├── benchmark.cpp
│   ├── client.hpp
│   ├── clientmanager.hpp
│   ├── makefile
│   ├── percentile.hpp
│   ├── stat.hpp
│   └── timer.hpp
├── common
│   ├── cmdline.cpp
│   ├── cmdline.h
│   ├── codec.hpp
│   ├── conn.hpp
│   ├── coroutine.cpp
│   ├── coroutine.h
│   ├── epollctl.hpp
│   ├── packet.hpp
│   └── utils.hpp
├── ConcurrencyModel
│   ├── Epoll
│   ├── EpollReactorProcessPoolCoroutine
│   ├── EpollReactorProcessPoolMS
│   ├── EpollReactorSingleProcess
│   ├── EpollReactorSingleProcessCoroutine
│   ├── EpollReactorSingleProcessET
│   ├── EpollReactorThreadPool
│   ├── EpollReactorThreadPoolHSHA
│   ├── EpollReactorThreadPoolMS
│   ├── LeaderAndFollower
│   ├── MultiProcess
│   ├── MultiThread
│   ├── Poll
│   ├── PollReactorSingleProcess
│   ├── ProcessPool1
│   ├── ProcessPool2
│   ├── Select
│   ├── SelectReactorSingleProcess
│   ├── SingleProcess
│   └── ThreadPool
├── readme.md
└── test
    ├── codectest.cpp
    ├── coroutinetest.cpp
    ├── makefile
    ├── packettest.cpp
    ├── unittestcore.hpp
    └── unittestentry.cpp

相关的目录说明如下。

• BenchMark是基准性能压测工具的代码目录。
• ConcurrencyModel是20种不同并发模型的代码目录，这个目录下有 20 个不同的子目录，每个子目录都代表着一种并发模型的实现示例。
• common是公共代码的目录。
• test目录为单元测试代码的目录。

3.预备工作

因为I/O模型是并发模型涉及到的关键技术点，所以我们也不会免俗，也会介绍一下常见的I/O模型。

为了降低实现难度，这里我们实现了一个简单的应用层协议，并实现一些通用的基础代码，以便后续高效的实现不同的并发实例。

3.1 常见I/O模型

常见的I/O模型有五种：阻塞I/O、非阻塞I/O、多路I/O复用、信号驱动I/O、异步I/O。其中的阻塞I/O、非阻塞I/O、多路I/O复用、信号驱动I/O都是同步IO。

同步I/O和异步I/O的区别在于，是否需要进程自己再调用I/O读写函数。同步I/O需要，异步I/O不需要。

3.1.1 阻塞I/O

在阻塞IO模式下，只要I/O暂不可用，读写操作就会被阻塞，直到I/O可用为止，在被阻塞期间，当前进程是被挂起的，这样就无法充分的使用CPU，导致并发效率低下。

3.1.2 非阻塞I/O

在非阻塞IO模式下，读写操作都是立即返回，此时当前进程并不会被挂起，这样就可以充分的使用CPU，非阻塞I/O通常会和多路I/O复用配合着一起使用，从而实现多个客户端请求的并发处理。

3.1.3 多路I/O复用

多路I/O复用实现了多个客户端连接的同时监听，大大提升了程序感知客户端连接可读写状态变化的效率。在Linux下多路I/O复用的系统调用为select、poll、epoll。

3.1.4 信号驱动I/O

通过注册SIGIO信号的处理函数，实现了一个I/O就绪的通知机制，在SIGIO信号的处理函数再进行读写操作，从而避免了低效的I/O是否就绪的轮询操作。

但是在信号处理函数中是不能调用异步信号不安全的函数，例如，有锁操作的函数就是异步信号不安全的，故信号驱动I/O应用的并不多。

3.1.5 异步I/O

前面的4种I/O模型都是同步IO，最后一种I/O模型是异步IO。异步I/O就是先向操作系统注册读写述求，然后就立马返回，进程不会被挂起。操作系统在完成读写操作之后，再调用进程之前注册读写述求时指定的回调函数，或者触发指定的信号。

3.2 应用层协议

20种并发示例实现的是最常见的Echo（回显）服务，这里我们设计了一个简单的应用层协议，格式如下图所示。

协议由两部分组成，第一部分是固定长度（4字节）的协议头部，协议头部用于标识后面的变长协议体的长度，第二部分就是是具体的变长协议体。

3.2.1 协议实现

协议的编解码在common目录的codec.hpp文件中实现，其中DeCode函数用于实现协议的流式解析。

采用流式解析，能避免拒绝服务攻击。例如，攻击者创建大量的连接，然后每个连接上只发送一个字节的数据，如果采用常见的解析方式，一直在socket上读取数据，直到完成一个完整协议请求的解析。

在不采用协程的情况下，不管是阻塞IO、非阻塞IO、IO复用，当前的工作进程或者线程不是被挂起（阻塞IO），就是CPU使用率飙升（非阻塞IO），服务可用的工作进程或者线程会快速被消耗完，导致服务无法对正常的客户端提供服务，从而形成拒绝服务攻击。

流式解析（来多少字节，就解析多少字节）+ 协程切换（IO不可用时切换到其他协程）+ Reactor定时器实现非阻塞IO的超时机制，就可以很好的解决这种拒绝服务攻击。

3.2.2 共享的二进制缓冲区

这里特别说明一下二进制缓冲区的实现。在实现协议的时候，通常会「存储读取到的网络数据缓冲区」和「协议解析的缓冲区」这两份独立的缓冲区。

而我这里思考后，发现其实不用多申请一块二进制缓冲区，写入读取到的网络数据和解析读取到的网络数据可以共享同一个二进制缓冲区，进而减少了内存的分配和两块内存之间的拷贝。共享的二进制缓冲区的示意图如下图所示。

共享的二进制缓冲区在common目录的packet.hpp文件中实现。

3.3 命令行参数解析

不管是BenchMark工具，还是不同的并发模型程序，都需要支持从命令行中读取动态参数的能力。因为参数解析的getopt系列函数并不易用，故参考Go语言的flag包实现，独立封装了一套易用的命令行参数解析函数。

具体的实现在common目录的cmdline.h和cmdline.cpp文件中。

3.4 协程池实现

因为有协程池相关的并发模型，所以需要实现协程池。协程池的实现在common目录的coroutine.h和coroutine.cpp文件中。

特别提一下，协程池这里通过getcontext、makecontext、swapcontext这三个库函数来实现，并且通过C++11的模版函数和可变参数模板的特性，实现了支持变参列表的协程创建函数。协程创建函数的实现如下所示。

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}

4.BenchMark 工具

正所谓工欲善其事，必先利其器。为了评估不同并发模型的性能，需要构建一个BenchMark工具来实现请求的发压。

4.1 并发模型

在看压测工具实现之前，需要思考一个问题，「如果压测工具本身并发能力不足，则无法产生足够的流量负载，也就无法测试出不同并模型的性能极限」，所以也需要设计好压测工具使用的并发模型。

我们使用多线程+Reactor的并发模型来实现压测工具。发起请求的每个线程都是一个单独的Reactor模型。Reactor模型简图如下图所示。

Reactor是一种事件监听和分发模型，配合epoll可以实现高效的并发处理，从而能充分的利用CPU，即使是单线程也能产生足够大的请求负载。

4.2 支持的特性

发压工具BenchMark的usage输出如下所示。

root@centos BenchMark $ ./BenchMark -h
BenchMark -ip 0.0.0.0 -port 1688 -thread_count 1 -max_req_count 100000 -pkt_size 1024 -client_count 200 -run_time 60 -rate_limit 10000 -debug
options:
    -h,--help                      print usage
    -ip,--ip                       service listen ip
    -port,--port                   service listen port
    -thread_count,--thread_count   run thread count
    -max_req_count,--max_req_count one connection max req count
    -pkt_size,--pkt_size           size of send packet, unit is byte
    -client_count,--client_count   count of client
    -run_time,--run_time           run time, unit is second
    -rate_limit,--rate_limit       rate limit, unit is qps/second
    -debug,--debug                 debug mode, more info print

发压工具实现了以下的特性。

• 支持对监听在指定的ip和port的服务发起压测。
• 支持多线程压测，并可以指定使用的线程数。
• 支持指定客户端连接建立成功之后，最多可以发起多少次请求。（ps：这个选项值如果设置为1，则请求就退化成通过短连接来完成）
• 支持指定请求包的大小，单位为字节。
• 支持指定每个线程下发起请求的客户端并发连接数。
• 支持指定总的压测时间，单位秒。
• 支持指定压测能产生的最大的流量负载，单位qps。
• 支持debug模式。

4.3 代码实现

在单独的BenchMark目录中实现了这个压测工具，大家可以自行查看相关代码。这里特别说明一下，我使用了状态机来实现客户端请求的持续发送，代码会稍显复杂。

4.4 执行效果

这里展示一下，压测其中一个单进程并发模型（EpollReactorSingleProcess）的效果，执行结果如下图所示。

从上图的压测结果可以看出，在请求数据长度为1k的情况下，EpollReactorSingleProcess的并发模型就能对外提供高达22万qps的并发处理能力。

上图中红色框出的内容就是最后的压测结果数据，这里的数据分为4部分。

• 接口的pct50、pct95、pct99和pct999的耗时数据。
• 请求成功数、请求失败数、尝试建立连接数、连接失败数、读失败数和写失败数。
• 客户端连接数和请求成功的qps数。
• 请求失败率和连接失败率。

5. 20种不同的并发模型

在本节，我们将展示20种不同的并发模型的具体实现。

ConcurrencyModel目录下的每一个子目录都对应一种并发模型的实现。例如，SingleProcess子目录就是单进程并发模型的实现。

ConcurrencyModel目录下的每个子目录下都只有一个cpp文件和一个makefile文件，而这个cpp文件就是这种并发模型的主流程代码，而makefile文件是用于编译的。例如，SingleProcess子目录下只有一个singleprocess.cpp文件。

5.1 简单模型

5.1.1 SingleProcess

单进程的并发模型是最简单的，每次只为一个客户端连接服务，直到读写失败或者客户端关闭了连接。对应的代码如下所示。

#include <sys/socket.h>
#include <unistd.h>

#include <iostream>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

void handlerClient(int client_fd) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      return;
    }
    if (not SendMsg(client_fd, msg)) {
      return;
    }
  }
}

void usage() {
  cout << "SingleProcess -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  int sock_fd = CreateListenSocket(ip, port, false);
  if (sock_fd < 0) {
    return -1;
  }
  while (true) {
    int client_fd = accept(sock_fd, NULL, 0);
    if (client_fd < 0) {
      perror("accept failed");
      continue;
    }
    handlerClient(client_fd);
    close(client_fd);
  }
  return 0;
}

在main函数中，开启网络监听之后，就陷入死循环，在循序中获取客户端的连接，并处理客户端的请求。

5.1.2 MultiProcess

多进程的并发模型是专门为每个客户端连接创建一个进程，进程服务完客户端之后再退出。对应的代码如下所示。

#include <signal.h>
#include <sys/socket.h>
#include <unistd.h>

#include <iostream>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

void handlerClient(int client_fd) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      return;
    }
    if (not SendMsg(client_fd, msg)) {
      return;
    }
  }
}

void childExitSignalHandler() {
  struct sigaction act;
  act.sa_handler = SIG_IGN;  //设置信号处理函数，这里忽略子进程的退出信号
  sigemptyset(&act.sa_mask);  //信号屏蔽设置为空
  act.sa_flags = 0;  //标志位设置为0
  sigaction(SIGCHLD, &act, NULL);
}

void usage() {
  cout << "MultiProcess -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  int sock_fd = CreateListenSocket(ip, port, false);
  if (sock_fd < 0) {
    return -1;
  }
  childExitSignalHandler();  // 这里需要忽略子进程退出信号，否则会导致大量的僵尸进程，服务后续无法再创建子进程
  while (true) {
    int client_fd = accept(sock_fd, NULL, 0);
    if (client_fd < 0) {
      perror("accept failed");
      continue;
    }
    pid_t pid = fork();
    if (pid == -1) {
      close(client_fd);
      perror("fork failed");
      continue;
    }
    if (pid == 0) {  // 子进程
      handlerClient(client_fd);
      close(client_fd);
      exit(0);  // 处理完请求，子进程直接退出
    } else {
      close(client_fd);  // 父进程直接关闭客户端连接，否则文件描述符会泄露
    }
  }
  return 0;
}

在main函数中，开启网络监听之后，就陷入死循环。在循环中，每获取到一个客户端的连接，就调用fork函数创建一个子进程，并在子进程中处理客户端的请求，处理完客户端的请求之后，子进程就直接退出。

5.1.3 MultiThread

多进程的并发模型是专门为每个客户端连接创建一个线程，线程服务完客户端之后再退出。对应的代码如下所示。

#include <sys/socket.h>
#include <unistd.h>

#include <iostream>
#include <thread>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

void handlerClient(int client_fd) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      break;
    }
    if (not SendMsg(client_fd, msg)) {
      break;
    }
  }
  close(client_fd);
}

void usage() {
  cout << "MultiThread -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  int sock_fd = CreateListenSocket(ip, port, false);
  if (sock_fd < 0) {
    return -1;
  }
  while (true) {
    int client_fd = accept(sock_fd, NULL, 0);
    if (client_fd < 0) {
      perror("accept failed");
      continue;
    }
    std::thread(handlerClient, client_fd).detach();  // 这里需要调用detach，让创建的线程独立运行
  }
  return 0;
}

在main函数中，开启网络监听之后，就陷入死循环。在循环中，每获取到一个客户端的连接，就调用thread函数创建一个子线程，并在子线程中处理客户端的请求，处理完客户端的请求之后，子线程就直接退出。

5.1.4 ProcessPool1

多进程的并发模型需要频繁的创建和销毁进程，这会导致系统开销高，资源占用较多。而进程池的并发模型，则是预先创建指定数量的进程，每个进程不退出，而是一直为不同的客户端提供服务。这种模型可以减少进程的创建和销毁，从而提高系统的并发处理能力，降低系统开销和资源占用。对应的代码如下所示。

#include <sys/socket.h>
#include <unistd.h>

#include <iostream>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

void handlerClient(int client_fd, int64_t& count) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      return;
    }
    if (not SendMsg(client_fd, msg)) {
      return;
    }
    count++;
  }
}

void handler(int worker_id, int sock_fd) {
  int64_t count = 0;
  while (true) {
    int client_fd = accept(sock_fd, NULL, 0);
    if (client_fd < 0) {
      perror("accept failed");
      continue;
    }
    handlerClient(client_fd, count);
    close(client_fd);
    count++;
    if (count >= 10000) {
      cout << "worker_id[" << worker_id << "] deal_1w_request" << endl;
      count = 0;
    }
  }
}

void usage() {
  cout << "ProcessPool1 -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  int sock_fd = CreateListenSocket(ip, port, false);
  if (sock_fd < 0) {
    return -1;
  }
  for (int i = 0; i < GetNProcs(); i++) {
    pid_t pid = fork();
    if (pid < 0) {
      perror("fork failed");
      continue;
    }
    if (0 == pid) {
      handler(i, sock_fd);  // 子进程陷入死循环，处理客户端请求
      exit(0);
    }
  }
  while (true) sleep(1);  // 父进程陷入死循环
  return 0;
}

在main函数中，开启监听之后，根据系统当前可用的CPU核数，预先创建数量与之相等的子进程。每个子进程都陷入死循环一直监听客户端连接的到来并给客户端提供服务。

5.1.5 ProcessPool2

在前面的进程池并发模型中，所有的子进程都会调用accept函数来接受新的客户端连接。这种方式存在竞争，当客户端新的连接到来时，多个子进程之间会争夺接受连接的机会。在操作系统内核2.6版本之前，所有子进程都会被唤醒，但只有一个可以accept成功，其他失败，并设置EGAIN错误码。这种方式会导致不必要的系统调用，降低系统的性能。

在内核2.6版本及之后，新增了互斥等待变量，只有一个子进程会被唤醒，减少了不必要的系统调用，提高了系统的性能。这种方式称为"惊群"问题的解决方案，可以有效地避免不必要的系统调用，提高系统的并发处理能力。

虽然内核2.6版本及之后只有一个子进程被唤醒，但仍然存在互斥等待，这种方式并不够优雅。我们可以使用socket套接字的SO_REUSEPORT选项，让多个进程同时监听在相同的网络地址（IP+Port）上，内核会自动在多个进程之间做连接的负载均衡，而不存在互斥等待行为，从而提高系统的性能和可靠性。

对应的代码如下所示。

#include <sys/socket.h>
#include <unistd.h>

#include <iostream>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

void handlerClient(int client_fd, int64_t& count) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      return;
    }
    if (not SendMsg(client_fd, msg)) {
      return;
    }
    count++;
  }
}

void handler(int worker_id, string ip, int64_t port) {
  // 开启SO_REUSEPORT选项
  int sock_fd = CreateListenSocket(ip, port, true);
  if (sock_fd < 0) {
    return;
  }
  int64_t count = 0;
  while (true) {
    int client_fd = accept(sock_fd, NULL, 0);
    if (client_fd < 0) {
      perror("accept failed");
      continue;
    }
    handlerClient(client_fd, count);
    close(client_fd);
    if (count >= 10000) {
      cout << "worker_id[" << worker_id << "] deal_1w_request" << endl;
      count = 0;
    }
  }
}

void usage() {
  cout << "ProcessPool2 -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  for (int i = 0; i < GetNProcs(); i++) {
    pid_t pid = fork();
    if (pid < 0) {
      perror("fork failed");
      continue;
    }
    if (0 == pid) {
      handler(i, ip, port);  // 子进程陷入死循环，处理客户端请求
      exit(0);
    }
  }
  while (true) sleep(1);  // 父进程陷入死循环
  return 0;
}

在main函数中，我们根据系统当前可用的CPU核数，预先创建数量与之相等的子进程。每个子进程都创建自己的socket套接字，设置SO_REUSEPORT选项，并在相同的网络地址开启监听。最后，每个子进程都陷入死循环，等待客户端请求的到来，并为其提供服务。

5.1.6 ThreadPool

在线程池的并发模型中，我们预先创建指定数量的线程，每个线程都不退出，一直等待客户端连接的到来，并为其提供服务。这种方式可以避免频繁地创建和销毁线程，提高系统的性能和效率，同时也可以降低系统的开销和资源占用。线程池并发模型的代码如下所示。

#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>

#include <iostream>
#include <thread>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

void handlerClient(int client_fd) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      return;
    }
    if (not SendMsg(client_fd, msg)) {
      return;
    }
  }
}
void handler(string ip, int64_t port) {
  // 开启SO_REUSEPORT选项
  int sock_fd = CreateListenSocket(ip, port, true);
  if (sock_fd < 0) {
    return;
  }
  while (true) {
    int client_fd = accept(sock_fd, NULL, 0);
    if (client_fd < 0) {
      perror("accept failed");
      continue;
    }
    handlerClient(client_fd);
    close(client_fd);
  }
}

void usage() {
  cout << "ThreadPool -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  for (int i = 0; i < GetNProcs(); i++) {
    std::thread(handler, ip, port).detach();  // 这里需要调用detach，让创建的线程独立运行
  }
  while (true) sleep(1);  // 主线程陷入死循环
  return 0;
}

在main函数中，我们根据系统当前可用的CPU核数，预先创建数量与之相等的线程。每个线程都创建自己的socket套接字，设置SO_REUSEPORT选项，并在相同的网络地址开启监听。最后，每个线程都陷入死循环，等待客户端请求的到来，并为其提供服务。

5.1.7 LeaderAndFollower

线程池的并发模型中，线程之间的关系是对等的。领导者/跟随者的并发模型是线程池的一种变种，一开始所有的线程都是follower，它们会竞争上岗，有一个线程成为leader。leader线程会监听客户端连接的到来，接受到客户端的连接时，它会放弃领导权，由其他follower去竞争。此时leader线程变成了worker线程，为新来的客户端提供服务。领导者/跟随者并发模型的代码如下所示。

#include <sys/socket.h>
#include <unistd.h>

#include <iostream>
#include <mutex>
#include <thread>

#include "../../common/cmdline.h"
#include "../../common/utils.hpp"

using namespace std;
using namespace MyEcho;

std::mutex Mutex;

void handlerClient(int client_fd) {
  string msg;
  while (true) {
    if (not RecvMsg(client_fd, msg)) {
      return;
    }
    if (not SendMsg(client_fd, msg)) {
      return;
    }
  }
}

void handler(int sock_fd) {
  while (true) {
    int client_fd;
    // follower等待获取锁，成为leader
    {
      std::lock_guard<std::mutex> guard(Mutex);
      client_fd = accept(sock_fd, NULL, 0);  // 获取锁，并获取客户端的连接
      if (client_fd < 0) {
        perror("accept failed");
        continue;
      }
    }
    handlerClient(client_fd);  // 处理每个客户端请求
    close(client_fd);
  }
}

void usage() {
  cout << "LeaderAndFollower -ip 0.0.0.0 -port 1688" << endl;
  cout << "options:" << endl;
  cout << "    -h,--help      print usage" << endl;
  cout << "    -ip,--ip       listen ip" << endl;
  cout << "    -port,--port   listen port" << endl;
  cout << endl;
}

int main(int argc, char* argv[]) {
  string ip;
  int64_t port;
  CmdLine::StrOptRequired(&ip, "ip");
  CmdLine::Int64OptRequired(&port, "port");
  CmdLine::SetUsage(usage);
  CmdLine::Parse(argc, argv);
  int sock_fd = CreateListenSocket(ip, port, false);
  if (sock_fd < 0) {
    return -1;
  }
  for (int i = 0; i < GetNProcs(); i++) {
    std::thread(handler, sock_fd).detach();  // 这里需要调用detach，让创建的线程独立运行
  }
  while (true) sleep(1);  // 主进程陷入死循环
  return 0;
}

在 main 函数中，首先会开启监听，然后根据系统当前可用的 CPU 核数，预先创建相同数量的线程。主线程会进入一个死循环，而所有从线程都会尝试获取锁。获取到锁的线程会开始监听客户端连接的到来，一旦有客户端连接到来，该线程会释放锁，并开始处理客户端的请求。其他线程则会继续尝试获取锁，以等待下一个客户端连接的到来。每个从线程的状态迁移如下图所示。

5.1.8 性能压测与对比

压测运行环境：云开发机，16核32G的CentOS主机，每个CPU的频率为2.59GHz。（被压测的服务和压测工具都运行在同一台云主机上）

压测的参数为：一个线程、单个请求包1k、单客户端单次最多请求40次、64个客户端、压测60秒、限频20万qps。

压测命令：BenchMark -ip 0.0.0.0 -port 1688 -thread_count 1 -max_req_count 40 -pkt_size 1024 -client_count 64 -run_time 60 -rate_limit 200000

所有简单模型压测后的性能指标如下表所示。

从上表的数据中，我们可以得出以下结论。

• 我们可以看到单进程的并发模型性能是最差的。
• 多进程和多线程的并发模型，由于能使用到多个CPU，所以性能有所提升，但是因为需要频繁的创建和销毁进程和线程，接口的pct50和pct95耗时较高。
• 进程池1和进程池2的并发模型，由于没有频繁的创建和销毁进程和线程的损耗，性能比多进程和多线程的并发模型高，接口的pct50和pct95耗时也更低。
• 进程池2并发模型接口的pct999耗时比进程池1并发模型的高出不少，这个是因为进程池2的并发模型是由操作系统来做负载均衡的，但这个策略并无法保证对流量负载做完美的均分，导致接口长尾的耗时较高，而进程池1的并发模型是多进程抢锁，每个进程的流量负载会更均衡，但因为有锁，所以进程池1的并发模型性能比进程池2的并发模型低一些。
• 线程池的并发模型和进程池2的并发模型，性能差异并不是很大，因为线程池的并发模型也是由操作系统来做负载均衡的，所以存在接口长尾的耗时较高的情况。
• 领导者/跟随者的并发模型和进程池1的并发模型很相似，这两个模型所有的指标都差异很小，领导者/跟随者的并发模型可以看到显式的使用锁，而进程池1的并发模型没有。

5.2 多路I/O复用简单模型

5.2.1 Select

多路I/O复用最早支持的系统调用是select函数，select函数存在天然的缺陷，因为select函数最多支持对1024个文件描述的事件进行监听，且文件描述符的值必须小于1024，而1024这个值是由FD_SETSIZE这个宏来决定的。使用select函数实现的并发模型代码如下所示。

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
0

在main函数中，使用select函数来监听所有客户端连接上的可读事件，如果客户端文件描述符值大于等于FD_SETSIZE，则直接关闭客户端的链接。当客户端连接上有可读事件发生时，就处理对应客户端连接上的请求，直到客户端关闭连接或者读写失败。

5.2.2 Poll

为了解决select函数文件描述符数量和值的限制，Linux 系统后续新增了poll函数。与select不同的是，poll函数在没有触碰到系统其他限制之前，理论上只要内存充足，则可以支持监听的文件描述符数量是没有上限的。使用poll函数实现的并发模型代码如下所示。

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
1

在main函数中，使用poll函数来监听所有客户端连接上的可读事件，当客户端连接上有可读事件发生时，就处理对应客户端连接上的请求，直到客户端关闭连接或者读写失败。

5.2.3 Epoll

epoll是poll的一种变种，它通过事件注册和通知机制，有效提升了事件监听效率，并且对更大数量文件描述符的监听有更好的可扩展性。相比于poll和select，epoll更加高效，能够处理更多的连接。使用epoll函数实现的并发模型代码如下所示。

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
2

在main函数中，使用epoll函数来监听所有客户端连接上的可读事件，当客户端连接上有可读事件发生时，就处理对应客户端连接上的请求，直到客户端关闭连接或者读写失败。

5.2.4 性能压测与对比

压测运行环境：云开发机，16核32G的CentOS主机，每个CPU的频率为2.59GHz。（被压测的服务和压测工具都运行在同一台云主机上）

压测的参数为：一个线程、单个请求包1k、单客户端单次最多请求40次、64个客户端、压测60秒、限频20万qps。

压测命令：BenchMark -ip 0.0.0.0 -port 1688 -thread_count 1 -max_req_count 40 -pkt_size 1024 -client_count 64 -run_time 60 -rate_limit 200000

多路I/O复用简单模型压测后的性能指标如下表所示。

从上表的数据中，我们可以得出以下结论。

如果仅仅只是把IO复用的事件监听当做简单的客户端处理的触发器，则这3个并发模型和单进程的并发模型的所有指标没什么差异，它们在同一时间只有一个客户端请求在被处理。

当然poll和epoll可以支持更多的客户端连接。

5.3 Reactor相关模型

最后介绍的10种不同的并发模型，都是Reactor模型相关的，最基础的Reactor并发模型如下图所示。

5.3.1 SelectReactorSingleProcess

使用select函数可以实现Reactor的并发模型，相关的代码如下所示。

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
3

在main函数中，我们使用select函数来监听多个客户端连接上的事件，因为select函数最大支持FD_SETSIZE个文件描述符，所以对于不支持的客户端连接，则直接关闭连接。select函数每次调用之前都需要重新更新一遍要监听的读写事件的文件描述符的集合。

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
4

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
5

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
6

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
7

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
8

template <typename Function, typename... Args>
int CoroutineCreate(Schedule& schedule, Function&& f, Args&&... args) {
  int id = 0;
  for (id = 0; id < schedule.coroutineCnt; id++) {
    if (schedule.coroutines[id]->state == Idle) break;
  }
  if (id >= schedule.coroutineCnt) {
    return kInvalidRoutineId;
  }
  Coroutine* routine = schedule.coroutines[id];
  std::function<void()> entry = std::bind(std::forward<Function>(f), std::forward<Args>(args)...);
  CoroutineInit(schedule, routine, entry);
  return id;
}
9

root@centos BenchMark $ ./BenchMark -h
BenchMark -ip 0.0.0.0 -port 1688 -thread_count 1 -max_req_count 100000 -pkt_size 1024 -client_count 200 -run_time 60 -rate_limit 10000 -debug
options:
    -h,--help                      print usage
    -ip,--ip                       service listen ip
    -port,--port                   service listen port
    -thread_count,--thread_count   run thread count
    -max_req_count,--max_req_count one connection max req count
    -pkt_size,--pkt_size           size of send packet, unit is byte
    -client_count,--client_count   count of client
    -run_time,--run_time           run time, unit is second
    -rate_limit,--rate_limit       rate limit, unit is qps/second
    -debug,--debug                 debug mode, more info print
0

root@centos BenchMark $ ./BenchMark -h
BenchMark -ip 0.0.0.0 -port 1688 -thread_count 1 -max_req_count 100000 -pkt_size 1024 -client_count 200 -run_time 60 -rate_limit 10000 -debug
options:
    -h,--help                      print usage
    -ip,--ip                       service listen ip
    -port,--port                   service listen port
    -thread_count,--thread_count   run thread count
    -max_req_count,--max_req_count one connection max req count
    -pkt_size,--pkt_size           size of send packet, unit is byte
    -client_count,--client_count   count of client
    -run_time,--run_time           run time, unit is second
    -rate_limit,--rate_limit       rate limit, unit is qps/second
    -debug,--debug                 debug mode, more info print
1

root@centos BenchMark $ ./BenchMark -h
BenchMark -ip 0.0.0.0 -port 1688 -thread_count 1 -max_req_count 100000 -pkt_size 1024 -client_count 200 -run_time 60 -rate_limit 10000 -debug
options:
    -h,--help                      print usage
    -ip,--ip                       service listen ip
    -port,--port                   service listen port
    -thread_count,--thread_count   run thread count
    -max_req_count,--max_req_count one connection max req count
    -pkt_size,--pkt_size           size of send packet, unit is byte
    -client_count,--client_count   count of client
    -run_time,--run_time           run time, unit is second
    -rate_limit,--rate_limit       rate limit, unit is qps/second
    -debug,--debug                 debug mode, more info print
2