最新资讯

1. 最简单的服务器模型

在学习 Linux 网络编程时，我们通常接触的第一个服务器模型，就是单线程阻塞模型。只看它的名字其实已经能够明白这个模型的原理了，它的逻辑简单清晰：先创建一个 socket，再绑定端口，然后监听，最后在一个死循环里等待客户端连接。

1.1 单线程阻塞echo服务器

我们先来写一个最最基础的 echo 服务器，它的功能非常简单：就是把客户端发来的任何消息原封不动地再发回去。这也是为什么叫他echo服务器的原因。

#include 
#include 
#include 
#include 
#include 
#include 

#define PORT 8080
#define BUFFER_SIZE 1024

int main() {
    int listen_fd, conn_fd;
    struct sockaddr_in serv_addr;
    char buffer[BUFFER_SIZE];

    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    if(listen_fd == -1)
    {
        perror("socket listen");
        exit(1);
    }
    
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));//设置端口可复用
    
    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_addr.sin_port = htons(PORT);
    
    if(bind(listen_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) == -1)
    {
        perror("bind");
        exit(1);
    }
    
    if(listen(listen_fd, 5) == -1)
    {
        perror("listen");
        exit(1);
    }
    
    printf("服务器已启动，等待连接...
");

    conn_fd = accept(listen_fd, NULL, NULL);//阻塞等待客户端链接
    if(conn_fd == -1)
    {
        perror("accept");
        exit(1);
    }
    printf("客户端已连接
");

    ssize_t n;
    while ((n = read(conn_fd, buffer, BUFFER_SIZE)) > 0) //这里是阻塞的
    {
        write(conn_fd, buffer, n);//回写
    }

    close(conn_fd);
    close(listen_fd);
    return 0;
}

这段代码虽然简单，但它完整地展示了 Linux TCP 服务器的经典的五个步骤：socket -> bind -> listen -> accept -> read/write。

其中还有几个值得注意的细节，我简单说一下：

第一个细节：端口复用。在socket之后，我加入了setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)); 。这一步至关重要，它允许我们的服务器在异常退出后，能够立即重启并重新绑定 8080 端口，而不会遇到 "Address already in use" 的错误。这是所有服务器的必备配置。

第二个细节：网络字节序。在填充地址结构体serv_addr时，PORT 和 INADDR_ANY 都要被 htons/htonl 函数处理。这是为了解决大小端问题。不同架构的 CPU 存储多字节整数的顺序可能不同。TCP/IP 协议栈规定了统一的网络字节序（大端）。这两个函数就是用来保证我们的数据符合标准，避免跨平台通信时出现数据解析错误。

第三个细节：这个模型最大的特点就是阻塞。程序首先会阻塞在 accept()，等待第一个客户端连接。连接成功后，程序会进入 while 循环，阻塞在 read()，等待客户端发送数据。正是这两个阻塞点，导致了它一次只能服务一个客户端的致命缺陷，也为我们后续的并发模型方向埋下了伏笔。

此外，为了展示最简单的服务器的功能到底有多么粗糙，我特意将 accept() 函数放在循环外面，这也就意味着，这个服务器一生只能服务一个客户端，当这个客户端断开连接时，服务器的运行也会结束。

下面我们来看看这个最简单的服务器到底都有哪些缺陷。

1.2 实验：致命的瓶颈

这个服务器看起来是能跑的，但他有一个致命的缺陷。我们可以通过一个简单的实验来看一下：

先运行上面的示例程序，就会显示 “服务器已启动…” 这些字，然后再开一个终端输入下面的命令：

telnet localhost 8080

就会显示 “客户端已连接” 。此时我在客户端输入一个 “hello” ，然后按回车，可以看到服务器在接收到 “hello” 之后又把他原封不动的发回给客户端了：

到这里，单线程阻塞echo服务器的功能就已经看到了。但如果我们再开一个终端尝试连接服务器呢？会发生什么事？

这里，我又开了一个终端连接服务器，如果尝试过的读者会发现这次服务器的那个终端并没有弹出 “客户端已连接” 这个表示链接成功的消息。此外，还能发现在这个新开的终端连接服务器，看似和前面那个客户端是相同的场景，但我们在输入 “hello” 并按下回车后，服务器并没有返回消息。这已经能够证明我们在这个终端上其实并没有连接到服务器。

为什么呢？

这里暴露了两个致命问题：

第一是无法并发：服务器的主线程在 accept 之后，就一头扎进了与第一个客户端通信的 while 循环里。只要这个客户端不主动断开，程序就永远没有机会去处理新的连接请求。

第二是一次性服务：更糟糕的是，一旦第一个客户端断开连接，while 循环结束，整个服务器程序也会跟着退出。

这种单线程、单任务、一次性的模型，在任何实际应用中都是不可接受的。我们需要让服务器：能同时服务多个客户端 (并发)，还能 7x24 小时运行，而不是服务完一个就退出。

2. 多进程模型

为了解决单线程模型的两大缺陷，我们自然而然地想到了 Linux 提供的多任务机制。最经典、最简单的就是多进程。

2.1 改进思路

前面的单线程阻塞模型问题主要出在无法并发和无法多次服务，因此我们主要针对这两点进行改进。

父进程，也就是主进程，它在一个while循环里面，主要负责接收 accept() ，也就是接收客户端连接请求。

子进程，负责处理客户端发来的信息。

他们的分工，我在这里简要总结一下：父进程一旦接收到客户端连接请求，立刻fork出一个子进程，然后把这个客户端交给子进程去处理，父进程自己继续去等待其他客户端了连接。

从这个改进思路上来看，已经实现了并发和多次服务的两个目标。下面我们来关注一下这个模型的实现。

2.2 多进程 echo 服务器

上面已经讲了改进思路，我们废话不多说，直接贴出代码：

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define PORT 8080
#define BUFFER_SIZE 1024

int main() 
{
    int listen_fd, conn_fd;
    struct sockaddr_in serv_addr;
    char buffer[BUFFER_SIZE];
    pid_t pid;

    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
	if(listen_fd == -1)
	{
		perror("socket listen");
		exit(1);
	}

	memset(&serv_addr,0,sizeof(serv_addr));
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_port = htons(PORT);
	serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    
	if(bind(listen_fd,(struct sockaddr *)&serv_addr,sizeof(serv_addr)) == -1)
	{
		perror("bind");
		exit(1);
	}

    if(listen(listen_fd, 5) == -1)
	{
		perror("listen");
		exit(1);
	}

    printf("服务器已启动，等待连接...
");

    while (1) 
    { 
        conn_fd = accept(listen_fd, NULL, NULL);
        if (conn_fd < 0) 
		{
            perror("accept");
            continue; //出错了没事，继续等下一个
        }

        pid = fork();

        if(pid < 0) 
		{
            perror("fork");
            close(conn_fd);
            continue;
        }
		else if(pid == 0) //子进程
		{
            close(listen_fd); 

            printf("子进程%d为新客户端服务(fd=%d)...
", getpid(), conn_fd);

            ssize_t n;
            while (((n = read(conn_fd, buffer, BUFFER_SIZE))) > 0) 
			{
                write(conn_fd, buffer, n);//回写的字节数一定要是read读到的字节数
            }
            
            printf("客户端断开，子进程%d退出。
", getpid());
            close(conn_fd);
            exit(0); 
        } 
		else 
		{
            close(conn_fd); 
        }
    }

    close(listen_fd); // 理论上走不到这里,因为父进程一直在循环里面
    return 0;
}

运行代码，看看效果怎么样：

这次我运行服务器后，直接开了三个终端连接上服务器，可以看到，服务器所在终端已经弹出了客户端已连接成功的消息：

我们在来看看这三个客户端的情况：

通过运行结果可以看到，这三个客户端都可以收到服务器回传的消息，说明我们已经达到了开始的目标。

到这儿，可能有人会提出问题，为什么服务器端显示连接上的三个客户端的文件描述符 fd 都是 4 呢？其实这并不是发生了错误，而是一个正常现象，来听我详细解释：

当父进程调用 fork 时，内核会为子进程创建一个几乎一模一样的内存空间和上下文，在这里我们只关注于文件描述符表。每个进程都有自己独立的文件描述符表，在 fork 的那一瞬间，子进程也得到了和父进程一模一样的文件描述符表。

好，理论讲完了，我们来关注一下过程，到底是什么原因导致了三个相同的文件描述符呢？

首先，在进程中，文件描述符的 0，1，2 已经默认被标准输入，标准输出和标准错误占据了，这时，进程进行 socket ，把返回的值（也就是3）赋给 listen_fd 。

然后，当第一个客户端连接时，父进程调用 accept ，返回一个新的 conn_fd ，由于 3 已经被占用了，所以 conn_fd 只能为 4。

之后，父进程调用fork创建第一个子进程（PID：11375），这个子进程得到了父进程完整的文件描述符表，子进程的fd=4也指向了和父进程相同的 TCP 连接。

接着，printf("子进程%d为新客户端服务(fd=%d)... ", getpid(), conn_fd);打印出的fd自然就是 4 了，因为在父进程那就是 4，从来没变过。

再然后，父进程关闭了自己的 fd=4 ，因为这对他没用，同样 listen_fd对子进程也没用，所以子进程关了 listen_fd 。

这时，当第二个客户端（PID = 11422）请求连接时，父进程的fd=4又空出来了，于是在父进程这里，第二个客户端有拿到了 fd=4 这个文件描述符，然后再次fork，子进程拿到的也是文件描述符fd=4对应的第二个客户端…

所以说，父进程每次将客户端交给子进程后，就会关闭文件描述符 4 ，这个文件描述符 4 空闲出来，下一次有客户端连接用的又是文件描述符 4 ，因此才导致了上面的情况。所以说，这其实是正常现象。

2.3 僵尸进程

多进程模型完美解决了并发问题，看起来似乎无懈可击。但它引入了一个新的、更隐蔽的、可能导致严重后果的东西——**僵尸进程 **。

我们来分析一下上面的程序，如果客户端正常断开连接，那么对应的子进程会执行exit(0)，子进程退出后，内核会保留它的 PID 和退出状态等信息，等待父进程过来收尸，但是，父进程此时正忙着在while循环里面accept新的连接，压根就没有时间。

最后，这些无人认领的尸体会越来越多，堆积在内核的进程表里。虽然它们不占很多内存，但会耗尽 PID 资源。当 PID 用完，系统将无法创建任何新进程，最终只能重启。

下面，我们来验证一下僵尸进程的存在：

我来解释一下这三张图，第一张图中我退出了 PID 为 11375 的子进程负责的客户端，然后又以相同的方式退出了其余的两个子进程负责的客户端。

第二张图显示了三个子进程已经执行exit(0);退出了。而父进程还在继续运行，此时父进程在accept等待新的客户端连接，并没有为子进程收尸，因此他们变成了僵尸进程。

第三张图是我查询僵尸进程的结果，图中显示，PID为 11375 ， 11422 ， 11490 的三个进程此时的状态（STAT）为 Z+，如果一个进程的状态为Z，表示这个进程是僵尸进程，如果是Z+，表示这个进程是一个属于前台进程组的僵尸进程。

一般情况下，如果想要清理一个僵尸进程，最好的办法是杀死它的父进程。但这种方法在这里显然是不行的。

试想一下，一台服务器正在运行，由于其中的几个进程退出后没有及时清理，产生了僵尸进程，而此时你想关闭这台服务器来进行清理僵尸进程，这等于直接断开正在与服务器通信的所有客户端，造成的损失是不可估量的。

下面我介绍一种比较优雅的方法，在服务器端不退出的情况下回收子进程。

2.4 SIGCHLD 信号

如何让父进程在不阻塞 accept 的情况下，及时回收子进程呢？答案是信号。

当一个子进程终止时，内核会给其父进程发送一个 SIGCHLD 信号。我们只需要为父进程注册一个处理这个信号的函数即可。

我们需要引入头文件，然后在main函数的前面写一个sigchld_handler函数，此外，还要在main函数里面注册这个信号，放在main函数第一行就行。其他部分不需要改变。

为了简便，我这里只给出最关键的部分，补全到上面多进程模型的代码里面就行了：

#include  

void sigchld_handler(int sig) 
{
    while (waitpid(-1, NULL, WNOHANG) > 0);//非阻塞等待所有子进程
}

int main()
{
    signal(SIGCHLD, sigchld_handler);
    
    //........中间部分不变........
    
    return 0;
}

代码修改之后，我又重新运行了一次。这次我直接创建了三个客户端然后退出。此时父进程还在运行，去查一下发现并没有僵尸进程，这就说明我们的信号处理方法已经生效了。

2.5 补充知识

考虑到可能会有人好奇，从内核向父进程发送SIGCHLD信号到父进程回收僵尸进程，这中间到底是怎样运转的。我增加了这一节，下面详细解释一下。

当子进程执行完毕退出后，内核将其标记为僵尸进程，然后内核去查找该子进程的父进程，并向父进程发送SIGCHLD信号。

父进程此时大概率阻塞在 accept() 上，当SIGCHLD信号到达时，内核会中断父进程的行为，accept() 会立即返回 -1，并将 errno 设置为 EINTR。除此之外，内核还会修改父进程的程序计数器（PC）和栈（STACK） ，使得父进程醒来后能从我们的信号处理函数sigchld_handler开始执行。当信号处理函数执行完 return 时，实际上是跳到了内核预设的一段跳板代码，这段代码会调用 sys_sigreturn 系统调用，通知内核恢复父进程之前的上下文，从而让进程醒来后继续从被SIGCHLD信号中断的地方执行。

而在信号处理函数内部，我们调用了 waitpid(-1, NULL, WNOHANG)。-1表示回收任何一个已经终止的子进程，WNOHANG表示非阻塞，没有僵尸进程回收的话就直接返回，防止父进程的main函数那边等太长时间。

为了方便大家理解这个过程，我们可以在前面代码的基础上进行修改，如下所示，在perror上面加了一个if判断：

         conn_fd = accept(listen_fd, NULL, NULL);
		if (conn_fd < 0) 
		{
			if(errno == EINTR)
			{
				printf("accept被SIGCHLD信号中断了，这是正常现象，程序继续运行...
");
			}
			perror("accept");
            continue; //出错了没事，继续等下一个
        }

但是就这样printf的内容是不会按照我们预想的那样打印出来的。因为在默认的signal情况下，内核会让父进程回到accept执行之前的状态，相当于重新运行accept，如果这次没有被SIGCHLD信号中断的话，accept也不会返回-1，errno也不会被置为EINTR，退一万步说，就算再次被SIGCHLD信号中断，下次父进程醒来还是要重新执行accept，也就是说在这种模式下printf的内容是永远不会被打印出来的。

要想看到printf的内容，证明errno确实被置为EINTR了，我们必须采用更高级的信号处理方式。为了方便大家复现，这次给出完整代码吧：

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define PORT 8080
#define BUFFER_SIZE 1024

void sigchld_handler(int sig)
{
	while(waitpid(-1,NULL,WNOHANG)>0);
}

int main() 
{
    //--------------------重点关注这里-----------------//
	struct sigaction sa;
	sa.sa_handler = sigchld_handler;
	sigemptyset(&sa.sa_mask);
	sa.sa_flags = 0;

	if (sigaction(SIGCHLD, &sa, NULL) == -1) 
	{
        perror("sigaction error");
        exit(1);
    }
    //------------------------------------------------//

    int listen_fd, conn_fd;
    struct sockaddr_in serv_addr;
    char buffer[BUFFER_SIZE];
    pid_t pid;

    listen_fd = socket(AF_INET, SOCK_STREAM, 0);
	if(listen_fd == -1)
	{
		perror("socket listen");
		exit(1);
	}

	memset(&serv_addr,0,sizeof(serv_addr));
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_port = htons(PORT);
	serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    
	if(bind(listen_fd,(struct sockaddr *)&serv_addr,sizeof(serv_addr)) == -1)
	{
		perror("bind");
		exit(1);
	}

    if(listen(listen_fd, 5) == -1)
	{
		perror("listen");
		exit(1);
	}

    printf("服务器已启动，等待连接...
");

    while (1) { 
        conn_fd = accept(listen_fd, NULL, NULL);
        if (conn_fd < 0) 
		{
            //-------------------------------还有这里-----------------
			if(errno == EINTR)
			{
				printf("accept被SIGCHLD信号中断了，这是正常现象，程序继续运行...
");
			}
            //--------------------------------------------------------
			perror("accept");
            continue; //出错了没事，继续等下一个
        }

        pid = fork();

        if(pid < 0) 
		{
            perror("fork");
            close(conn_fd);
            continue;
        }
		else if(pid == 0) 
		{
            close(listen_fd); 

            printf("子进程%d为新客户端服务(fd=%d)...
", getpid(), conn_fd);

            ssize_t n;
            while (((n = read(conn_fd, buffer, BUFFER_SIZE))) > 0) 
			{
                write(conn_fd, buffer, n);
            }
            
            printf("客户端断开，子进程%d退出。
", getpid());
            close(conn_fd);
            exit(0); 
        } 
		else 
		{
            close(conn_fd); 
        }
    }

    close(listen_fd); // 理论上走不到这里
    return 0;
}

运行后结果如下，这次我们可以看到，printf的内容和perror的内容都被打印出来了，因为默认情况下 signal() 会隐含 SA_RESTART 标志，导致系统调用自动重启。我们把sa.sa_flags设为 0 表示不自动重启，我们才能在 accept 返回处捕获 EINTR。

3. 多线程模型

在解决了僵尸进程后，我们的多进程服务器已经非常健壮了。但是，如果我们要面对成千上万个并发连接，每来一个连接就 fork() 一个进程，这种方法开销实在太大了。

我们需要一种更轻量的方案——多线程。

3.1 改进思路

进程拥有独立的内存空间（堆、栈、代码段）。创建一个进程，内核需要复制页表、文件描述符表等，成本昂贵。

线程共享进程的内存空间。创建一个线程，只需要分配一个小小的栈空间，成本极低。

了解了进程和线程的区别，我们的改进思路就非常明确了，如下：

主线程负责accept()等待客户端连接。每当一个客户端连接上，就pthread_create()一个工作线程去负责。

工作线程负责处理对应客户端的信息。

3.2 线程模型的优缺点

线程模型的优点：

线程创建速度比进程快 10~100 倍。

线程间共享全局变量和堆内存，不需要复杂的 IPC（管道、共享内存等）。

线程模型的缺点：

一个线程崩溃，整个进程（包括所有其他线程）都会挂掉。而多进程模型中，一个子进程崩了，不影响父进程和其他子进程。

共享资源（如全局变量）需要加锁，写不好容易死锁或数据竞争。

虽然比进程轻，但如果是 1 万个连接，依然需要 1 万个线程。操作系统的调度器会被累死，内存也会被撑爆。在 Linux 中，每个新线程的默认栈大小通常为 8 MB，1万个线程栈就是 80 GB。

3.3 小总结

整体上来看，多线程模型和多进程模型的思路其实是相同的，多线程模型虽然减轻了创建开销，但一个连接一个线程 的本质没有变。面对海量并发，我们必须彻底抛弃这种方式，转向一种全新的思维模式。

下一章要介绍的就是 Linux 网络编程的终极武器——I/O 多路复用 (epoll)。

4. I/O多路复用

虽然多线程模型减轻了创建和销毁的开销，但本质上它依然遵循一个连接一个线程的思路。当并发连接数上升到 1 万甚至 100 万时，这种模型会面临两个无法逾越的障碍：

第一个是内存爆炸，1 万个线程，每个线程栈 8MB，光栈空间就需要 80GB 内存。

第二个是调度崩溃，操作系统在成千上万个线程之间进行上下文切换，CPU 的时间全花在切换上，而不是干正事儿上。

因此，我们需要一种机制，能够用一个线程监控成千上万个连接。这就是 I/O 多路复用。

4.1 从线性轮询到事件驱动

在 epoll 出现之前，Linux 使用 select 和 poll。它们虽然能实现多路复用，但在内核实现上存在本质的缺陷。

select 和 poll 的工作机制非常原始。每次调用它们，都需要将所有要监控的文件描述符集合从用户态完整拷贝到内核态。

内核在检测时，会遍历整个集合**（O(N) 复杂度）。如果有事件发生，内核只是简单地标记有事件已经就绪，但不会告诉说明到底是哪一个。这也就是说，select 返回后，用户态程序还需要再遍历一遍**整个集合，才能找到真正需要处理的那个文件描述符。

当连接数达到 10万，但活跃连接只有 1 个时，select 依然要空转 10万次，这造成了大量的性能浪费。并且select 受限于 FD_SETSIZE (默认是 1024)，是无法处理海量连接的。

而epoll的出现完美解决了上面的问题。epoll在内核设计了两个核心的数据结构：

第一个是红黑树，epoll_create 会在内核开辟一块空间，当我们调用 epoll_ctl 添加 socket 时，内核会将这个 socket 节点插入到一棵红黑树中。红黑树的查找、插入、删除复杂度均为 O(logN)，即使管理百万级连接，效率依然极高。且这棵树一直在内核里面，无需像 select 那样每次都要重复拷贝文件描述符集合。

第二个是就绪链表，内核为红黑树中的每个 socket 注册了一个回调函数。当网卡收到数据，中断处理程序触发回调，自动将该 socket 节点拷贝到一个**双向链表（也就是就绪链表）**中。

而**epoll_wait要干什么呢？它只需要检查那个就绪链表是否为空**，如果不为空，直接把链表里的节点复制回用户态。

下面我列个表格对比一下，这几种I/O多路复用的方式：

4.2 核心代码实现

下面是一个基于 epoll 的单线程并发服务器。请注意观察，代码中不再有 fork 或 pthread_create，所有的连接都在同一个 while(1) 循环中被串行处理。

#include 
#include 
#include 
#include 
#include 
#include 
#include 

#define PORT 8080
#define BUFFER_SIZE 1024
#define MAX_EVENTS 10

int main()
{
	int listen_fd,conn_fd,epoll_fd,n_read;
	struct sockaddr_in server_addr;
	char buffer[BUFFER_SIZE];

	struct epoll_event ev,events[MAX_EVENTS];

	listen_fd = socket(AF_INET,SOCK_STREAM,0);
	if(listen_fd == -1)
	{
		perror("socket listen");
		exit(1);
	}
	printf("监听socket创建成功
");

	int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));//设置端口可复用

	memset(&server_addr,0,sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(PORT);
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);

	if(bind(listen_fd,(struct sockaddr *)&server_addr,sizeof(server_addr)) == -1)
	{
		perror("bind");
		exit(1);
	}
	printf("绑定成功
");

	if(listen(listen_fd,5) == -1)
	{
		perror("listen");
		exit(1);
	}

	epoll_fd = epoll_create1(0);
	if(epoll_fd == -1)
	{
		perror("epoll_create");
		exit(1);
	}
	printf("epoll实例创建成功
");

	ev.events = EPOLLIN;
	ev.data.fd = listen_fd;
	if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,listen_fd,&ev) == -1)
	{
		perror("epoll_ctl");
		exit(1);
	}
	printf("listen_fd添加成功
");
	printf("服务器正在监听%d端口....
",PORT);

	while(1)
	{
		n_read = epoll_wait(epoll_fd,events,MAX_EVENTS,-1);//阻塞等待
		if(n_read == -1)
		{
			perror("epoll_wait");
			exit(1);
		}

		for(int i=0;i<n_read;i++)
		{
			if(events[i].data.fd == listen_fd)//有新的连接到来
			{
				conn_fd = accept(listen_fd,(struct sockaddr *)NULL,NULL);
				printf("接受新的连接，fd = %d
",conn_fd);
				ev.events = EPOLLIN;
				ev.data.fd = conn_fd;
				epoll_ctl(epoll_fd,EPOLL_CTL_ADD,conn_fd,&ev);
			}
			else
			{
				int client_fd = events[i].data.fd;
				ssize_t n = read(client_fd,buffer,BUFFER_SIZE);

				if(n == 0)
				{
					printf("客户端fd = %d关闭了连接
",client_fd);
					epoll_ctl(epoll_fd,EPOLL_CTL_DEL,client_fd,NULL);
					close(client_fd);
				}
				else if(n > 0)
				{
					write(client_fd,buffer,n);
				}
				else
				{
					perror("read");
					close(client_fd);
				}
			}
		}
	}
	close(listen_fd);
	return 0;
}

这段代码才用的是LT（水平触发）模式，在while循环之前，我们需要把监听socket（listen_fd）添加到我们需要关注的事件中。

下面简要解释一下这段代码的核心逻辑。

进入while循环后，epoll_wait就会阻塞等待，当它返回时，events数组中就存放着真正有事件发生的文件描述符。相比于 select 需要遍历整个文件描述符集合，epoll 直接把需要处理的文件描述符单独拎出来给我们，这就是它在海量连接下性能不衰减的原因。

然后我们会进入for循环，它会把events数组中需要处理的文件描述符一个一个拿出来，判断他们到底是哪种类型的，是有新的连接到来，还是说旧的连接发消息了，这个判断过程体现在if判断中。

如果有新的连接到来，那我们就accept，并且把这个新链接上的文件描述符添加到我们需要关注的事件中。

如果是旧的连接在发消息，那我们就先read，把这个消息存放在buffer中。在这之后，我们必须要判断一下read的返回值，因为客户端退出会体现在这里。如果客户端退出，read会返回 0，这时我们要把相应的文件描述符从我们需要关注的事件中删除，然后关闭这个文件描述符。如果是正常读写，那我们就再给他回写过去。如果真的出错，那就提示就完事儿了。

4.3 效果展示与新的瓶颈

为了展示这个单线程epoll的效果怎么样，我直接开了三个客户端连接它，并且发送hello都可以正常回写，运行结果如下：

但是请看这里：

				else if(n > 0)
				{
					write(client_fd,buffer,n);
				}

这是我们在接收到客户端消息后进行的处理，只有一条简单的write系统调用。试想一下，如果这里的业务逻辑变成了视频解码，或者复杂的数据库查询，需要消耗 1 秒，会发生什么情况呢？

因为我们只有一个线程在死循环里处理任务。当它正在处理那个耗时 1 秒的任务时，epoll_wait 无法被调用，新的连接进不来，其他客户端的消息也卡在缓冲区里无法处理。整个服务器在这一秒时间里会处于 “假死” 状态。

单线程 epoll 虽然解决了 I/O 的并发，但无法利用多核 CPU 处理计算密集型任务。

因此，我们需要将 epoll (只负责 I/O) 与 线程池 (只负责处理) 结合起来。这就是最终形态——Reactor 模型。

5. epoll + 线程池

为了解决单线程 epoll 无法利用多核 CPU，且计算任务会阻塞 I/O 这个痛点，我们需要将 I/O 处理 和 业务逻辑 彻底分离。这也是现代高性能服务器（如 Nginx，Netty）通用的 Reactor 模型。

5.1 模型设计

我们引入线程池 来负责具体的业务计算，主线程只负责 I/O 调度。

主线程：负责 epoll_wait、accept 新连接、read 数据。收到数据后，立刻封装成一个任务，扔进任务队列。

任务队列：连接主线程与工作线程的桥梁。这里我们复用了之前项目中手搓的高性能 RingBuffer。感兴趣的朋友可以去我前面发的文章看一下这个RingBuffer的具体实现，这里我就不再详细描述了。

工作线程：先从队列取任务，然后执行比较耗时的逻辑（如数据库查询、加密解密），最后将结果write 回客户端。它能充分利用多核 CPU 进行并行计算。

下面这张图是我画的一张示意图，主线程负责将任务push到 RingBuffer，工作线程获取任务并处理完毕后，直接将响应结果写入客户端 Socket。

5.2 Ring Buffer 实现

我直接将前面那篇《高并发异步日志库》的 RingBuffer 进行简单的修改，如下：

#ifndef RING_BUFFER_H
#define RING_BUFFER_H

#include 
#include 

#define BUFFER_SIZE 1024

//任务结构体
typedef struct {
    int client_fd;           // 记录是谁发的请求
    char data[BUFFER_SIZE];  // 具体数据
} Task;

// 环形缓冲区管理器
typedef struct {
    Task *entries;  // 指向 malloc 的大数组
    int capacity;   // 总容量
    
    int head;
    int tail;
    int count;

    pthread_mutex_t mutex;
    pthread_cond_t cond_producer;
    pthread_cond_t cond_consumer;

    bool is_running; // 优雅退出
} RingBuffer;

void rb_init(RingBuffer *rb, int capacity);
void rb_push(RingBuffer *rb, const Task *entry);
int  rb_pop(RingBuffer *rb, Task *entry);
void rb_stop(RingBuffer *rb);
void rb_destroy(RingBuffer *rb);

#endif

ring_buffer.c文件里面的LOG也要改为TASK，但代码逻辑不需要改变。

5.3 核心代码实现

#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include "ring_buffer.h"

#define PORT 8080
#define MAX_EVENTS 10
#define THREADS_NUM 4  //四个线程处理任务

volatile sig_atomic_t  g_is_running = 1;
RingBuffer *g_rb = NULL;

void sig_handler(int sig)
{
	g_is_running = 0;

	if(g_rb)
	{
		rb_stop(g_rb);
	}
}

void *work_thread(void *arg)
{
	RingBuffer *rb = (RingBuffer *)arg;
	Task task;

	while(rb_pop(rb,&task))//优雅退出
	{
		printf("工作线程%ld拿到任务，处理fd = %d
",pthread_self(),task.client_fd);

		sleep(5);//模拟耗时的运算

        int len = strlen(task.data);
		for(int i=0;i<len;i++) 
		{
			task.data[i] = toupper(task.data[i]);
		}
		write(task.client_fd,task.data,strlen(task.data));
	}
	return NULL;
}

int main()
{
    signal(SIGPIPE, SIG_IGN);
	signal(SIGINT,sig_handler);

	RingBuffer rb;
	rb_init(&rb,10);
	g_rb = &rb;    //让全局指针指向main的rb

	pthread_t threads[THREADS_NUM];
	for(int i=0;i<THREADS_NUM;i++)
	{
		pthread_create(&threads[i],NULL,work_thread,(void *)&rb);
	}

	int listen_fd,conn_fd,epoll_fd,n_read;
	struct sockaddr_in server_addr;

	struct epoll_event ev,events[MAX_EVENTS];

	listen_fd = socket(AF_INET,SOCK_STREAM,0);
	if(listen_fd == -1)
	{
		perror("socket listen");
		exit(1);
	}
	printf("监听socket创建成功
");
    
    int opt = 1;
	setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

	memset(&server_addr,0,sizeof(server_addr));
	server_addr.sin_family = AF_INET;
	server_addr.sin_port = htons(PORT);
	server_addr.sin_addr.s_addr = htonl(INADDR_ANY);

	if(bind(listen_fd,(struct sockaddr *)&server_addr,sizeof(server_addr)) == -1)
	{
		perror("bind");
		exit(1);
	}
	printf("绑定成功
");

	if(listen(listen_fd,5) == -1)
	{
		perror("listen");
		exit(1);
	}

	epoll_fd = epoll_create1(0);
	if(epoll_fd == -1)
	{
		perror("epoll_create");
		exit(1);
	}
	printf("epoll实例创建成功
");

	ev.events = EPOLLIN;
	ev.data.fd = listen_fd;
	if(epoll_ctl(epoll_fd,EPOLL_CTL_ADD,listen_fd,&ev) == -1)
	{
		perror("epoll_ctl");
		exit(1);
	}
	printf("listen_fd添加成功
");
	printf("服务器正在监听%d端口....
",PORT);

	while(g_is_running)
	{
		n_read = epoll_wait(epoll_fd,events,MAX_EVENTS,500);//500ms超时
		if(n_read < 0)
		{
			continue;
		}

		for(int i=0;i<n_read;i++)
		{
			if(events[i].data.fd == listen_fd)//有新的连接到来
			{
				conn_fd = accept(listen_fd,(struct sockaddr *)NULL,NULL);
				printf("接受新的连接，fd = %d
",conn_fd);
				ev.events = EPOLLIN;
				ev.data.fd = conn_fd;
				epoll_ctl(epoll_fd,EPOLL_CTL_ADD,conn_fd,&ev);
			}
			else
			{
				int client_fd = events[i].data.fd;
				Task new_task;
				new_task.client_fd = client_fd;

				ssize_t n = read(new_task.client_fd,new_task.data,BUFFER_SIZE-1);

				if(n == 0)
				{
					printf("客户端fd = %d 关闭了连接
",client_fd);
					epoll_ctl(epoll_fd,EPOLL_CTL_DEL,client_fd,NULL);
					close(client_fd);
				}
				else if(n > 0)
				{
					new_task.data[n] = '';
					rb_push(&rb,&new_task);
				}
				else
				{
					perror("read");
					close(client_fd);
				}
			}
		}
	}
	printf("主循环退出，准备关闭服务器
");
	for(int i=0;i<THREADS_NUM;i++)
	{
		pthread_join(threads[i],NULL);
	}

	printf("所有线程已经全部退出
");
	close(listen_fd);
	rb_destroy(&rb);
	printf("服务器已经关闭
");

	return 0;
}

这段核心代码虽然不长，但它诠释了 Reactor 模型的核心思想。下面我要解读一下关键的部分：

主线程运行在 epoll 事件循环中，当 epoll 通知有数据时，主线程调用 read 将数据从内核读到用户态内存。但它自己不进行任何业务处理，而是立刻将 client_fd 和数据封装成一个 Task，push进 RingBuffer。

4 个工作线程从 RingBuffer 中获取任务。如果队列为空，它们会挂起，不占用 CPU，等待条件变量唤醒。这里我特意用 sleep(5) 模拟了耗时的业务逻辑。在前面的单线程模型中，这会导致服务器卡死 5 秒。但在本模型中，这只是其中一个工作线程在睡觉，主线程依然在飞快地接收新请求，其他工作线程依然在忙自己的事情。业务处理完毕后，工作线程直接调用 write 将结果发回给客户端，到这里完成了闭环。

还有，可能有人注意到了，在 read 时，我使用了 BUFFER_SIZE - 1，预留了一个字节给字符串结束符 ，防止缓冲区溢出。

如果客户端突然断开，工作线程还往里写数据，内核会直接发信号杀掉进程。因此我还在main 函数开头添加了 signal(SIGPIPE, SIG_IGN)，防止因客户端意外断开导致服务器写入时崩溃。

此外，我还实现了优雅退出，当你在服务器端按下Ctrl+C时，程序并不会立刻退出，而是去执行sig_handler信号处理函数。返回后主线程会跳出while循环，因为在函数rb_stop中已经通知工作线程结束了，因此主线程在main函数中要进行回收，回收完之后关闭。

5.4 效果验证

为了验证这个模型的效果，运行服务器端后，我还是像前面一样开了三个客户端并连接上服务器。然后我先给这三个客户端每个都分别输入一段内容，但是先不按回车，等输入完了，飞快的切换客户端并按下回车。最终会发现，大约5秒后，这三个客户端是同时收到回复的，这完全说明了这个模型的并发性。

尽管每个任务都要睡 5 秒，但我们可以看到，主线程依然能流畅地接受新的连接请求，没有丝毫卡顿。

观察服务器端的日志，处理 fd=5, fd=6, fd=7 的是不同的工作线程。这说明线程池正在全速运转，正在进行真正的并行工作。

最后直接在服务器端按下Ctrl+C，可以看到优雅退出的信息，工作线程也会随之结束。

6. 标题有六个字

最后再回看正篇文章。如果用的是 fork ，1 万个连接就是 1 万个进程，光是上下文切换就够 CPU 吃一壶得了。而用我们现在的 Reactor 模型，1 万个连接在 epoll 眼里不过是红黑树上的 1 万个节点，只有真正活跃的那几十个连接才会触发回调。

我把完整的 Reactor 模型的源码放在 github 里面了，有需要的可以自取，感觉有帮助的朋友也可以点上一手关注

👉源码点击这里