🔥 本文专栏：Linux网络
🌸作者主页：努力努力再努力wz

💪 今日博客励志语录：遗憾是过去的灰尘，而希望是未来的光；别因为忙着擦灰，而错过了推开窗户的机会。

★★★ 本文前置知识：

HTTP

---

引入

那么此前我们已经完成了网络原理相关内容的学习，其中包括使用网络编程（即通过 TCP 和 UDP 类型的套接字）进行通信，并且认识了 TCP/IP 模型 中每一层协议栈的基本内容。我们知道，一台主机能够与另一台主机进行通信，即在网络中完成数据的传输。但需要明确的是，网络通信模型并不是主机与主机之间直接通信，而是位于不同主机上的进程之间进行通信。

当位于不同主机上的进程需要接收对方通过网络发送过来的数据报时，其前提是双方都必须创建 套接字（socket）。套接字可以被看作一个通信的容器：网络中接收到的数据会被内核放入套接字的接收缓冲区，随后进程通过系统调用从套接字的接收缓冲区中读取应用层数据，并对其进行后续解析。因此，从应用程序的角度来看，通信的起点实际上就是创建套接字并通过套接字进行数据收发。

在实际的网络应用中，进程之间的通信几乎都采用 客户端–服务端（Client–Server）模型。也就是说，一个进程扮演客户端的角色，另一个进程扮演服务端的角色。客户端主动向服务端发送请求报文；服务端成功接收到请求报文之后，对请求进行解析并生成响应报文，随后再将响应发送回客户端。

对于服务端来说，其通信模式通常不可能是简单的点对点通信。所谓点对点通信，是指一个服务端只与一个客户端进行通信，而在真实场景中，服务端通常需要同时为多个客户端提供服务。因此，服务端与客户端之间通常呈现出一对多的关系。这意味着服务端在任意时刻都可能接收到来自不同客户端发送的大量请求报文。为了能够正确处理这些请求，服务端必须具备在同一时间并发处理多个请求的能力，即在接收到请求后完成请求解析、业务处理以及响应构建，并最终将响应发送回对应的客户端。因此，从系统设计的角度来看，服务端天然需要具备较高的并发处理能力。

虽然服务端是与位于不同主机上的客户端进行通信，但从服务端自身的视角来看，其处理请求的过程本质上仍然是一个 I/O 过程：
服务端进程通过系统调用从套接字的接收缓冲区读取应用层数据，对数据进行处理，然后再通过系统调用将生成的响应报文写入到套接字的发送缓冲区。

换言之，服务端进程调用系统调用接口从套接字接收缓冲区读取应用层数据，以及调用系统调用将应用层数据写入发送缓冲区，这两个过程本质上都属于 I/O 操作。

当然，从宏观角度来看，服务端确实是在与远端客户端进行网络通信。例如，客户端发送的 IP 数据包会经过网络中的路由器节点逐跳转发，最终到达目标主机。随后，目标主机的网卡会将接收到的物理信号转换为数字信号（即二进制数据流），并将其存储在网卡缓冲区中。接下来网卡会对数据帧进行基本校验（例如帧头与帧尾的校验）。若校验通过，网卡通过 DMA 将数据从网卡缓冲区拷贝到主机内存中，然后触发 硬件中断。CPU 在接收到中断后会切换到内核态，对数据帧进行处理：首先由数据链路层接收并解析帧结构，然后逐层向上交付到网络层协议栈，再交付到传输层协议栈，最终由内核将数据放入对应套接字的接收缓冲区，从而被应用层进程读取。

然而，从服务端进程的视角来看，它并不会关心底层网络中所经历的这些复杂过程。例如，当服务端接收到一个 HTTP 请求时，进程并不会感知数据在网络中的转发路径，也不会感知内核协议栈对数据帧的逐层处理以及向上交付的具体细节。

因此，从服务端进程的角度来看，其核心只涉及两类工作：

1. I/O 操作 —— 从套接字读取请求数据，并将响应数据写回套接字。
2. 业务逻辑处理 —— 对请求数据进行解析，并生成对应的响应结果。

也正因为如此，当我们希望提升服务端的整体效率或突破性能瓶颈时，关键点往往在于提高 I/O 处理的效率。在高并发服务器程序中，I/O 处理能力通常决定了系统性能的上限。因此，要想理解如何提升服务端性能，首先就必须理解 I/O 模型（I/O Model） 的工作机制。

I/O模型

那么在认识具体的 IO 模型之前，根据上文我们已经知道，对于服务端而言，其所执行的工作大致可以分为两个部分：IO 操作以及业务逻辑处理。而对于 IO 操作来说，又可以进一步划分为两个阶段，分别为等待阶段以及拷贝阶段。

从抽象角度来看，IO 操作本质上涉及两个方向的数据流动，分别对应 I（Input） 和 O（Output）：

• 数据从应用层流向内核
• 数据从内核流向应用层

---

那么首先我们来看数据从应用层流向内核的情况。

这一过程通常对应写操作，例如调用 write 接口或者 send 接口。该操作首先会经历等待阶段。例如，当内核缓冲区已经满时，此时写操作就会进入阻塞状态，直到缓冲区中有可用空间为止，而这里需要注意的是，对于写操作的等待阶段，我们通常会简单地认为：当发送缓冲区已满时，进程就会陷入阻塞等待。但如果进一步细化其内部机制，可以发现实际情况要更加复杂。

如果通信双方是通过 TCP 连接进行数据传输，那么当进程调用 send 或 write 接口时，首先会将数据从用户态缓冲区拷贝到内核态的套接字发送缓冲区。然而，这并不意味着这些数据会立刻从当前主机发送到网络中。

这是因为 TCP 协议本身具有一套发送控制策略，例如 滑动窗口机制以及 Nagle 算法等。这些机制会影响数据何时真正被封装为 TCP 报文段并发送到网络中。例如，当对端的接收窗口较小，或者 TCP 仍在等待合适的发送时机时，数据可能会暂时停留在发送缓冲区中，而不会立即发出。

因此，从更具体的角度来看，写操作进入等待阶段的本质原因是：套接字发送缓冲区已满，而内核暂时无法继续向网络发送数据。在这种情况下，进程需要等待内核在后续发送过程中腾出缓冲区空间（例如数据被成功发送并得到确认，从而释放缓冲区），当发送缓冲区重新具备可用空间后，写操作才能继续进行。

换言之，这里的等待阶段本质上是等待发送条件满足，从而释放发送缓冲区空间，而不是简单意义上的“等待数据发送完成”。；而如果内核缓冲区并未满，则写操作不会进入阻塞状态。

当等待阶段结束之后，接下来便进入拷贝阶段。所谓拷贝阶段，就是将数据从用户态缓冲区复制到内核缓冲区中，例如 TCP/UDP 套接字的发送缓冲区，或者文件对应的内核缓冲区。

---

而另一个数据流动方向，即从内核到应用层，对应的是读操作。

该过程同样会首先经历等待阶段。在这个阶段中，系统需要等待外部 IO 设备（例如磁盘或网卡）将数据传输到内存，并写入到对应的内核缓冲区中。以网络通信为例，这实际上是等待接收缓冲区中出现可读取的数据。

当等待阶段结束之后，接下来进入拷贝阶段，即将数据从内核缓冲区复制到用户缓冲区，从而完成一次读操作。

---

因此，无论是从应用层流向内核的写操作，还是从内核流向应用层的读操作，整个 IO 过程通常都包含两个阶段：

• 等待阶段 （Waiting Phase）
• 拷贝阶段 （Copy Phase）

其中，对于等待阶段来说，当相关条件尚未就绪时（例如接收缓冲区为空或发送缓冲区已满），进程往往会进入阻塞状态。此时进程会释放 CPU，被移出就绪队列，并被放入等待队列中。在这种状态下，进程无法继续推进当前任务的执行，也无法处理其他业务逻辑。

根据上文的讨论，对于一个进程而言，其性能瓶颈往往在于 IO 操作的效率。而提高 IO 操作效率的关键之一，就是减少等待阶段在整个 IO 流程中的占比。换言之，需要尽可能缩短等待时间，因为在等待阶段中，进程处于阻塞状态，无法执行任何有效工作。

---

单线程阻塞式IO

基于上述背景，首先可以引入第一种 IO 模型，即单线程阻塞式 IO 模型。

在这种模型下，系统中只有一个主线程，而主线程既负责 IO 操作，也负责 业务逻辑处理。这意味着 IO 操作与业务逻辑处理之间是串行执行的。

一旦某个 IO 操作因为条件未就绪而发生阻塞，那么整个进程都会被挂起，从而导致后续业务逻辑无法继续执行。因此，在这种模型下，系统的整体处理效率通常是较低的。

单线程/多线程非阻塞式IO

由此便自然引出了第二种 IO 模型，即单线程非阻塞 IO 模型。

在该模型中，相比于第一种模型，其主要变化在于采用了非阻塞 IO。当进程发现 IO 条件尚未就绪时（例如接收缓冲区为空或发送缓冲区已满），系统调用不会使进程进入阻塞状态，而是立即返回。随后进程可以继续执行后续的业务逻辑。

当业务逻辑执行完毕之后，程序会再次回到开头，重新检查 IO 条件是否已经就绪。如果仍然未就绪，则重复这一过程，也就是通过**主动轮询（Polling）**的方式不断检查条件状态。

---

相比于第一种模型，该模型的优点在于：进程不会被动地陷入阻塞等待，而是可以继续执行其他逻辑。

不过需要注意的是，等待本身在内核层面依然是存在的。例如在网络通信场景中，内核仍然需要等待远端主机发送 IP 数据包，该数据包在网络中经过多个路由器逐跳转发后到达目标主机。随后网卡将接收到的物理信号转换为数字信号，并完成帧头与帧尾校验；校验通过后提取五元组信息，通过哈希计算查找对应的队列，并将数据放入相应的环形缓冲区。随后触发硬件中断，使 CPU 切换到内核态，并经过数据链路层、网络层以及传输层等协议栈的处理，最终才可能被应用程序读取。

因此，等待过程并没有消失，只是应用进程不再与内核一起被动等待条件就绪。当进程发现条件尚未就绪时，它不会原地阻塞，而是继续执行后续逻辑。

---

那么，这是否意味着该模型没有缺点呢？答案显然是否定的。该模型同样存在明显的问题。

首先，在该模型下，进程需要主动轮询，即不断重复调用系统调用来检查 IO 状态。虽然进程不会被移出就绪队列并放入等待队列，但需要注意的是，每一次系统调用都会涉及用户态与内核态之间的模式切换。

具体来说，当程序从用户态进入内核态时，系统首先需要保存当前进程的上下文，例如各个寄存器的值（包括栈指针、程序计数器等），并将其压入该进程对应的内核栈中；随后 CPU 才会切换到内核态执行内核代码。而当系统调用结束返回用户态时，还需要从内核栈中恢复之前保存的寄存器状态，从而恢复进程上下文。

因此，主动轮询会导致频繁的模式切换开销。

此外，在服务端场景下，一个服务器往往需要同时与大量客户端进行通信。在单线程非阻塞模型下，需要由一个线程以非阻塞方式轮询 多个文件描述符（fd），不断检查它们的状态是否就绪。

在这种情况下：

• 用户态与内核态之间的切换会更加频繁
• 进程还需要花费大量时间进行轮询操作

换言之，需要不断遍历所有 fd 来检查其状态是否就绪，而这种遍历的时间复杂度通常为 O(N)。并且在这种 O(N) 的遍历过程中，绝大多数检查实际上都是无效的，因为在大多数时间里，只有极少数 fd 处于就绪状态，因此在遍历过程中，大部分检查都是无效的。这种以 O(N) 复杂度进行的轮询会导致 CPU 大量时间消耗在无意义的状态检测上，从而造成 CPU 空转，降低系统整体的处理效率。

---

那么有的读者读到这里，可能会想到：此前讨论的大多是单线程模型。在这种模型下，一个线程串行地完成 I/O 操作以及业务逻辑处理。而在服务端场景中，这种单线程模型的缺点会被进一步放大，因为服务端通常是 一对多 的通信模式。也就是说，一个服务端需要同时与多个客户端进行通信。

这意味着一个线程需要不断检查 多个文件描述符的读条件是否就绪。虽然通过主动轮询的方式可以避免进程陷入被动的阻塞等待，但主动轮询本身也是有代价的。特别是在连接数较多的情况下，会带来 大量系统调用产生的模式切换开销，以及轮询遍历文件描述符集合的开销。

因此读者可能会想到另一种方案：多线程 + 非阻塞调用。在这种模型下，主线程不再直接处理 I/O 操作，而是作为 I/O 操作的发起者和调度者；主线程创建出多个工作线程，由这些工作线程负责执行具体的 I/O 操作以及业务逻辑处理。

假设当前有 1000 个连接，那么主线程可以创建 1000 个线程，每个线程分别负责一个连接，1000个线程并发检查对应文件描述符的读条件是否就绪，并在就绪后执行相应的业务逻辑。

在这种 多线程非阻塞调用模型 下，相比于 单线程非阻塞轮询模型，其优点在于减少了轮询遍历的代价。因为每个线程只需要关注 一个文件描述符 是否就绪，而不需要遍历整个文件描述符集合。

但是这种方案的缺陷同样非常明显。

首先需要明确的是：线程的创建与销毁本身是有成本的。在 Linux 系统中创建线程通常通过 pthread_create 接口完成，而 pthread_create 在底层会调用 clone 系统调用来创建线程。

从本质上来说，一个线程对应的是 一个独立执行的用户态函数调用上下文。既然是函数执行，就需要为其分配 栈空间（线程栈） 用于保存函数调用帧以及局部变量。每一个线程都拥有自己独立的执行上下文。

虽然线程之间的资源是高度共享的，例如 文件描述符表、页表、全局数据区等资源 都是共享的，但由于线程的执行是 可独立调度和切换的，因此每一个线程都必须拥有自己独立的 线程栈，用于存储其函数调用过程中的局部变量和返回地址等信息。

线程栈显然不可能在进程的主线程栈上进行分配，因为如果多个线程的栈空间发生重叠，一旦线程发生上下文切换，就会导致严重的问题。例如某个线程时间片耗尽后，CPU 会进行上下文切换：当前线程的寄存器上下文会被保存到对应的 内核栈 中，然后切换到下一个线程，并从该线程的内核栈中恢复寄存器状态，其中就包括 栈指针寄存器（SP）。

如果多个线程的栈空间存在重叠，那么在恢复栈指针之后，线程的栈操作可能会覆盖其他线程的数据，从而导致 内存破坏以及数据不一致的问题。因此线程栈必须拥有 独立且不重叠的内存空间。

在 Linux 中，线程栈通常是在 用户态堆区分配的一块独立虚拟地址空间。因此每创建一个线程，都需要额外分配一块用于线程栈的内存空间。

这意味着线程的创建会带来一定的 内存开销。此外，线程常被称为“轻量级进程（Lightweight Process）”，是因为Linux 内核在管理线程时，依然是通过 task_struct 结构体来描述和管理调度实体的。因此创建线程不仅需要分配线程栈，还需要分配对应的 task_struct 以及内核栈等内核资源。

在 Linux 系统中，默认的线程栈大小通常为 8MB。如果创建 1000 个线程，那么仅线程栈就可能占用 约 8GB 的虚拟地址空间。因此线程数量过多时，会带来显著的 内存资源消耗。

其次，多线程还会带来 上下文切换（Context Switch） 的开销。

我们知道，在同一个进程中的多个线程之间，大部分资源是共享的，例如 页表、文件描述符表以及代码段和数据段 等。因此线程之间的切换相比进程切换来说成本更低，因为不需要切换整个地址空间。

但即便如此，线程切换依然存在一定的成本。

CPU 在执行线程代码时，其访问内存使用的是 虚拟地址。而实际访问内存需要 物理地址，因此每一次内存访问都需要经过 虚拟地址到物理地址的转换。这个转换过程由 MMU（Memory Management Unit，内存管理单元） 完成。

当 CPU 进行地址转换时，MMU 会首先查询 TLB（Translation Lookaside Buffer）。TLB本质上是一个高速缓存，用于缓存最近使用过的 页表项映射关系（虚拟地址 → 物理地址）。

如果 TLB 命中，那么可以直接得到物理地址；如果 TLB 未命中，则需要访问内存中的页表结构进行页表遍历（Page Walk）。页表通常是 多级页表结构，例如 x86 架构下的多级页目录结构。

由于内存访问速度远慢于 CPU 缓存，因此 TLB 的存在可以显著减少页表访问开销。

在 线程切换 时，如果仍然是在同一进程内进行线程切换，则通常 不会更换页表基地址寄存器（CR3），因此不会强制刷新整个 TLB。同进程线程切换的主要开销不是 TLB 失效，而是 CPU 缓存中的热数据失效——切换前后两个线程访问的是完全不同的内存区域，缓存里之前预热好的内容换了个线程就全部失效了。

由于局部性原理的存在，CPU 在访问内存时通常不会只加载当前所需的那一个数据。例如以 32 位机器 为例，若数据总线宽度为 32 位，则一次总线传输理论上可以读取 4 字节的数据。但在实际的缓存体系中，CPU 在进行一次内存访问时，并不会只读取这 4 字节，而是会按照 Cache Line（64字节缓存行） 的粒度，将目标地址附近的一整块连续数据加载到缓存中，也就是说，当发生缓存未命中时，CPU 会发起一次内存访问请求，随后内存子系统会占用内存总线，通过多个总线传输周期，将整个 64 字节的 Cache Line 搬运到 CPU 缓存中。例如在 32 位数据总线的情况下，需要经过约 16 次总线传输才能完成这一过程。。

这样做的原因在于 空间局部性：程序在访问某个内存地址之后，很有可能在接下来访问其附近的地址。因此，CPU 会提前将目标数据周围的一段数据一并加载到缓存中，以提高后续访问时的 缓存命中率（Cache Hit Rate），从而减少对主存的访问次数。

因此，这里的线程上下文切换开销通常并不是由于 TLB 被刷新所导致的，而是由于线程切换后，原先加载到 CPU Cache中的数据很可能不再被当前运行的线程使用，不同线程的工作数据集往往不同，从而使得之前加载到 CPU Cache 中的数据逐渐失效并被新的数据覆盖，从而导致缓存局部性被破坏（Cache Locality Loss），进而增加缓存未命中的概率。

除了上述问题之外，多线程还会增加 CPU 调度的开销。

在早期的 Linux 内核实现中，就绪队列（Runqueue）是通过 优先级数组（Priority Array） 实现的。该数组中的每一个元素都是一个指针，指向一个链表，而数组下标对应不同的优先级。

链表中的每一个节点就是一个 task_struct，同一个链表中的进程具有相同的优先级。

Linux 中的优先级分为 静态优先级（Static Priority） 和 动态优先级（Dynamic Priority）。静态优先级通常反映程序员的意图，例如通过调整 nice 值 来影响进程的调度优先级。

然而内核调度器需要从全局角度管理所有可运行进程，因此仅依赖静态优先级是不够的。调度器还会根据 进程的运行时间等因素 动态调整其优先级，从而得到最终的调度优先级。

为了加快优先级数组的查找速度，调度器还会维护一个 位图（bitmap），用于快速判断哪些优先级队列是非空的，从而可以在 O(1) 时间内找到最高优先级的可运行任务。

当时的就绪队列实际上包含两个队列：

• active 队列
• expired 队列

当进程时间片耗尽时，会重新计算其时间片和动态优先级，然后将其放入 expired 队列。当 active 队列中的任务全部执行完毕后，调度器只需要交换两个队列的指针即可。

因此这种调度算法的时间复杂度为 O(1)。

进程从 active 队列被调度出来运行
  → 时间片耗尽
  → 重新计算新的时间片以及动态优先级
  → 移入 expired 队列

active 队列彻底空了
  → 直接交换 active 和 expired 两个指针
  → 原来的 expired 队列变成新的 active 队列
  → 重新开始

而在现代 Linux 内核中，就绪队列采用的是 红黑树 结构进行组织。在这种实现中，红黑树的键值并不是传统意义上的优先级，而是 虚拟运行时间（vruntime）。

虚拟运行时间的计算主要取决于两个因素：进程的静态优先级（nice 值）以及进程实际的运行时间。如果一个进程实际运行的时间越久，那么其虚拟运行时间就会越大；而如果进程的 nice 值越大（优先级越低），那么其虚拟运行时间增长的幅度也会越大。

因此，每当一个进程完成一次调度周期并被换出 CPU 时，内核都会根据该进程本次的 实际运行时间 重新计算并更新其虚拟运行时间。

具体来说，虚拟运行时间的增长是通过如下方式计算的：将 实际运行时间 乘以一个 权重比例系数。该比例系数由 nice 值为 0 时对应的基准权重（NICE_0_LOAD） 与当前进程的权重之比得到。也就是说：

• nice 值越大（优先级越低），对应的权重越小，因此虚拟运行时间增长得越快；
• nice 值越小（优先级越高），对应的权重越大，因此虚拟运行时间增长得越慢。

因此，在相同的实际运行时间下，低优先级进程的 vruntime 会增长得更快，而高优先级进程增长得更慢。

在计算出新的 虚拟运行时间 之后，调度器会将该调度实体重新插入到 红黑树 中。由于红黑树是一种自平衡二叉搜索树，因此插入操作的时间复杂度为 O(log N)。

为了保证调度的公平性，需要推动各个进程的执行进度，因此需要优先照顾整体运行时间最少的进程，从而推进其任务的执行。因此调度器会选择 虚拟运行时间最小的调度实体，也就是红黑树中的 最左节点 进行调度。

vruntime += 实际运行时间 × (NICE_0_LOAD / 该进程的权重)

/* 其中 NICE_0_LOAD 是 nice 值为 0 时对应的基准权重，
   在 Linux 中该值为 1024。

   Linux 内核维护了一张静态权重表：
   nice 值每降低 1，权重大约增加 25%
   nice 值每升高 1，权重大约减少 20%

   nice 值低（优先级高）的进程：
   权重大 → vruntime 增长慢 → 更容易再次被调度

   nice 值高（优先级低）的进程：
   权重小 → vruntime 增长快 → 更快移动到红黑树右侧
   从而让出 CPU
*/

---

那么以上，我们介绍了几种 IO 模型，分别是单线程阻塞式 IO 模型以及单线程 / 多线程的非阻塞主动轮询式 IO 模型。通过前文的分析可以明显感受到，这类模型存在一些较为明显的缺点。

首先，在阻塞式 IO 模型中，线程需要在条件未就绪时被动挂起，等待条件满足后才会被唤醒。其次，在许多实现方式中往往采用一个线程处理一个连接的模式，而在典型的服务端场景下，服务端通常需要同时面对大量客户端，即存在一对多的连接关系。这意味着服务端需要创建大量线程，每一个线程对应一个连接。

然而，大量线程的存在会带来一系列额外开销。例如：

• 线程之间频繁的上下文切换开销即CPU 缓存失效所带来的性能损耗；
• 用户态与内核态之间的频繁模式切换；
• 每个线程自身所占用的栈空间等内存资源；
• 线程数量增多后，CPU 调度器的调度压力显著增加。

I/O多路复用

因此，接下来便过渡到多路复用 IO 模型（I/O Multiplexing）。该模型的提出，正是为了缓解甚至解决上述模型所带来的问题。实际上，前述模型的核心缺陷在于：创建了过多线程。而其根本原因则在于一个线程仅负责管理一个连接。

如果希望解决这一问题，本质思路便是减少线程数量。而减少线程数量的关键方法，就是让一个线程同时管理多个连接，而不是仅管理一个连接。这一思想，正是多路复用 IO 模型的核心设计理念。

在具体介绍实现多路复用的系统调用接口（例如 select、poll、epoll 等）之前，我们首先从思想层面理解多路复用的基本机制，并分析其相比之前模型的优势所在。

所谓多路复用，其核心思想是：由一个线程同时检查多个文件描述符（file descriptor，fd）的就绪状态。具体来说，线程会调用相应的系统调用，将一组文件描述符交给内核进行监视；当这些文件描述符中任意一个或多个条件就绪时，内核再唤醒阻塞的线程。

从表面上看，这种机制似乎与最初介绍的阻塞式 IO 模型类似，因为线程同样会进入阻塞状态等待条件就绪。但两者之间存在一个关键区别：
在阻塞式 IO 模型中，线程只等待一个 fd 的就绪事件；而在多路复用模型中，线程同时等待一组 fd 的就绪事件。

这意味着线程并不容易长时间陷入无意义的阻塞状态。因为在一组 fd 中，只要任意一个 fd 条件就绪，线程便会被内核唤醒。由于同时监视多个 fd，这也显著提高了条件就绪事件发生的概率，从而使线程能够更加频繁地被有效唤醒。

并且，当线程被唤醒时，往往不仅仅是一个 fd 就绪，而是可能有多个 fd 同时就绪。这一过程可以类比为在河边钓鱼：
在传统模型中，相当于一个人在河边只放置一根鱼竿，然后等待鱼儿上钩；而在多路复用模型中，则相当于在河边同时放置了多根鱼竿。虽然每一条鱼咬钩的概率并没有改变，但鱼竿数量越多，被鱼咬钩的概率就越大，因此一次能够收获的鱼也就越多。类似地，一个线程一次便可能检测到多个 fd 处于就绪状态。

当检测到多个文件描述符就绪之后，每一个文件描述符通常都对应着特定的业务处理逻辑。因此，接下来便可以根据具体情况，将这些就绪的 fd 分配给相应的线程进行处理。

相比之前的多线程非阻塞 IO 模型，那种模型往往是由主线程为每一个连接创建或分配一个线程，每个线程只负责管理一个 fd，并通过非阻塞方式不断轮询该 fd 的状态。这种方式存在一个明显问题：CPU 会出现大量空转。

原因在于，对于单个线程而言，其所监视的 fd 在大多数轮询周期内往往并不会立即就绪，因此线程会不断重复检查，却很少获得有效结果。对于典型的客户端—服务端模型来说，这里通常关注的是 TCP 套接字是否就绪，例如：

• 接收缓冲区是否存在可读数据；
• 发送缓冲区是否存在可写空间。

而服务端通常需要先接收到客户端发送的请求报文，才能继续执行后续业务逻辑比如解析请求以及构建响应等。如果线程只负责一个 fd，而该 fd 长时间未就绪，那么线程既无法执行有效工作，又持续占用 CPU 进行 轮询检查，从而导致CPU 出现大量无效空转，造成资源浪费。

因此，多路复用模型的优势可以概括为以下几点：

首先，通过一个线程管理多个连接，显著减少了线程数量，从而降低了线程上下文切换、内存占用以及调度开销。

其次，由于线程一次监视的是一组 fd 而不是单个 fd，因此线程在阻塞等待时，更容易等到某个 fd 就绪，从而减少无意义的等待时间。

最后，当线程被唤醒时，往往能够同时检测到多个 fd 就绪事件，随后再将这些就绪的 fd 分发给相应的工作线程进行处理。这样既避免了线程管理单个连接所带来的无效轮询问题，也显著减少了 CPU 空转所带来的资源浪费。

---

select

那么认识到这一点之后，接下来我们就可以进一步介绍与I/O 多路复用相关的系统调用接口。首先要介绍的是 select 系统调用接口。根据上文的分析，多路复用的核心思想在于：进程本身仍然会陷入阻塞等待，但阻塞的对象不再是单个文件描述符，而是一组文件描述符，当其中任意一个文件描述符的条件就绪时，进程就会被唤醒。

由于一个线程需要同时等待一组文件描述符，因此在调用 select 接口时，需要向内核传递一个待检测的文件描述符集合。在 select 的实现中，这个集合是通过**位图（bitmap）**结构来表示的。

• 系统调用：select
• 头文件：<sys/select.h>
• 函数声明：int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);
• 返回值：返回值大于 0 表示就绪的文件描述符数量；返回值等于 0 表示在指定超时时间内没有任何文件描述符就绪；返回值小于 0 表示发生错误，并会设置 errno。

这里的 fd_set 类型本质上是一个结构体，其内部维护了一个位图数组（通常是 long 类型的数组）。位图中的每一个比特位都对应一个特定的文件描述符。例如：

• 数组第一个元素的第一个比特位对应 文件描述符 0；
• 第一个元素的第三个比特位对应 文件描述符 2。

#define __FD_SETSIZE    1024  

typedef struct {
    unsigned long fds_bits[__FD_SETSIZE / (8 * sizeof(unsigned long))];
} fd_set;

当某个比特位的值为 1 时，表示需要检测该比特位对应的文件描述符是否满足指定条件；当该比特位为 0 时，则表示不需要检测对应的文件描述符。

由于向内核传递的是一个位图形式的文件描述符集合，因此该位图中并不一定所有比特位都为 1。实际上，位图中既包含需要检测的文件描述符，也包含不需要检测的位置。对于内核来说，它必须遍历这个位图，找出值为 1 的比特位，并检查对应的文件描述符状态。因此，这个遍历过程的时间复杂度为 O(N)。

而在遍历过程中，不可避免地会出现无效遍历——即扫描到那些并未被使用的比特位。为了减少这种无效遍历，select 的第一个参数 nfds 就发挥了作用。

nfds 表示**需要检测的最大文件描述符值加一 **。换句话说，它告诉内核：位图只需要遍历到该位置即可，后续部分无需扫描。这样可以在一定程度上减少无效遍历，提高检测效率。

接下来需要讨论的是 select 的第二个和第三个参数，即 readfds 和 writefds。在上文中我们多次提到：线程需要检测文件描述符对应的条件是否就绪。然而这里的“条件”并不是单一的，它取决于文件描述符所关联的具体对象类型。

我们知道，一个文件描述符在内核中的本质表示是一个 file 结构体指针。而 Linux 的设计哲学是 “一切皆文件（Everything is a file）”，因此各种资源都会通过文件对象进行抽象。例如：

• 磁盘文件：磁盘中的普通文件在被进程打开后，会在内核中创建对应的 file 结构体；
• 内存级文件：例如匿名管道（pipe），虽然本质上是内存缓冲区，但同样通过 file 结构体进行描述；
• 网络套接字（socket）：网络通信对象同样会映射为一个 file 结构体。

因此，file 结构体既可以表示磁盘级文件，也可以表示内存级文件，还可以表示网络套接字。而不同类型对象的就绪条件自然也不同。

在前文中，我们已经按照数据流向将 I/O 操作分为两大类：

• 读操作：数据从内核空间流向用户空间
• 写操作：数据从用户空间流向内核空间

接下来以读条件为例进行说明。

1. 磁盘文件

对于磁盘文件而言，文件由属性和内容两部分构成。file 结构体内部维护了一个读写偏移量 f_pos（注意：读和写共用同一个偏移量）。同时，file 结构体还会关联一个 inode 结构体，用于记录文件的元数据，其中包括文件内容大小 i_size。

因此，对于磁盘文件来说，读条件是否就绪取决于：

• 当前读偏移量 f_pos 是否到达文件末尾 i_size。

如果 f_pos < i_size，说明文件中仍然存在未读取的数据，此时读条件是就绪的；如果 f_pos == i_size，则说明已经读取到文件末尾，读条件不再就绪。

2. 管道（Pipe）

对于内存级文件，例如匿名管道，其内部实现本质上是一个环形缓冲区。管道分为读端和写端：

• 读端对应一个 file 结构体
• 写端对应另一个 file 结构体

但这两个 file 结构体都会关联到同一个管道对象。

在内核中，file 结构体会通过指针关联一个 pipe_inode_info 结构体，其中维护两个关键指针：

• head：读指针
• tail：写指针

读条件是否就绪取决于这两个指针是否重合：

• 如果 head == tail，说明缓冲区为空，读条件不就绪；
• 如果 head != tail，说明缓冲区中仍然存在数据，读条件就绪。

3. 套接字（Socket）

对于套接字而言，file 结构体中的 private_data 字段会指向 socket 结构体，而 socket 结构体内部又会关联一个核心的 sock 结构体，用于维护：

• 网络层和传输层的状态信息
• 接收缓冲区与发送缓冲区

因此，对于套接字来说：

• 读条件就绪：接收缓冲区不为空
• 写条件就绪：发送缓冲区未满

写条件的判断同样依赖于对象类型：

• 磁盘文件：只要可以继续写入（例如允许扩展文件），写条件通常认为是就绪的；
• 管道：当 (tail + 1) % size != head 时，说明缓冲区未满，可以继续写入；
• 套接字：当发送缓冲区未满时，写条件就绪。

因此，select 的第二个和第三个参数实际上对应两类文件描述符集合：

• readfds：需要检测读条件的文件描述符集合
• writefds：需要检测写条件的文件描述符集合

如果 readfds 不为 NULL 而 writefds 为 NULL，则表示只检测这些文件描述符的读条件；反之亦然。如果两者都不为 NULL，则两种条件构成一种逻辑或关系：只要某个文件描述符满足读或写条件中的任意一个，进程就会被唤醒。

第四个参数 exceptfds 表示异常条件集合，例如带外数据等情况。但在实际开发中使用较少，通常会将其设置为 NULL。

读者看到这里，可能会产生一个疑问：

既然不同对象的读条件判断逻辑不同，那么在调用 select 时，我们仅仅传递了一个 读集合，并没有告诉内核该文件描述符具体属于哪种对象。那么内核在检测这些文件描述符时，是否还需要通过大量 if-else 来判断对象类型？

实际上并不需要。因为 file 结构体内部维护了一个文件操作函数表 f_op（即 file_operations 结构体），它本质上是一组函数指针。其中就包含一个 poll 方法。

在创建 file 结构体时，内核会根据该文件对象的具体类型（磁盘文件、管道、socket 等），为 f_op 填充不同的函数实现。因此，在 select 执行检测时，内核只需要统一调用 file->f_op->poll()，而具体的就绪条件判断逻辑则由各个对象类型对应的 poll 实现完成。

这种设计本质上体现了一种面向对象中的多态思想：

• 调用接口是统一的
• 实际执行逻辑由对象类型决定

因此，内核无需显式判断文件类型，只需调用统一接口即可完成不同对象的就绪状态检测。

而这里需要额外补充一点：对于套接字而言，我们之前提到读条件就绪通常意味着接收缓冲区不为空。然而对于 TCP 套接字 来说，需要进一步区分 监听套接字（listening socket） 与 已连接套接字（connected socket）。

虽然这两类套接字在内核中的结构体系是相同的，均属于同一条继承链：最顶层是包含发送缓冲区和接收缓冲区的 sock 结构体，然后在其基础上扩展为 inet_sock 结构体，随后是 inet_connection_sock 结构体，最终具体实现为 tcp_sock。也就是说，从数据结构角度来看，它们共享同一套 TCP 套接字对象模型。

但是，监听套接字与已连接套接字在职责上是完全不同的，因此它们在实际运行时访问 sock 体系结构中的不同部分：

• 已连接套接字：主要用于进行实际的数据收发，因此会访问 sock 结构体中的发送缓冲区和接收缓冲区。
• 监听套接字：其职责仅仅是接收新的 TCP 连接请求，并不会参与具体的数据收发。因此它并不会使用 sock 中的发送缓冲区和接收缓冲区，而是主要访问 inet_connection_sock 中维护的 半连接队列（SYN queue） 和 全连接队列（accept queue）。

因此，这里需要特别注意：监听套接字与已连接套接字的读就绪条件是不同的。

对于 已连接套接字 而言，读条件就绪通常意味着接收缓冲区中存在可读数据。

而对于 监听套接字 而言，其读条件并不是接收缓冲区不为空，因为监听套接字根本不会进行数据接收操作。监听套接字的职责仅仅是接收新连接，因此它的读条件就绪实际上是：

全连接队列（accept queue）不为空。

一旦有新的 TCP 连接完成三次握手并进入全连接队列，内核就会判定该监听套接字读就绪。

---

所以这里对于服务端来说，其初始状态通常是：创建一个监听套接字，然后主线程将该监听套接字加入 select 需要检测的读集合中。

当主线程因为 select 被唤醒，并检测到监听套接字处于读就绪状态时，说明当前全连接队列中存在已经完成三次握手的连接。此时服务端会调用 accept，从全连接队列中取出一个连接，并获得对应的已连接套接字文件描述符。随后再将该文件描述符加入到读集合中，用于检测该连接上是否存在可读数据。

---

而 select 的最后一个参数是一个 timeval 结构体，用于指定超时时间。该结构体包含两个字段，分别表示秒和微秒：

/* timeval 结构定义 */
struct timeval {
    long tv_sec;   /* 秒 */
    long tv_usec;  /* 微秒 */
};

timeval 结构体的作用是为 select 指定一个最大等待时间。当 select 被调用后，内核会为其设置一个定时器。如果在该时间内没有任何文件描述符就绪，一旦定时器到期，内核也会唤醒因调用 select 而阻塞的进程。

从行为上来看，这相当于让 select 具备超时返回机制，从而避免无限期阻塞。

不过在很多典型的服务器设计中，select 通常由主线程调用，而主线程的主要职责往往只是负责监听新连接并分发任务。因此即使因为超时而被唤醒，主线程往往也没有其他实际工作需要执行，所以这种超时机制在很多场景下意义并不特别大。相比之下，后文将要介绍的 poll 在某些设计模式下会更有帮助。

---

此外，对于 select 的第二、第三、第四个参数，也就是读集合、写集合以及异常集合，以及最后的 timeout 参数，都属于输入输出型参数（in/out parameter）。

在调用 select 时：

• 进程首先将需要检测的文件描述符加入到读集合、写集合或异常集合中，此时这些集合属于输入参数。
• 内核随后会遍历这些集合对应的位图（bitmap），检查哪些比特位为 1，并对对应的文件描述符逐个检测其状态是否满足就绪条件。

当遍历检测结束后，内核还需要把检测结果返回给用户态。为此，内核会直接修改进程传入的位图结构，将其覆盖为新的结果集合：

• 若某个比特位仍为 1，表示该文件描述符已经就绪；
• 若被清零，则表示该文件描述符当前未就绪。

因此，在 select 返回之后，应用程序仍然需要再次遍历这些位图，查找哪些位置为 1，从而确定究竟是哪些文件描述符已经就绪。

---

timeout 同样也是一个输入输出型参数。当 select 返回时，内核会将 timeout 修改为剩余尚未消耗的时间。因此在循环中反复调用 select 时，必须在每次调用之前重新设置 timeout。

这一点与 每次调用 select 前需要重新构造文件描述符集合 的原因是完全一致的——因为这些参数在返回时都已经被内核修改过。

那么在认识了 select 系统调用之后，在调用 select 接口之前，首先需要由用户态准备一个读集合或者写集合所对应的位图数组。该位图数组用于标识当前需要被内核检测状态的文件描述符集合。

位图数组需要进行初始化。对此，Linux 内核已经为我们提供了相应的辅助接口，因此无需自行实现相关逻辑。通常我们会创建一个 fd_set 类型的栈对象，而由于栈内存中的内容在初始化时可能是随机值，因此必须首先将该对象中的所有比特位初始化为 0。也就是说，需要将位图数组清空，此时就需要调用 FD_ZERO 接口。

void FD_ZERO(fd_set *set);

// 使用示例
fd_set readfds;
FD_ZERO(&readfds);  // 将所有位清零

在完成清零操作之后，接下来就需要对位图数组进行设置。由于位图数组中的每一个比特位都对应一个特定的文件描述符，因此我们需要将需要检测的文件描述符在位图数组中的对应比特位设置为 1。此时就需要使用 FD_SET 函数来完成该操作。

该函数接收两个参数：一个是文件描述符 fd，另一个是位图数组 fd_set。其底层实现本质上是通过位运算，根据传入的文件描述符编号计算对应的比特位位置，并将该比特位置为 1。因此这里的第二个参数（位图数组）属于输出型参数。

void FD_SET(int fd, fd_set *set);

// 使用示例
FD_SET(sockfd, &readfds);   // 将sockfd对应的位设为1

而上述两个接口，即 FD_ZERO 和 FD_SET，主要是在调用 select 之前使用，用于将位图数组作为输入参数传递给内核，从而告知内核需要检测哪些文件描述符的事件状态。

需要注意的是，在 select 接口的参数中，除了第一个参数 nfds 之外，其余参数均属于输入输出型参数。也就是说，当进程被唤醒之后，内核会直接修改用户传入的位图数组，仅保留那些已经就绪的文件描述符。因此，在 select 返回之后，用户态需要重新遍历之前传入的位图数组，以确定哪些文件描述符已经就绪。

为此，Linux 提供了 FD_ISSET 接口用于检测某个文件描述符是否仍然在集合中。该函数同样接收两个参数：文件描述符 fd 以及位图数组 fd_set。其底层实现依然是通过位运算，根据文件描述符编号定位到对应的比特位，然后检测该比特位是否为 1。若结果为非零，则说明该文件描述符处于集合中（即该事件就绪），否则返回 0。

int FD_ISSET(int fd, fd_set *set);

// 返回值：如果fd在集合中返回非0，否则返回0

// 使用示例
if (FD_ISSET(sockfd, &readfds)) {
    // sockfd可读
    recv(sockfd, buf, sizeof(buf), 0);
}

此外，还有一个接口 FD_CLR，其作用是从位图数组中清除某个文件描述符，即将对应的比特位从 1 重新设置为 0，表示不再检测该文件描述符。该函数同样接收文件描述符以及位图数组两个参数，其中位图数组属于输出型参数。

不过在实际开发中，相比于前面几个接口，FD_CLR 的使用频率通常较低。

void FD_CLR(int fd, fd_set *set);

// 使用示例
FD_CLR(sockfd, &readfds);   // 将sockfd对应的位清除为0

---

接下来，我们结合上述接口以及 select 接口，来说明 select 的典型使用方式。

select 接口在实际使用中经常被认为较为复杂，其中一个重要原因就在于：除第一个参数外，其余参数全部为输入输出型参数。这意味着每次调用 select 之后，这些集合都会被内核修改，因此需要在下一次调用之前重新进行构造。

对于一个服务端进程而言，其初始阶段通常需要完成如下步骤：

1. 创建监听套接字（获取监听套接字对应的文件描述符）
2. 绑定地址（bind）
3. 将监听套接字设置为监听状态（调用 listen）

在传统的实现方式中，主线程在完成 listen 之后通常会进入一个 while 循环，并不断调用 accept 获取新的连接。一旦 accept 成功返回已连接套接字，就将该连接分配给线程池中的工作线程进行处理。

而在采用 I/O 多路复用模型 之后，整体逻辑会发生变化。对于主线程而言，其不再只关注监听套接字是否就绪，而是同时关注：

• 监听套接字的读事件（表示有新的连接到来）
• 已连接套接字的读事件（表示客户端发送了数据）

因此，主线程在调用 listen 之后，不会直接进入一个仅调用 accept 的循环，而是首先初始化一个读集合。由于服务端刚启动时只有监听套接字，因此初始状态下只需要将监听套接字的文件描述符加入读集合中。同时还需要准备 timeval 结构体，用于指定 select 的超时时间。通常在实际系统中，select 并不会无限阻塞，而是设置一定的超时时间。

完成读集合和 timeval 的初始化之后，程序进入主循环。

由于读集合会作为输入输出型参数传递给 select，因此内核在返回时会直接覆盖该位图数组。如果下一轮循环继续使用该集合，就必须重新构造。因此这里不能直接将原始读集合传递给 select，而是需要在循环内部创建一个副本（临时集合），然后将该副本传递给 select。

此外，在服务端程序中，还需要额外维护一个辅助数组。原因在于：对于 TCP 连接而言，一旦客户端主动关闭连接，服务端也会关闭对应的已连接套接字，此时该文件描述符将不再有效。因此，服务端需要动态维护当前仍然需要检测的文件描述符集合。

换句话说，已连接套接字的生命周期是动态变化的。因此需要一个用户态的数据结构来记录当前有效的文件描述符集合。在本实现中，这个结构就是所谓的读集合辅助数组。

该辅助数组本质上是：

用户态的“真实状态（source of truth）”

而 fd_set 只是：

每次调用 select 时提交给内核的瞬时快照（snapshot）

因此，在调用 select 之前，需要根据辅助数组重新构造读集合，并计算当前最大的文件描述符值（max_fd）。随后再复制一份集合副本传递给 select。

当 select 返回之后，需要检查其返回值：

• 大于 0：表示有文件描述符就绪，此时调用专门的处理函数判断是监听套接字还是已连接套接字就绪
• 等于 0：表示超时，继续下一轮循环
• 小于 0：表示 select 调用失败，此时通常需要记录日志并退出

需要注意的是，select 的最后一个参数（timeval）同样是输入输出型参数，内核在返回时会修改其中的剩余时间。因此在每一次循环开始之前，都需要重新初始化 timeval 结构体。

代码示例如下：

class Select_Server{
Select_Server(uint16_t _port=8888, std::string _ip=_deafault)
                :port(_port)
                , ip(_ip)
            ,read_tp(threadpool::getinstance())
        {       
                for (int i = 0; i < fd_num_max; i++)
                {       
                        read_fd_array[i] = -1;
  }
          for(int i=0;i<fd_num_max;i++)
          {
            write_fd_array[i]=-1;
          }
        }
    //....
     bool init()
        {
                listensock.socket();
                listensock.bind(ip, port);
        return true;
        }
      void handlertask(fd_set* read_set)
        {
               //...
        }
void start()
        {
                signal(SIGPIPE, SIG_IGN); // 🌟 全局忽略 SIGPIPE 信号
                listensock.listen();
                read_tp.start();
                 //...
                int listensock_fd = listensock.fd();
                read_fd_array[listensock_fd] = 1;
                fd_set read_set;
                struct timeval time = { 5,0 };
        int max_fd=-1;
                while (1)
                {
            FD_ZERO(&read_set);
            max_fd=-1;
           for(int i=0;i<fd_num_max;i++)
           {
              if(read_fd_array[i]==1)
              {
                FD_SET(i,&read_set);
                if(i>max_fd)
                {
                    max_fd=i;
                }
              }
           }
           fd_set temp_set=read_set;
           time={5,0};
           int n = select(max_fd + 1, &temp_set, NULL, NULL, &time);
            if(n<0)
         {
                lg.logmessage(Fatal,"select error");
                 break;
            }else if(n==0)
            {
                lg.logmessage(info,"select timeout");
            }else{
                        handlertask(&temp_set);
            }
                }
        }
private:
        sock listensock;
        uint16_t port;
        std::string ip;
        int read_fd_array[fd_num_max];
         int write_fd_array[fd_num_max];
        threadpool& read_tp;
    //...
};

而一旦 select 的返回值 大于 0，则说明当前已经存在读条件就绪的文件描述符。此时程序就会进入 handlertask 函数进行处理。该函数的核心职责是：判断具体是哪个文件描述符发生了读事件，以及该事件对应的类型。

具体实现方式是：遍历传递给 select 接口的读集合（即被内核修改后的 fd_set），并依次检查每一个文件描述符是否在集合中。如果某个文件描述符对应的比特位仍然为 1，则说明该文件描述符已经就绪。

随后需要根据文件描述符的类型进行不同的处理：

1. 如果就绪的是监听套接字
   说明有新的客户端连接到达，此时需要调用 accept 接口获取新的连接，即获取对应的已连接套接字文件描述符。随后将该文件描述符加入到辅助数组中，以便在下一次调用 select 之前能够重新构造读集合，从而持续监控该连接的读事件。
2. 如果就绪的是已连接套接字
   说明对应客户端已经向服务端发送了数据。此时主线程不会直接处理数据，而是将该文件描述符封装为任务对象，并分配给线程池中的工作线程进行处理，从而实现主线程负责事件分发，工作线程负责业务处理的并发模型。

对应代码如下：

class Select_Server{
   //...
   void Accepet()
        {
                struct sockaddr_in Client;
                memset(&Client, 0, sizeof(Client));
                socklen_t Client_len = sizeof(Client);
                int connfd = listensock.accept(&Client,&Client_len);
                if(connfd<0)
                {
                    lg.logmessage(Fatal,"accept error");
                    return;
                }

        if(connfd>=fd_num_max)
        {
            lg.logmessage(Fatal,"fd overflow");
            return;
        }

                read_fd_array[connfd]=1;
                last_active_time[connfd]=time(NULL);

                send_queues_mtx.lock();
                send_queues[connfd]=new SendQueue;
                send_queues_mtx.unlock();
        }

  void Distribute(int fd)
        {
                   read_fd_array[fd]=-1;
                   Task t(fd,read_fd_array,write_fd_array,last_active_time);
                   read_tp.push(t);
        }

        void handlertask(fd_set* read_set)
        {
                for (int i = 0; i < fd_num_max; i++)
                {
                        if (FD_ISSET(i, read_set))
                        {
                                if (i == listensock.fd())
                                {
                                        Accepet();
                                }
                                else
                                {
                                        Distribute(i);
                                }
                        }
                }
        }
        //...
}

在该实现中，handlertask 函数本质上承担的是**事件分发器（Event Dispatcher）**的角色：

• 如果监听套接字就绪 → 处理新连接建立
• 如果已连接套接字就绪 → 将任务交由线程池处理

这种设计使得主线程始终保持轻量，仅负责 I/O 事件检测与任务分发，而具体的业务处理则由线程池中的工作线程完成，从而提高整体系统的并发处理能力。

另外需要注意的一点是，在 Distribute 函数中：

read_fd_array[fd] = -1;

该操作的含义是：在任务被分发给工作线程之后，暂时停止主线程对该文件描述符读事件的监控，以避免在工作线程尚未处理完成时再次触发读事件，从而导致同一连接被重复分发处理。这实际上是一种简单的并发保护策略。

select线程（主线程）
        │
        │ 事件检测
        ▼
handlertask（事件分发）
        │
        ├── 新连接 → accept
        │
        └── 已连接FD → 投递线程池
                         │
                         ▼
                    工作线程处理业务

那么认识了如何调用 select 接口来实现 I/O 多路复用 之后，接下来需要进一步了解 select 接口本身所存在的一些局限性。

首先，select 所能够检查的文件描述符数量是存在上限的。通常情况下，其最多只能检查 1024 个文件描述符。然而，对于一个进程而言，其文件描述符表的容量实际上是可以超过 1024 的。因此，当服务端主线程不断接收新的连接，并通过 accept 获取已连接套接字对应的文件描述符时，就可能出现一个问题：部分文件描述符无法被 select 进行监听。

具体来说，select 所使用的 fd_set 本质上是一个 位图数组（bitmap），其大小由宏 FD_SETSIZE 决定。在绝大多数系统实现中，FD_SETSIZE 被硬编码为 1024。这意味着该位图数组仅包含 1024 个比特位，因此只能表示 0～1023 范围内的文件描述符。由于文件描述符的数值本身就是文件描述符表中的数组下标，一旦某个文件描述符的值 ≥1024，那么在位图中就不存在对应的位置，也就无法通过 FD_SET 将其加入到监听集合中。

在这种情况下，即便服务端已经通过 accept 获取到了新的已连接套接字，其对应的文件描述符也无法加入 select 的监听集合。因此主线程无法判断该套接字的 读事件是否就绪，也无法将其纳入事件驱动的处理流程。实际工程中通常只能选择拒绝该连接，或者采用其他机制进行处理。由此可见，这一限制直接约束了 select 的可扩展性与并发能力，这是 select 最显著的局限之一。

其次，select 在事件检测过程中存在 O(N) 的时间复杂度开销。select 接口需要传入读集合、写集合以及异常集合，这些集合均以 位图数组 的形式表示。需要注意的是，这些集合中既包含真正需要检测的文件描述符，也包含未被使用的位置。因此在内核侧实现中，每次执行 select 时，内核都必须遍历 fd_set 位图来检查对应文件描述符的状态，其时间复杂度为 O(N)。

此外，当 select 从内核态返回用户态（即进程从阻塞状态被唤醒）时，用户态进程同样需要再次遍历一遍传递给 select 的集合，以找出哪些文件描述符真正就绪。这一遍历过程同样具有 O(N) 的复杂度，并且其中往往包含大量的无效遍历，因此，在文件描述符数量较多的场景下，这种 “内核一次遍历 + 用户态再一次遍历” 的模式会带来明显的性能开销。

最后一个常被诟病的问题是：select 接口的大部分参数都是 输入输出型参数（in-out parameter）。除了第一个参数 nfds 仅作为输入参数之外，其余参数（读集合、写集合和异常集合）在调用过程中都会被内核原地修改。也就是说，当 select 返回时，这些集合中只会保留已经就绪的文件描述符。

这就导致在实际编程时，我们通常需要维护一个额外的辅助数组，用于保存当前需要监听的全部有效文件描述符。在每一次调用 select 之前，都必须根据该辅助结构重新构造一份 fd_set 位图，并且通常还需要准备一个副本传递给 select，以避免原始集合被破坏。这不仅增加了代码的复杂度，也带来了额外的内存操作开销。

综上所述，select 在设计上存在多方面限制：包括 文件描述符数量上限较低、事件检测效率为 O(N)、以及接口参数设计带来的使用复杂度。这些因素使得 select 难以支撑真正的高并发服务器模型。因此，在现代高并发网络编程中，select 的实际应用场景已经相对较少，并逐渐淡出主流使用范围。

不过，在学习 I/O 多路复用机制时，理解 select 的工作原理仍然具有重要意义。正是由于 select 暴露出的这些问题，后续系统接口（例如 poll 和 epoll）才在设计上针对这些缺陷进行了改进。因此，先理解 select 的设计思想与局限，有助于我们更加自然地过渡到后续 poll 与 epoll 的学习，并为深入掌握这些高性能 I/O 多路复用机制打下基础。

poll

那么接下来就让我们过渡到 poll。根据上文的分析，我们已经认识到 select 接口所存在的一些局限性，而 poll 接口 的设计初衷正是为了在一定程度上缓解这些问题。

首先，select 的一个重要限制在于：其内部使用 位图（bitmap） 来表示需要检测的文件描述符集合，而该位图的长度在实现上通常是固定的。因此，select 能够检测的文件描述符数量是有限的。在 Linux 的默认实现中，通常只能检测 文件描述符值小于 1024 的情况。一旦文件描述符的值超过该范围，主线程就无法再将其加入到对应的读集合或写集合中，这就对高并发场景下的连接规模形成了明显限制。

其次，在 select 的实现中，无论是 用户态 还是 内核态，都需要对整个位图数组进行遍历。在遍历过程中，大量位置实际上并没有对应有效的文件描述符，因此会产生较多 无效遍历（invalid traversal），从而带来额外的时间开销。

此外，select 的接口参数中，除了第一个参数（最大文件描述符值 +1）以外，其余多个参数都属于 输入输出型参数（in-out parameters）。这意味着调用前需要由用户态重新初始化，而调用返回后又需要从同一块内存中读取结果，这种设计在接口语义上并不十分清晰，也增加了使用复杂度。

正因为上述这些问题，poll 接口被引入，以对 select 的这些缺陷进行改进。

• 系统调用：poll
• 头文件：<poll.h>
• 函数声明：int poll(struct pollfd *fds, nfds_t nfds, int timeout)
• 返回值：成功：返回 revents 非零的结构体数量；超时：返回 0；错误：返回 -1 并设置 errno。

poll 的第一个参数是一个指针，该指针指向一个 结构体数组。数组中的每个元素类型为 struct pollfd，用于描述一个需要被监视的文件描述符及其对应的事件。

struct pollfd {
    int   fd;      /* 要监视的文件描述符 */
    short events;  /* 关心的事件（输入） */
    short revents; /* 实际发生的事件（输出） */
};

可以看到，struct pollfd 结构体包含三个字段。该结构体的设计在一定程度上解决了 select 接口中 输入输出参数混合 的问题。

需要注意的是：整个结构体数组在语义上仍然属于 输入输出型参数，但是 poll 通过结构体内部字段的划分，使 输入信息与输出信息在逻辑上进行了分离。其中：

• events 字段作为 输入参数，用于告诉内核当前关心哪些事件；
• revents 字段作为 输出参数，由内核在返回时填写，用于表示该文件描述符上实际发生的事件。

此外，这个结构体数组中的每一个元素都显式地描述了一个 需要被监视的文件描述符（即fd字段），因此不存在像 select 位图那样的大量无效位置。内核只需要遍历该数组，读取 fd 字段，就可以有针对性地检测对应文件描述符的状态，从而减少不必要的遍历开销。

因此可以认为，struct pollfd 的引入在接口层面上同时改进了两点：

1. 输入与输出信息逻辑分离，使接口语义更加清晰；
2. 避免位图结构带来的大量无效遍历问题。

接下来再来看 events 字段。该字段属于 输入型参数，用于告诉内核需要检测哪些类型的事件，例如读事件、写事件等。而 revents 字段则由内核在返回时填写，其值同样由 Linux 内核定义的一组 事件宏（event flags） 表示：

#define POLLIN     0x001    // 二进制：0000 0000 0001  有数据可读
#define POLLPRI    0x002    // 二进制：0000 0000 0010  有紧急数据可读
#define POLLOUT    0x004    // 二进制：0000 0000 0100  可以写入数据
#define POLLERR    0x008    // 二进制：0000 0000 1000  发生错误（仅输出）
#define POLLHUP    0x010    // 二进制：0000 0001 0000  挂起（仅输出）
#define POLLNVAL   0x020    // 二进制：0000 0010 0000  无效请求（fd 未打开，仅输出）

由于这些事件宏本质上是 位标志（bit flags），因此如果需要同时检测多个事件，只需要通过 按位或运算（bitwise OR） 将多个事件组合即可。例如同时检测读事件和写事件，可以设置为：

events = POLLIN | POLLOUT

在实际执行过程中，poll 的底层行为大致可以概括为以下步骤：

1. 内核接收到用户态传入的 struct pollfd 数组；
2. 遍历该数组，读取每个元素的 fd 字段；
3. 根据 fd 找到对应的 file 结构体；
4. 根据 events 字段判断需要检测的事件类型；
5. 调用该文件对象 file_operations 中的 poll 函数（即 f_op->poll）；
6. 若对应条件满足，则在 revents 字段中设置相应的就绪事件标志。

poll 的第二个参数 nfds 表示 结构体数组的长度，即需要监视的文件描述符数量。

最后一个参数 timeout 是一个 输入型参数，用于指定 poll 的超时时间（单位为毫秒）。如果在指定时间内没有任何事件就绪，那么 poll 会因为 超时 而返回，此时返回值为 0，进程也会从阻塞状态被唤醒。

需要特别注意的是：即使使用 poll，其底层仍然不可避免地会存在 O(N) 级别的遍历操作。具体来说，仍然包括两部分遍历：

1. 内核态遍历：内核遍历 pollfd 数组，检查每个文件描述符的状态；
2. 用户态遍历：poll 返回后，用户程序仍然需要遍历该数组，以确认哪些 revents 字段被设置，从而确定哪些文件描述符已经就绪。

不过，与 select 相比，poll 的遍历对象不再是固定大小的位图，而是 只包含有效文件描述符的结构体数组。因此，无论在内核态还是用户态，都可以显著减少无效遍历的数量，从而在实践中获得更好的效率表现。

此外，poll 也 不再对文件描述符的最大值做固定限制。由于其使用的是数组结构，并通过 nfds 参数显式指定数组长度，因此该数组可以是 动态分配的。只要传入正确的 nfds 值，内核就能够知道需要检查多少个文件描述符，从而在接口层面解决了 select 的 文件描述符数量上限问题。

用Poll实现高并发的HTTP服务器

Poll_Server

至此，上文已经为我们建立了 poll 的理论基础，并了解了如何使用 poll 接口。接下来，我们便可以着手实现一个 HTTP 服务器。在实现过程中，我们选择使用 C++ 进行代码编写。由于 C++ 是一门典型的 面向对象编程语言，因此这里将服务器抽象为一个对象，通过一个 Poll_Server 类来进行描述。

而这个 Poll_Server 类首先必然会包含一个 监听套接字。这里我们将套接字进一步封装为一个 sock 类，其中 sock 类内部包含一个原始的 文件描述符（file descriptor），并且将所有与套接字相关的系统调用接口封装为 sock 类的成员函数，例如 socket 成员函数以及 accept 成员函数等。同时，对这些系统调用的返回值也进行了统一处理：如果出现错误，则会记录相应的日志并终止程序执行。

此外，该封装还应用了 RAII（Resource Acquisition Is Initialization）思想，即将套接字资源的生命周期交由对象自动管理。当 sock 对象被销毁时，其析构函数会自动调用 close 释放套接字资源，从而避免资源泄漏。因此，Poll_Server 类会包含一个 sock 类型的对象，该对象用于表示监听套接字。同时，Poll_Server 类中还会包含一个 string 类型的成员变量，用于表示 IP 地址，以及一个 uint16_t 类型的成员变量，用于表示 端口号。

Poll_Server 类的构造函数主要用于初始化监听套接字绑定的 IP 地址和 端口号。该构造函数接收两个参数，分别用于初始化 string 类型的 IP 地址以及端口号，并且这两个参数都设置了 缺省值。其中 IP 地址的缺省值为 "0.0.0.0"，表示监听本机所有网络接口发送到该端口号的数据报。

接下来是 init 函数。init 函数的主要职责是完成服务器启动前的初始化工作。首先会创建一个监听套接字，即调用 sock 对象的 socket 接口。成功创建监听套接字之后，接下来会调用 bind 函数，使该监听套接字绑定指定的 IP 地址和端口号。

除此之外，这里还会进行一个额外的配置：将监听套接字的默认行为设置为 非阻塞模式。至于为什么需要将其设置为非阻塞，这一点将在后文进行详细解释，这里先暂时埋下一个伏笔。因此，init 函数完成的主要工作就是：

1. 创建监听套接字
2. 绑定 IP 地址与端口号
3. 将监听套接字设置为非阻塞模式

如果以上操作全部成功，则 init 函数返回 true。

class Poll_Server
{
public:
        Poll_Server(uint16_t _port=8888, std::string _ip=_deafault)
                :port(_port)
                , ip(_ip)
        {
           
        }
      bool init()
        {
                listensock.socket();
                listensock.bind(ip, port);
                listensock.setnonblock();
                return true;
        }
    //...
    private:
        sock listensock;
        uint16_t port;
        std::string ip;
        //...
};

当 init 函数调用成功之后，接下来便会调用 start 函数。start 函数首先会将监听套接字设置为 监听状态，也就是调用 listen 接口。当监听套接字成功进入 listen 状态之后，接下来需要将监听套接字的fd添加进 辅助数组。

pollmanager以及辅助数组以及快照机制

这个辅助数组同样是 Poll_Server 对象中的一个成员变量。我们知道，对于 poll 接口而言，它与 select 不同：poll 并不存在类似 FD_SETSIZE 的固定文件描述符数量限制，因此这里用于维护文件描述符集合的数据结构选择使用 vector 来实现。

具体来说，这个 vector 数组中的每一个元素类型都是 pollfd。之所以不选择使用静态数组，是因为 vector 内部维护的是 动态数组结构，可以在运行过程中进行扩容，而静态数组的长度在编译期已经固定。因此，从设计角度来看，vector 更符合 poll 不限制监控文件描述符数量的特点。

然而，这里又会引入一个新的问题。在之前实现 select 时，我们使用的是 静态数组 来实现辅助数组。由于 select 最多检测 1024 个文件描述符，因此只需要定义一个长度为 1024 的静态数组即可。同时，数组下标天然就可以与 文件描述符的值一一对应。例如，当我们需要将某个文件描述符加入读集合时，只需要将该 fd 作为数组下标，然后直接访问对应位置并设置为 1 即可。

而现在这里采用的是 vector（动态数组） 的方式来维护文件描述符集合。当需要将某个文件描述符加入读集合时，只能通过 尾插（push_back） 的方式加入数组。这就意味着 数组下标与文件描述符之间不再具有天然的映射关系。

这种变化带来的主要影响体现在 删除文件描述符 的场景。例如，当客户端主动关闭连接，或者连接生命周期结束时，服务器也需要关闭对应的套接字。一旦关闭套接字，与之对应的 file 结构体也会被释放，此时对应的 文件描述符失效，就必须将其从读集合对应的 vector 数组中删除。

需要注意的是，这个 vector 中维护的全部都是 当前仍然有效的文件描述符。因此这里的删除操作并不是简单地将某个位置的值覆盖，而是需要真正缩小数组大小，也就是执行 尾删操作（pop_back）。

但是由于 fd 与数组下标没有直接关联，如果需要删除某个指定的 fd，就必须先遍历整个数组找到它的位置，其时间复杂度为 O(N)。而在 高并发服务器 的场景中，服务器在任意时刻都可能接受成千上万个连接，同时也可能关闭成千上万个连接。如果每次删除都需要进行一次 O(N) 的线性遍历，那么开销将会非常巨大。

因此，这里不能仅仅依赖一个 vector 数组，还需要额外维护一个 哈希表（unordered_map）。这个哈希表的作用是 快速定位某个 fd 在 vector 数组中的位置。其中：

• 键（key）：文件描述符 fd
• 值（value）：该 fd 在 vector 数组中的下标

这样，在执行删除操作时，就可以通过哈希表 O(1) 地获取数组下标，然后将待删除元素与 数组最后一个元素交换，再执行 pop_back，即可完成删除操作。

基于这种设计，这里定义了一个 Pollmanager 对象。该对象内部包含：

• 一个 vector<pollfd> 数组
• 一个 unordered_map<int,int> 哈希表

同时还提供了两个接口：

• add_fd：将 fd 添加到读集合或写集合
• del_fd：将 fd 从读集合或写集合中删除

不过在实现这两个接口时，还存在一个非常重要的 并发问题。

由于服务器采用 线程池模型，每一个工作线程都会负责处理一个客户端连接。当某个线程处理完客户端发送的最后一个 HTTP 请求报文之后，会完成请求处理、生成响应并发送给客户端，随后关闭连接，并将该 fd 从读集合中删除。

然而这里的 Pollmanager 对象是一个 全局共享对象，会被多个工作线程同时访问。因此，在运行过程中可能会出现 多个线程同时调用 del_fd 删除 fd 的情况。

而删除操作不仅涉及 vector 数组的修改，还涉及 哈希表的修改。需要注意的是：

• vector::pop_back() 并不是原子操作
• 其内部需要修改 vector 维护的三个指针（start / finish / end_of_storage）中的 finish 指针

这个修改过程实际上包含多条机器指令，例如：

1. 从内存读取当前 finish 指针
2. 在寄存器中执行减一操作
3. 将结果写回内存

如果在执行过程中发生 线程切换，那么其他线程可能会同时修改该指针，从而导致 数据不一致问题。

同样地，unordered_map 也不是线程安全的。其内部本质上是一个 指针数组 + 链表结构。当执行插入操作时，如果负载因子超过阈值，则可能触发 rehash（扩容）。扩容过程中会：

1. 分配新的桶数组
2. 重新计算每个节点的哈希索引
3. 将节点迁移到新的桶中
4. 释放旧的桶数组

而这里访问哈希表中的节点，例如调用 [] 运算符重载函数时，其内部执行流程通常是：首先根据键值计算对应的 哈希索引，然后定位到对应的桶（bucket），该桶通常是一个链表或其他冲突解决结构，接着再遍历该链表中的节点，最终返回节点 val 值的引用。

然而，这一整个访问过程同样 不是原子操作。如果在访问过程中发生线程并发，例如某个线程正在访问哈希表，而另一线程触发了哈希表的 扩容（rehash），那么问题就可能出现。

具体来说，在 rehash 过程中，哈希表会重新申请一块新的桶数组，然后将原有节点重新计算哈希索引并迁移到新的桶数组中，随后 释放旧的桶数组内存。如果此时某个线程正在执行查找操作，例如已经根据键值计算出了哈希索引，并准备访问旧桶数组中的链表节点，但在这一过程中线程发生了切换，而另一线程完成了 rehash 并释放了旧的桶数组，那么当前线程恢复执行后就可能继续访问 已经被释放的内存地址。

这种情况会直接导致 非法内存访问（invalid memory access），从而引发程序崩溃或者未定义行为。因此，在并发环境下，如果没有进行同步控制，对哈希表进行读写访问同样可能导致 数据不一致问题，甚至产生更严重的内存安全问题。

因此，如果不进行同步控制，vector 与 unordered_map 在并发访问下都可能产生 数据不一致甚至崩溃的问题。

为了解决这一问题，这里需要引入 互斥锁（mutex）。锁的粒度设计为：同时保护 vector 与 unordered_map 这两个数据结构。也就是说，在执行 add_fd 或 del_fd 时，必须首先获取这把锁，随后才能对这两个数据结构进行访问和修改。

这样做的目的，是保证在访问期间：

• 哈希表结构不会发生变化
• vector 内部状态保持一致

从而避免并发导致的数据破坏问题。

因此，Pollmanager 对象除了 vector 和 unordered_map 之外，还需要维护一把 互斥锁。

接下来即可实现 add_fd 函数。函数首先尝试获取锁（加锁），如果加锁成功，则先检查该 fd 是否已经存在于 vector 数组中。如果已经存在，说明哈希表中已经有对应映射，此时直接返回即可，避免重复添加。

如果不存在，则创建一个 pollfd 对象并进行初始化。初始化时不仅需要设置 fd 字段，还需要设置 events 字段。在当前服务器的设计场景下，主要涉及两类事件：

• 接收新连接或接收客户端请求 —— 读事件
• 向客户端发送响应 —— 写事件

因此 Pollmanager 内部维护一个 _event 标志位：

• 1 表示读集合
• 2 表示写集合

在初始化 events 字段时，根据该标志位设置相应的事件宏（POLLIN 或 POLLOUT）。

完成初始化之后，将该 pollfd 尾插到 vector 数组中，并在哈希表中建立对应的 fd → index 映射关系。

对于 del_fd 函数，同样首先获取锁。随后判断该 fd 是否已经被删除，即查询哈希表。如果未命中，则说明已经被删除，直接返回。

如果命中，则继续执行删除逻辑：

1. 记录当前 fd 在数组中的索引
2. 判断该元素是否为数组最后一个元素

如果是最后一个元素，则直接 pop_back 即可；
如果不是，则需要：

1. 获取数组最后一个元素
2. 更新该元素在哈希表中的索引
3. 将最后一个元素覆盖到待删除位置
4. 执行 pop_back

在锁的使用方面，这里采用的是 C++ 标准库中的 std::lock_guard。该类同样遵循 RAII 思想：当 lock_guard 对象创建时自动加锁，当函数结束时对象析构，自动调用析构函数完成解锁，因此无需手动释放锁。

class Pollmanager
{
    public:
    Pollmanager(int _event)
    :event(_event)
    {

    }
    void add_fd(int fd)
    {
       std::lock_guard<std::mutex> lock(mtx);
       if(fd_index_map.find(fd)!=fd_index_map.end())
       {
             return;
       }
       pollfd pfd;
       pfd.fd=fd;
       if(event==1)
       {
         pfd.events=POLLIN;
       }else if(event==2)
       {
        pfd.events=POLLOUT;
       }
       pfd.revents=0;
       pollfds.push_back(pfd);
       fd_index_map[fd]=pollfds.size()-1;
    }
    bool del_fd(int fd)
    {
       std::lock_guard<std::mutex> lock(mtx);
       if(fd_index_map.find(fd)==fd_index_map.end())
       {
         return false;
       }
       int index=fd_index_map[fd];
       if(index==pollfds.size()-1)
       {
           pollfds.pop_back();
           fd_index_map.erase(fd);
           return true;
       }
       int last_fd=pollfds.back().fd;
       pollfds[index]=pollfds.back();
       pollfds.pop_back();
       fd_index_map[last_fd]=index;
       fd_index_map.erase(fd);
        return true;
    }
    std::vector<pollfd> get_snapshot()
    {
        std::lock_guard<std::mutex> lock(mtx);
        return pollfds;
    }
    private:
    std::mutex mtx;
    std::vector<pollfd> pollfds;
    std::unordered_map<int,int> fd_index_map;
    int event;
};

而这里还需要注意一个关键问题。由于当前锁的粒度是 同时保护整个哈希表以及整个 vector 数组，因此无论是主线程还是线程池中的工作线程，在访问读集合时都会涉及并发访问这一共享数据结构。

其中最核心的问题出现在 主线程。主线程需要调用 poll 接口，而 vector 数组中维护的动态数组正是当前 有效且需要被检测的文件描述符集合。在调用 poll 时，需要将 vector 内部维护的动态数组传递给 poll 接口，而这个数组在语义上属于 输入输出型参数：用户态将其传递给内核，随后内核会遍历该数组并检查对应文件描述符的就绪状态，并在返回时修改其中的 revents 字段。

然而需要注意的是，其他工作线程同样可能会并发访问该 vector 中的动态数组，例如调用 add_fd 或 del_fd 对文件描述符集合进行修改。如果此时主线程直接将 vector 内部的动态数组传递给 poll，那么为了保证该数组在 poll 阻塞等待期间 结构保持一致且不被修改，就必须一直持有锁。

换言之，如果主线程在调用 poll 之前加锁，并在 poll 阻塞等待期间一直持有该锁，那么在整个等待期间，只有内核在访问这个数组，而所有其他线程都无法获取这把锁。这就会导致工作线程在执行 add_fd 或 del_fd 时无法获取锁，从而长期阻塞，严重情况下甚至可能导致 线程饥饿（thread starvation）。

为了解决这一问题，就需要引入 快照（snapshot）机制。具体做法是：主线程在调用 poll 之前，先短暂地获取锁，然后获取当前时刻下 vector 数组的状态，即 拷贝一份 vector 数组副本。这个拷贝过程是短暂的，相比于之前主线程在阻塞等待期间一直持有锁的方式，这种做法只在拷贝阶段短时间持有锁。

当拷贝完成之后，主线程会立即释放锁，而后续的 poll 调用则基于这份 vector 副本 进行检测。这样一来，即使 poll 在内核中阻塞等待，主线程也不会再持有锁，其他工作线程仍然可以正常竞争该锁并对原始 vector 数组进行修改。这正是 get_snapshot 函数的核心作用。

此时读者可能会产生一个疑问：既然工作线程也会并发调用 add_fd 和 del_fd，而这些函数同样需要竞争同一把锁，那么这里是否仍然可能出现 线程饥饿问题？

需要注意的是，虽然锁的粒度确实是 保护整个 vector 与哈希表结构，但 add_fd 和 del_fd 这两个函数的执行过程非常短暂。线程在获取锁之后，只会进行少量的数据结构操作，例如插入、删除以及更新映射关系，因此锁的持有时间非常短。线程在完成操作之后会立即释放锁，因此即使存在锁竞争，也只会产生 短时间的临界区等待，而不会像主线程直接持锁调用 poll 那样，在阻塞等待期间长时间占用锁。

因此，引入 快照机制 的核心目的，就是避免主线程在调用 poll 阻塞等待时 长期持有锁，从而保证其他工作线程仍然能够及时获取锁并修改文件描述符集合，提高整个服务器在并发场景下的可伸缩性。

class Poll_Server
{
//...
private:
  Pollmanager read_pollmanger{1};
  //...
};

那么这里读者可能会产生一个疑问：引入 快照（snapshot）机制 之后，主线程可以在调用 poll 之前短暂持有锁，从而获取当前时刻 vector 的状态，然后拷贝一份 vector 副本，随后立即释放锁，并基于这份副本调用 poll 接口。接下来，内核会基于该 vector 副本进行遍历，并检测其中各个 fd 的事件条件是否就绪。

然而，由于服务器中的线程是 并发执行 的，可能会出现这样一种情况：某个工作线程在处理请求时，向客户端发送响应报文，但 send 调用失败，于是该线程关闭了对应的套接字 fd。而对于内核来说，此时 poll 所持有的仍然是 之前时刻获取的快照副本。因此，就会出现 快照中的状态与当前真实状态不一致 的情况。

例如，当内核在遍历该 vector 副本时，某个 fd 在当前系统状态下已经被关闭。那么这里就会产生一个问题：这种情况是否会导致程序崩溃，或者引发其他异常行为？

实际上不会。这里需要注意的是，当内核在遍历 pollfd 数组时，如果发现某个 fd 已经 不再是一个有效的文件描述符，那么内核不会崩溃，而是会在该 fd 对应的 pollfd 结构体中的 revents 字段设置 POLLNVAL 标志位。同时，poll 的返回值（表示就绪的文件描述符数量）不会将该 fd 计入统计结果。

因此，当 poll 调用返回之后，主线程会遍历这份 vector 数组，并检查每个 pollfd 结构体中的 revents 字段。通常的做法是将 revents 与需要关注的事件宏进行 按位与运算，例如：

• revents & POLLIN
• revents & POLLOUT

如果结果非零，则说明该 fd 对应的事件已经就绪。

而在刚才描述的场景中，由于 revents 中被设置的是 POLLNVAL 标志位，因此在执行上述按位与运算时，并不会与 POLLIN 或 POLLOUT 匹配成功。换句话说，这个 pollfd 结构体在遍历过程中会被 自动忽略。

因此，即使出现 快照状态与当前真实状态不一致 的情况，也不会对程序造成影响。实际上，这种情况在并发服务器中是 允许出现的正常现象，并不需要额外的防御性处理。原因在于：

1. POLLIN 与 POLLOUT 的位掩码与 POLLNVAL 互不重叠
2. 在事件检测阶段通过 按位与运算 会自动过滤掉这些无效或已经关闭的 fd

因此，在遍历 poll 结果时，这些已经失效的文件描述符会被自然地忽略，不会引发崩溃或逻辑错误，也无需额外的特殊处理。

---

start()

那么接下来我们就可以进一步完善 start 函数的逻辑了。首先，start 函数会调用 listen 接口，将监听套接字设置为 监听状态。随后，会将监听套接字对应的 fd 加入到 读集合 中，用于检测新的连接事件。

完成上述初始化之后，程序会进入一个 while 死循环。在进入该循环之前，还需要进行一个额外的处理：忽略 SIGPIPE 信号。

这里需要解释一下原因。如果客户端正常关闭连接，也就是调用了 close 接口，那么客户端会向服务端发送一个 FIN 报文。此时服务端的 TCP 状态机会感知到对端关闭连接。如果服务端此时仍然尝试向客户端发送数据，那么在第一次写操作时，服务端会收到对方返回的 RST 报文；如果此后再次对该连接执行写操作，则会触发 SIGPIPE 信号。另外，如果客户端是 异常退出（例如进程被 kill 终止），那么操作系统通常也会向对端发送 RST 报文。在这种情况下，如果服务端继续调用 send 或 write 向该连接发送数据，同样可能触发 SIGPIPE 信号。

但是这里实际上往往 来不及进行第二次写操作的检测。因为在很多情况下，第一次 send 调用是会成功返回的，应用层并不会立即感知到对端已经关闭连接。

其典型过程如下：

对端调用 close，发送 FIN
  → 本端收到 FIN，但本端尚未关闭连接，此时 TCP 进入半关闭状态（对端关闭写，本端仍可写）
  → 本端第一次 send：数据成功进入发送缓冲区，TCP 将数据发送出去
    应用层得到成功返回，因此完全感知不到对端已经关闭
  → 对端收到该数据，但由于对端已经关闭读取方向，因此返回 RST
  → 本端收到 RST，TCP 将该连接状态标记为异常
  → 本端第二次 send：内核检测到该连接已经处于异常状态，从而触发 SIGPIPE
    同时 send 返回 -1，并将 errno 设置为 EPIPE

因此，从应用层的角度来看：

• 第一次 send 通常会成功返回，因为数据只是被写入本端发送缓冲区；
• 只有在 第二次写操作 时，内核才会检测到连接已经处于异常状态，从而触发 SIGPIPE 信号。

而 SIGPIPE 的默认行为是 终止进程。因此，如果服务器程序没有显式忽略该信号，那么在向已经关闭的连接再次写入数据时，进程就可能被操作系统直接终止。

这也是为什么在服务器程序启动时，通常会显式 忽略 SIGPIPE 信号，从而避免因为客户端关闭连接而导致整个服务器进程异常退出。

需要注意的是，SIGPIPE 信号的 默认行为是终止进程。而对于服务器程序来说，这是完全不可接受的。服务器通常需要 长期稳定运行（例如 24 小时持续提供服务），不可能因为某个客户端关闭连接或者异常退出，就导致整个服务器进程被终止。换句话说，服务器不应该因为客户端的退出而“陪葬”。

因此，在进入主循环之前，需要显式忽略 SIGPIPE 信号，以避免服务器在写入已经关闭的套接字时被系统终止。

完成信号处理之后，程序便进入 while 循环。在每一次循环中，首先会调用读集合对应的 get_snapshot 函数，获取当前时刻 vector 的状态。该函数的执行流程是：

1. 先获取锁
2. 拷贝当前 vector 数组
3. 释放锁

从而得到一份 当前时刻的 vector 副本。

随后，主线程会调用 poll 接口，并将这份副本传递给 poll，同时还会传入 数组长度 以及 超时时间。此时主线程会进入阻塞等待状态，直到：

• 有文件描述符事件就绪
• 超时时间到达
• 或者系统调用被中断

当 poll 返回之后，主线程首先需要检查返回值：

• 如果 返回值大于 0，说明存在事件就绪的文件描述符；
• 如果 返回值等于 0，说明本次 poll 调用发生了超时；
• 如果 返回值小于 0，说明 poll 调用失败，此时需要记录错误日志并结束程序。

对于返回值 大于 0 的情况，接下来就会进入 事件分发阶段，也就是调用 handlertask 函数，对已经就绪的文件描述符进行进一步处理。

clas Poll_Server
{
    //...
void start(bool isdaemon=false)
        {
                signal(SIGPIPE, SIG_IGN); // 🌟 全局忽略 SIGPIPE 信号
                  if(isdaemon)
                {
                    Daemon d;
                    d.daemon();
                }
               //..
                listensock.listen();
                //..
                int listensock_fd = listensock.fd();
                read_pollmanger.add_fd(listensock_fd);
                while (1)
                {
                  std::vector<pollfd> poll_fds =read_pollmanger.get_snapshot();
                   int n=poll(poll_fds.data(),poll_fds.size(),10);
            if(n<0)
         {
                lg.logmessage(Fatal,"poll error");
                 break;
            }else if(n==0)
            {
                lg.logmessage(info,"poll timeout");
            }else{
                        handlertask(&read_pollmanger,poll_fds);
            }
                }
        }
}

接下来，start 函数还提供了一个 可选的守护进程化功能。函数接收一个 bool 类型的参数 isdaemon，默认值为 false。如果该参数被设置为 true，则会调用 Daemon 类的 daemon() 方法，将当前服务器进程转换为 守护进程（daemon process）。

守护进程（daemon）本质上是一种 长期运行的后台服务进程，通常没有控制终端，并且会在后台持续提供服务，例如 sshd、httpd 等系统服务。

在 Linux 中，一个标准的守护进程化过程通常包括：

1. fork() 创建子进程并让父进程退出
2. 调用 setsid() 创建新会话，脱离控制终端
3. 修改工作目录（通常为 /）
4. 关闭或重定向标准文件描述符
5. 设置 umask 等运行环境

这样进程就能够完全脱离终端，在后台稳定运行。

这种运行方式在实际部署服务器程序时非常常见，因为服务器程序通常需要 长期后台运行，而不依赖终端环境。

---

接下来就是 handlertask 函数的实现。对于 handlertask 函数，其接收一个读集合以及一个 vector 对象，这个 vector 对象实际上是传递给 poll 接口作为输入/输出型参数的快照。随后在 handlertask 函数内部会遍历该快照，并检查究竟是哪一类对象的事件就绪。

这里的对象实际上可以分为两类：监听套接字以及已连接套接字。因此在遍历时，需要判断当前就绪的文件描述符属于哪一种类型。

• 如果是监听套接字就绪，则说明有新的客户端连接到达，此时需要进行对应事件的处理，即调用 Accept 函数获取新连接，并获得新的已连接套接字。
• 如果是已连接套接字就绪，则说明客户端发送了数据，请求已经到达，此时接收缓冲区不为空，需要读取数据。因此这里会进入 Distribute 函数。

而 Distribute 函数的核心逻辑是：创建任务对象。该任务对象中包含已连接套接字对应的 fd，随后将该任务对象放置到环形缓冲区中，等待线程池中的工作线程取走并进行处理。

---

读写分离

首先我们关注一下 Distribute 函数。

当某个已连接套接字的读事件就绪时，对应的文件描述符会被传递给 Distribute 函数。Distribute 函数所做的工作就是构建一个任务对象，然后将其放入环形缓冲区，等待线程池中的线程取走并执行。

此时读者可能会认为：线程池中的工作线程是否就是负责一个连接的处理。从某种角度来说，这种理解并没有问题。

但是这里与传统线程池模型仍然存在差异。在传统模型中，一个工作线程通常会完整负责一个连接的整个生命周期：
即读取请求 → 处理请求 → 构建响应 → 发送响应 → 线程返回线程池。

也就是说，一个线程会一步执行到底：
线程首先读取完整的 HTTP 请求，然后处理该请求，接着构建 HTTP 响应报文，最后再将响应发送回客户端。

然而在学习了IO 多路复用之后，我们知道：一个线程其实可以同时检测多个文件描述符的读写事件是否就绪。因此如果重新审视传统做法，可以发现一个问题：让同一个线程既负责读事件又负责写事件，在效率上并不是最优的。

原因在于，读事件与写事件通常不会连续完成，两者之间往往存在时间间隔。例如：

• 线程完成读取之后，需要等待业务处理
• 写入操作可能会因为发送缓冲区已满而阻塞

也就是说，在读写之间往往存在一个IO 等待阶段。

而我们知道，要提高服务端的并发能力，就需要尽可能减少线程在 IO 等待阶段的时间，让线程更多地运行在 CPU 上执行任务，而不是陷入阻塞等待。

而如果一个线程既负责完成读事件也负责完成写事件，那么一旦写事件未就绪，该线程就可能因为执行写操作而陷入阻塞。对于一个 TCP 长连接来说，客户端通常会在同一连接上连续发送多个 HTTP 请求报文。在这种情况下，如果工作线程因为写事件未就绪而阻塞，那么即使后续请求已经到达并且接收缓冲区中已经存在新的数据，该线程也无法继续处理这些请求。

这样就会导致一种情况：线程被写事件阻塞，从而影响读事件的处理能力，进而降低服务端的整体并发性能。

而实际上，读取完整的 HTTP 请求、处理请求并构建响应这一系列操作，与调用 send 发送响应报文这一动作在逻辑上是可以并发进行的。也就是说，请求的读取与处理并不需要等待响应报文成功发送之后才能继续执行。

因此，通过将读事件与写事件进行职责分离，可以使读线程永远不会因为写事件未就绪而发生阻塞。原因在于，读线程本身并不负责数据发送，其职责仅限于：

1. 读取 HTTP 请求报文
2. 解析并处理请求
3. 构建对应的响应报文

完成这些操作后，读线程只需要将响应数据放入对应连接的发送队列，并通知写线程即可。

由于读线程不参与发送操作，因此即使某个连接因为发送缓冲区已满而暂时无法写入数据，也不会影响读线程继续处理新的请求。这样读线程就可以持续、高效地处理客户端请求，其处理能力不会受到网络发送速度的直接限制，从而显著提升系统的整体吞吐量（Throughput）。

基于这一考虑，这里对读写任务进行了职责分离。所谓分离，是指：

不再让一个线程同时完成读事件与写事件，而是让线程只负责其中一种类型的事件。

---

因此，对于线程池中的工作线程，我更倾向于将其称为读线程。因为线程池中的线程仅负责完成读事件处理：即读取 HTTP 请求、解析请求并构建响应报文。

而写事件则交由专门的写线程来完成，其职责是将构建好的响应报文发送给客户端。因此，在调用 poll 开始检测事件之前，会预先创建一批写线程。

---

但这里会引出一个新的问题。

由于线程之间是相互独立、互不干扰的执行单元，读线程负责读事件，写线程负责写事件。然而对于 TCP 连接来说，读与写在逻辑上其实是存在严格时序关系的。

具体来说：

1. 服务端必须先读取客户端发送的 HTTP 请求报文
2. 然后处理请求并构建响应报文
3. 最后才能将响应报文发送给客户端

然而读线程与写线程的CPU 调度顺序是不可预知的。如果写线程先被 CPU 调度，而此时读线程尚未读取完 HTTP 请求、也尚未构建响应报文，那么写线程实际上是无法执行发送任务的。

因此可以看到，读线程与写线程虽然在职责上是分离的，但在逻辑上仍然存在协作关系。

这种协作关系的核心就在于写集合。写集合用于维护一组需要检测写事件是否就绪的有效文件描述符。

对于读线程来说，当其：

1. 成功读取完整的 HTTP 请求报文
2. 处理请求并构建响应报文

之后，就需要通知写线程可以开始发送响应。

这里所谓的“通知”，实际上就是将对应的 fd 添加到写集合中，从而让写线程后续通过 poll 检测该 fd 的写事件是否就绪。

---

根据上文可知，这里定义了一个 pollmanager 对象。该对象内部包含：

• 一个 vector 数组（用于维护 pollfd 集合）
• 一个哈希表（用于记录 fd 与 vector 数组下标的映射关系）
• 一把用于保护 vector 与哈希表的互斥锁

对于写线程来说，它并不是只负责一个连接的数据发送。如果每个写线程只负责检测一个 fd 的写事件，那么主线程就必须创建大量写线程。

而线程数量过多会带来明显的系统开销，例如：

• 频繁的上下文切换
• 较大的内存占用

因此这里的设计是：每个写线程同样使用 poll 来检测多个 fd 的写事件是否就绪。当检测到某个 fd 写事件就绪时，就可以直接发送对应的响应数据。

换句话说，写线程的职责是：

• 同时检测多个 fd 的写事件
• 对写就绪的 fd 执行响应报文发送

---

但此时仍然存在一个关键问题。

写线程虽然负责发送响应报文，但其前提是：必须能够获取对应的响应报文数据。而响应报文的构建是由读线程完成的。

然而线程之间的线程栈是相互独立的，因此必须设计一个全局共享的数据结构，用于在读线程与写线程之间进行数据交换。

这里我设计了一个哈希表结构。该哈希表的每一个节点都与一个已连接套接字相关联：

• 键（Key）：文件描述符 fd
• 值（Value）：Connection 结构体

Connection 结构体中包含一个重要字段：发送队列（send queue）。

class Connection
{
   public:
  //...
   std::queue<std::string> out_buffer;
   //...
};
 

该发送队列中的每一个节点都对应一个响应报文对象。

因此对于读线程来说，其操作流程为：

1. 根据 fd 在哈希表中查找对应的 Connection 对象
2. 定位到该连接的发送队列
3. 将构建好的响应报文 push 到发送队列中

之所以将该结构设计为发送队列，是因为队列天然具备**先进先出（FIFO）**的特性。

在 HTTP 场景下，客户端可能会连续发送多个请求。例如客户端先发送请求 A，随后发送请求 B。那么服务端在语义上就应当按照顺序返回：

• 响应 A
• 然后再返回响应 B

因此使用队列结构可以很好地保证响应报文的发送顺序与请求顺序保持一致。

---

而这里需要注意的是：对于每一个写线程来说，其都拥有各自独立的写集合，即各自对应一个 pollmanager 对象。pollmanager 对象内部包含一个 vector 动态数组，用于维护需要被 poll 监测的 fd 集合。这意味着写线程可以同时检测多个文件描述符的写事件是否就绪。

由于 vector 是动态扩展的数据结构，因此从结构上来说，写线程所能够监测的 fd 数量并不存在固定上限（仅受系统资源限制）。

根据上文可知，这里为每一个连接设计了发送队列。对于读线程来说，当其读取完整的 HTTP 请求报文并解析请求之后，就会构建对应的响应报文，然后将该响应报文放置到该连接对应的发送队列中，随后需要通知写线程进行发送操作。

而对于写线程来说，其核心逻辑则非常简单：
从对应连接的发送队列队首取出响应报文，然后调用 send 或 write 接口，将该响应报文发送给客户端。

需要注意的是，对于写线程而言，其并不关心具体是哪个 fd 的写事件就绪。写线程只负责：

1. 检测写事件是否就绪
2. 根据就绪的 fd 定位到对应的连接对象
3. 找到该连接的发送队列
4. 取出队首节点
5. 调用 send 接口完成发送

因此，对于读线程来说，其只需要在构建好响应报文之后，将该响应报文交给其中一个写线程负责发送即可。

---

而对于主线程来说，由于每一个写线程都能够检测一组文件描述符集合，因此主线程只需要创建固定数量的写线程即可。因为一个写线程已经可以同时检测多个 fd，如果创建过多写线程，反而会带来额外的系统开销，例如：

• 线程上下文切换成本增加
• 线程栈空间导致的额外内存占用

因此这里的设计是：主线程只创建固定数量的写线程。

同时需要注意的是，这里的写线程与读线程池中的线程模型是不同的。读线程池中的线程会并发访问同一个共享资源——即环形缓冲区，用于从中取走任务对象。而写线程则不同，每个写线程都拥有各自独立的 pollmanager 对象，因此不会出现多个线程并发访问同一个 vector 动态数组的问题。

因此写线程在调用 poll 时不需要像主线程那样对集合进行快照处理。

接下来，对于读线程来说，还需要解决一个问题：将构建好的响应交给哪一个写线程进行发送。

由于写线程数量是固定的，因此可以采用一种简单而高效的负载分配策略：通过模运算来确定写线程编号。例如：

fd % write_thread_num

通过该计算即可确定该 fd 应该由哪个写线程负责发送。

因此，poll_server 对象内部会维护一个 pollmanager 数组，每一个元素都对应一个写线程所维护的 fd 集合。在主线程创建写线程之前，会先创建对应的 pollmanager 对象，并将其尾插到数组中，然后再将该 pollmanager 对象作为参数传递给对应的写线程。

const int write_thread_num = 16;

class Poll_Server
{
     Poll_Server(uint16_t _port=8888, std::string _ip=_deafault)
                :port(_port)
                , ip(_ip)
            ,read_tp(threadpool::getinstance())
        {
                }    
    //..
     void start(bool isdaemon=false)
        {
         //...
           for(int i=0;i<write_thread_num;i++)
        {
            write_pollmangers.push_back(new Pollmanager(2));
            std::thread(write_thread_func,write_pollmangers.back(),&read_pollmanger).detach();
        }
       //....  
     }
  private:
        sock listensock;
        uint16_t port;
        std::string ip;
        Pollmanager read_pollmanger{1};
        std::vector<Pollmanager*> write_pollmangers;
        threadpool& read_tp;
}

其中，Poll_Server 内部还包含一个读线程池对象。该读线程池采用单例模式实现，因此主线程在创建写线程之前，会首先调用线程池的 start 函数完成初始化，即创建一批读线程。

---

接下来来看 Distribute 函数。

Distribute 函数接收一个 fd 以及读集合指针。在函数内部会创建一个任务对象，然后将该任务对象 push 到环形缓冲区中。

需要注意的是，这里的任务对象不仅仅包含一个 fd，同时还包含：

• 对应的读集合
• 对应写线程的写集合

class Task
{
public:
    Task()
        :socketfd(-1)
        ,read_manager(nullptr)
        ,write_manager(nullptr)
    {

    }
    //...
    private:
    int socketfd;
    static std::unordered_map<std::string, std::string> map;
    Pollmanager* read_manager;
    Pollmanager* write_manager;
  //...  
};
    std::unordered_map<std::string, std::string> Task::map = {
    {".html","text/html"},
    {".css","text/css"},
    {".png","image/png"},
    {".jpg","image/jpeg"}
};

当读线程从环形缓冲区中取走该任务对象并完成响应构建之后，只需要将 fd 添加到对应的写集合中，即可通知对应的写线程。随后写线程在检测到该 fd 写事件就绪时，就能够定位到对应连接的发送队列，并取出响应报文进行发送。

 void Distribute(int fd,Pollmanager* read_arg)
        {
                   read_arg->del_fd(fd);
                   Task t(fd,read_arg,write_pollmangers[fd%write_thread_num],last_active_time);
                   read_tp.push(t);
        }

这里需要特别注意：在 Distribute 函数内部，在将任务对象 push 到环形缓冲区之前，首先从读集合中删除了该 fd。

之所以要这样设计，是因为在当前的架构中，读线程的职责是：只处理一个完整的 HTTP 请求。也就是说，一个读线程会执行如下流程：

1. 读取完整的 HTTP 请求
2. 解析 HTTP 请求
3. 构建响应报文
4. 将响应报文放入发送队列
5. 通知写线程发送
6. 线程回到线程池

需要注意的是，所谓读事件就绪，本质上只是表示接收缓冲区不为空。但对于 TCP 连接来说，由于 TCP 是面向字节流（byte stream）的协议，对方一次发送的数据\可能只是一个不完整的 HTTP 请求报文。

而读线程必须在读取到完整的 HTTP 请求报文之后，才能继续执行后续流程，例如解析请求并构建响应。

因此在某些情况下，读线程可能需要等待客户端继续发送剩余的数据。

如果此时不将 fd 从读集合中删除，那么当客户端发送剩余数据、内核将其写入接收缓冲区后，如果主线程再次被 CPU 调度并检测到该 fd 的读事件就绪，那么主线程可能会再次分配一个新的读线程来处理该 fd。

虽然 TCP 是全双工通信，读和写确实可以并发执行（因为它们分别访问不同的缓冲区），但在当前场景下，如果多个读线程同时读取同一个连接的接收缓冲区，就会出现问题：

• 新分配的读线程可能会先读取部分数据
• 原来的读线程就无法再拼接出完整的 HTTP 请求报文

从而导致数据解析错误，甚至出现协议层面的不一致。

因此，为了避免这种数据竞争问题，需要保证：一个连接的读缓冲区在同一时刻只能被一个读线程访问。

也正因为如此，当主线程将某个 fd 的读事件分配给一个读线程之后，就必须暂时从读集合中删除该 fd，使主线程不再继续检测该 fd 的读事件。

当读线程完成处理并准备回到线程池之前，如果连接仍然处于活跃状态，则会重新将该 fd 加回读集合。这样当客户端后续继续发送请求数据时，主线程仍然可以检测到新的读事件。

当然，这里重新加入读集合的条件远不止如此简单。在后续的实现中，还需要根据不同的连接状态和多种复杂场景来决定是否将该 fd 重新加入读集合，这些内容将在后文进一步进行详细分析。

发送队列的线程安全

所以根据上文可知，这里需要创建一个全局哈希表。该哈希表中的每一个节点对应一个 Connection 对象，而每一个 fd 都对应一个 Connection 对象。Connection 对象内部包含一个发送队列。

因此，对于线程池中的读线程来说，其主要执行流程如下：

1. 从 TCP 套接字的接收缓冲区中读取完整的 HTTP 请求报文
2. 解析 HTTP 请求
3. 构建对应的 HTTP 响应报文
4. 将响应报文 push 到该连接对应的发送队列中

具体实现上，就是根据 fd 查询哈希表定位到对应的 Connection 对象，然后访问 Connection 对象中的发送队列字段，将构建好的响应报文加入队列。

而对于写线程来说，当其通过 poll 检测到某个 fd 的写事件就绪之后，会根据该 fd 定位到对应的 Connection 对象，然后访问该对象中的发送队列。写线程的操作是从队列中 pop（弹出）队首元素，随后调用 send 接口将响应报文发送给客户端。

我们知道，读线程与写线程是独立运行的执行单元。从逻辑上来说：

• 读线程负责读取 HTTP 请求、解析请求并构建响应
• 写线程负责从发送队列中取出响应并调用 send 发送

这两个行为是可以并发执行的。原因在于 TCP 是全双工通信协议，读线程与写线程访问的是 TCP 套接字的不同缓冲区：

• 读线程访问 接收缓冲区（receive buffer）
• 写线程访问 发送缓冲区（send buffer）

因此在 TCP 层面，读与写之间并不会产生冲突。

但是需要注意的是，对于这里设计的发送队列而言，读线程与写线程会对其进行并发访问：

• 读线程会执行 push 操作
• 写线程会执行 pop 操作

如果发送队列的底层实现是一个双向链表，那么在并发环境下就可能出现数据不一致的问题。

例如，假设当前队列中只有一个节点，读线程准备执行 push 操作。此时 push 的过程通常需要修改多个指针，例如：

• 修改**哨兵节点（sentinel node）**的前驱指针
• 修改原头节点的后继指针
• 将新节点插入链表

然而这些指针修改操作并不是原子的。

假设读线程在执行 push 时，已经完成了哨兵节点前驱指针的修改，但还没有完成后续操作时，线程发生了上下文切换，CPU 切换到写线程执行。

此时写线程执行 pop 操作，并修改了哨兵节点的后继指针，从而删除当前头节点。写线程执行完毕后再次切换回读线程，而读线程继续执行之前未完成的指针修改操作，例如修改头节点的后继指针。

但此时该头节点已经被写线程删除，读线程继续访问该节点就可能导致非法内存访问（use-after-free），从而引发严重的内存错误。

因此可以看到，当读线程与写线程并发访问发送队列时，如果缺乏同步机制，就会产生**数据竞争（data race）**以及数据结构的不一致问题。

解决这一问题的方式也很直接：必须对发送队列的访问进行同步保护，最简单可靠的方法就是加锁。

这里锁的粒度就是保护发送队列本身。因此，在 Connection 对象中除了包含发送队列之外，还需要包含一把互斥锁，用于保护发送队列的并发访问。

也就是说，读线程与写线程在执行 push 或 pop 操作之前，都需要先竞争该互斥锁。只有在成功获取锁之后，才能对发送队列进行修改操作。

示例结构如下：

class Connection
{
 public:
 //...
 std::queue<std::string> out_buffer;
 std::mutex mtx;
 //...
};

在这种设计下：

• 读线程在 push 响应报文之前需要先加锁
• 写线程在 pop 响应报文之前同样需要先加锁

从而保证发送队列在任何时刻只会被一个线程修改，避免并发访问带来的数据不一致问题。

读线程整体架构

认识到读写分离之后，接下来我们就重点关注读线程的设计。读线程实际上就是线程池中的工作线程。当读线程从环形缓冲区中获取到 Task 任务对象之后，会调用 Task 的 run 函数执行相应的业务逻辑，例如：读取 HTTP 请求、解析 HTTP 请求以及构建响应报文等。

首先需要说明的一点是，在设计读线程模型时，很多读者可能会产生这样一种直觉：既然已经实现了读写分离，那么对于读线程来说，其任务似乎应该是读取一个连接上的所有 HTTP 请求报文，然后依次解析这些请求，并将构建好的响应报文依次 push 到发送队列中。

从功能正确性的角度来看，这种设计确实是可行的。特别是在 TCP 长连接（Keep-Alive） 场景下，客户端可能会在同一条连接上连续发送多个 HTTP 请求报文，因此读线程似乎理应负责依次读取并处理所有请求。

然而，在高并发服务器的设计场景中，我们需要进一步关注这种设计在效率层面所带来的问题。

按照上述设计，也就是由一个读线程负责读取并处理一个连接上的所有请求报文。但需要注意的是，对于客户端来说，其并不一定会一次性将所有请求报文连续发送给服务器。换句话说，读线程并不一定会遇到一种“理想情况”：即 TCP 接收缓冲区中已经包含了客户端发送的所有请求报文数据。

实际上，更常见的情况是：客户端在调用 send 发送请求之前，本身也需要执行一定的业务逻辑。因此客户端往往是分批发送数据，而不是一次性发送所有请求报文。

这就会带来一个问题：读线程可能已经读取并处理了当前 TCP 接收缓冲区中包含的所有完整请求报文，但连接本身并没有关闭，例如 HTTP 头部中的 Connection: keep-alive 字段表明该连接仍然保持长连接状态。这意味着客户端在未来仍然可能继续发送新的请求报文。

既然客户端还会继续发送请求，那么对于当前读线程来说，如果继续等待新的请求报文到达，其唯一的选择就是阻塞等待。因为此时既无法读取到新的完整 HTTP 请求，也无法继续执行后续业务逻辑。

而我们知道，在高并发服务器设计中，提高性能的关键之一就是减少线程在 IO 等待阶段的时间。导致这一问题的根本原因在于：我们习惯性地将读线程绑定到某一个连接的生命周期上。

因此，在这里的设计中，读线程不会伴随一个连接的整个生命周期，而是只负责处理一次就绪的读事件。

具体流程如下：

1. 主线程通过 poll 检测到某个 fd 的读事件就绪
2. 将该 fd 封装为任务对象并放入环形缓冲区
3. 读线程从环形缓冲区中获取任务对象
4. 从 TCP 接收缓冲区中读取数据
5. 尝试解析完整的 HTTP 请求
6. 构建响应报文并放入发送队列
7. 将 fd 加入对应写线程的写集合，通知写线程发送响应

如果该连接是长连接（keep-alive），意味着客户端未来仍然可能继续发送请求报文。那么在读线程完成当前请求处理之后，就需要将该 fd 重新加入读集合，使主线程能够继续检测该 fd 的读事件。当新的数据到达时，主线程会再次分配一个新的读线程来处理，从而形成一种接力式的读事件处理机制。

在这种设计下，读线程只负责处理已经就绪的读事件。这意味着 TCP 接收缓冲区此时至少包含部分数据，从而避免读线程因为没有数据而陷入阻塞等待。

---

然而这里又会引出一个新的问题。

虽然读线程只负责读取一个完整的 HTTP 请求，但 TCP 是**面向字节流（byte stream）*的协议。对方发送的数据并不一定刚好对应一个完整的 HTTP 请求报文，很可能只是*部分请求数据。

那么问题就变成：如果读线程在当前时刻无法读取到完整的 HTTP 请求报文，是否应该阻塞等待客户端继续发送数据？

答案显然是否定的。正如前文所述，读线程不再绑定于某一个连接，它只是一个专门负责读取、解析以及构建响应的执行单元。可以将其理解为流水线上的工人：当当前流水线暂时没有足够的数据时，该工人应当立刻返回线程池，去处理其他已经就绪的任务。

但这里的关键在于：当前读线程已经读取了一部分 HTTP 请求数据。即使读线程在返回线程池之前将 fd 重新加入读集合，由新的读线程继续接力读取，新的读线程从 TCP 接收缓冲区中读取到的仍然可能是不完整的请求数据。

由于每个线程的线程栈是相互独立的，因此必须设计一种全局共享的数据结构来保存已经读取到但尚未解析完成的请求数据。

为了解决这一问题，这里在 Connection 对象中引入了一个接收缓冲结构 in_buffer。其中队列中的每一个节点表示一个正在构建的 HTTP 请求数据片段。

之所以选择队列结构，是因为 HTTP 请求通常需要按照到达顺序进行处理。因此队首节点代表最早接收到的请求数据，读线程在处理请求时，只需要优先处理队首节点即可。

in_buffer 从逻辑上看虽然使用的是 queue 容器，但其实际语义更接近于一个单节点的滚动缓冲区（rolling buffer），而不是多个并列节点同时存在的普通队列结构。

具体来说，in_buffer 的队首节点始终代表当前正在拼接的 HTTP 请求数据。读线程每次从 TCP 接收缓冲区读取到新的字节流后，都会将这些数据追加到队首节点的末尾，从而逐步累积形成完整的 HTTP 请求报文。

只有当一个 HTTP 请求被成功解析并处理完成之后，才会将该请求对应的节点从队列中 pop 掉，此时新的队首节点才开始用于累积下一个 HTTP 请求的数据。

因此，在绝大多数时间里，in_buffer 中实际上只会存在一个正在增长的字符串节点，用于持续拼接当前请求的数据。

class Connection
{
 public:
 //...
 std::queue<std::string>in_buffer;   
 std::queue<std::string> out_buffer;
 std::mutex mtx;
 //...
};

因此，对于读线程来说，其会调用 recv 接口，从 TCP 接收缓冲区中读取数据，并将读取到的字节流追加到 in_buffer 队首节点的末尾。

这里队列中的元素类型使用 std::string，主要原因在于 HTTP 协议本身是一种文本协议，因此使用字符串结构进行拼接与解析会更加方便。

随后读线程会对当前队首的字符串进行解析。由于 HTTP 请求行以及请求头中的每一项都以 CRLF（\r\n） 作为结束符，并且请求头与请求体之间存在一个空行（\r\n\r\n），因此可以通过调用字符串的 find 函数来查找当前字符串中是否已经出现该空行分隔符。

如果找到了该空行，则说明已经成功读取到了完整的请求头。

在 HTTP 协议中，请求大致可以分为两类：

• GET 请求：通常用于获取某种资源，该资源通常对应服务器文件系统中的某个文件路径，其路径来源于请求行中的 URL。
• POST 请求：通常用于提交数据或访问动态资源，请求体中会携带参数，而请求行中的 URL 通常是一个逻辑路径，在服务器端可能映射为某个处理函数或可执行程序。

如果当前字符串中仍然没有找到空行分隔符，则说明当前读线程读取到的数据（包括之前读线程读取到的数据）仍然不足以构成完整的请求头。此时读线程不会继续阻塞等待，而是直接返回线程池。在返回之前，需要将该 fd 重新加入读集合，等待新的数据到达后由其他读线程继续接力处理。

---

为了实现这种读事件的接力处理机制，系统必须维护一种状态延续机制。因为当一个读线程开始处理某个 fd 时，在此之前可能已经有其他读线程对该连接执行过部分读取和解析操作。

也就是说，虽然不同读线程是独立执行的，但它们处理的是同一个连接的连续状态。

例如，某个读线程在抽干 TCP 接收缓冲区之后，可能已经在 in_buffer 队首节点中拼接出了完整的请求行与请求头。此时该读线程会将请求头部分从数据流中分离出来，并解析请求头中的 Content-Length 字段，以确定请求体的长度。

随后读线程需要判断当前 in_buffer 中是否已经包含了完整的请求体。如果当前只有部分请求体数据，则该读线程同样不会阻塞等待，而是返回线程池，并等待下一个读线程继续接力处理。

因此，后续接力的读线程必须能够感知当前解析状态。例如：

• 是否已经解析出完整的请求头
• 是否已经进入请求体读取阶段

如果请求头已经解析完成，那么新的读线程就不需要重新解析请求头，而是可以直接尝试读取剩余的请求体数据。

为了实现这种状态延续机制，这里在 Connection 对象中引入了一个解析状态字段，用于记录当前 HTTP 请求解析的阶段。

当前设计中解析状态可以分为两个阶段：

• 解析请求头阶段（PARSE_HEADER）
• 解析请求体阶段（PARSE_BODY）

此外，Connection 对象还会维护以下字段：

• header：保存已经解析出的完整请求头
• header_length：表示请求行与请求头的总长度
• body_length：表示请求体长度

示例结构如下：

enum ParseState 
{
    PARSE_HEDER,
    PARSE_BODY
};

class Connection
{
   public:
Connection()
:parse_state(PARSE_HEDER)
,header_length(0)
,body_length(0)
,header("")
 //....   
{

}
   std::queue<std::string> in_buffer;
   std::queue<std::string> out_buffer;
   std::mutex mtx;
   //...
   ParseState parse_state;
   size_t header_length;
   size_t body_length;
   std::string header;
};

通过这种设计，不同读线程在处理同一个连接时，就能够基于 Connection 中记录的状态信息继续执行解析流程，从而实现一种跨线程的请求解析接力机制。

---

写线程

那么在讲解读线程实现的具体细节之前，我们先将视角切换到写线程。因为读线程的某些实现细节，需要结合写线程的设计以及整体架构才能理解清楚。

根据上文的分析，我们已经了解了读线程的整体设计框架，接下来将重点分析写线程的整体框架。

对于写线程而言，其核心职责是负责将已经构建好的 HTTP 响应报文发送给客户端。根据之前的设计，写线程会调用 poll 接口检测写事件集合中的文件描述符是否就绪。一旦某个 fd 的写事件就绪，则说明该 TCP 套接字的发送缓冲区当前未满，此时写线程就可以尝试发送数据。

因此写线程的大致流程为：

1. 调用 poll 检测写集合中的 fd
2. 找到写事件就绪的 fd
3. 根据 fd 定位对应的 Connection 对象
4. 访问 Connection 中的发送队列
5. 弹出队首响应报文
6. 调用 send 发送数据

以上便是写线程的整体框架。接下来我们进一步深入写线程的具体实现细节。

---

写线程的核心逻辑被封装在一个 while 死循环中。在线程循环开始时，首先需要调用 poll 来检测写集合中的 fd 是否就绪。

然而需要注意的是，写集合会被读线程和写线程并发访问。如果写线程在调用 poll 时直接持有写集合的锁，那么一旦 poll 阻塞等待事件，就会导致该锁被长时间持有，从而阻塞读线程对写集合的修改操作。这会导致线程竞争加剧，甚至可能出现线程饥饿，从而降低整体系统吞吐量。

因此，这里采用一种常见的并发设计策略：快照（snapshot）机制。

具体做法是：

写线程首先在短暂加锁的情况下，获取当前写集合 vector<pollfd> 的副本，随后立即释放锁。之后写线程将这个副本作为参数传递给 poll 接口进行事件检测。

这样一来，写线程在 poll 阻塞期间不会持有锁，从而避免影响读线程对写集合的更新。

需要额外注意的是，在服务器刚启动时，主线程通常会先创建写线程，然后才开始处理新的连接并检测读事件。因此在初始阶段，很可能还没有任何连接被加入写集合，此时写集合是空的。

而 poll 不能接收空数组，否则可能导致未定义行为。因此在调用 poll 之前，需要先判断当前 vector 是否为空。

如果写集合为空，则说明服务器刚启动，还没有需要发送数据的连接。此时写线程无需忙等，可以主动让出 CPU，例如通过 usleep 进行短暂休眠，以避免线程空转。

如果写集合不为空，则可以将快照传递给 poll。

另外还需要注意 poll 的超时时间设置。由于这里传递给 poll 的是某一时刻的快照，而不是实时的写集合状态，因此当前时刻可能已经有新的连接加入写集合。如果 poll 的超时时间设置过长，那么写线程将无法及时检测到新加入的 fd，从而影响吞吐量。

因此通常建议将超时时间设置得较短，例如 10ms，这样可以在较短时间内重新获取新的快照并扩大检测范围。

示例代码如下：

static void write_thread_func(Pollmanager* write_arg, Pollmanager* read_arg)
{
    while (1)
    {
        std::vector<pollfd> poll_fds = write_arg->get_snapshot();

        if (poll_fds.empty())
        {
            usleep(10000);
            continue;
        }

        int n = poll(poll_fds.data(), poll_fds.size(), 10);

        if (n < 0)
        {
            lg.logmessage(Fatal, "Poll error");
            break;
        }
        else if (n == 0)
        {
            lg.logmessage(info, "write_thread poll timeout");
            continue;
        }

        // ...
    }
}

---

当 poll 返回值大于 0 时，说明至少有一个 fd 的写事件已经就绪。此时写线程需要遍历 poll 返回的数组，检查哪些 fd 的写事件触发。

需要注意的是，写事件就绪仅表示 TCP 发送缓冲区当前未满，并不代表此时一定有数据可以发送。例如读线程仍在处理 HTTP 请求或者对方尚未发送新的 HTTP 请求导致发送队列当前为空

因此，即使某个 fd 写事件就绪，也必须进一步检查该 Connection 对象中的发送队列是否为空。

而发送队列会被读线程和写线程并发访问，因此在访问发送队列之前必须进行加锁，以保证队列结构的一致性。

写线程加锁之后：

• 如果发送队列为空，则直接解锁并跳过该 fd
• 如果发送队列不为空，则继续处理发送逻辑

---

接下来写线程会尝试清空发送队列。队首节点通常对应一个完整的 HTTP 响应报文。

我之所以说“尝试”，是因为对于写线程来说，其并不一定能够将发送队列中的所有数据一次性全部 send 出去。虽然此时该 fd 的写事件已经就绪，这意味着当前 TCP 发送缓冲区尚未写满，但这并不代表写线程能够立即清空整个发送队列。

当写线程检测到发送队列不为空时，其目标是尽可能清空发送队列，即依次将发送队列中的所有节点发送给客户端。

对于第一次 send 调用，一般都会成功返回。这里所谓的“成功”，指的是 send 调用能够立即返回，并成功写入一定数量的字节到 TCP 发送缓冲区。但需要注意的是，send 返回成功并不意味着这些数据已经真正通过网络发送到对端主机。

这是因为 send 的语义仅仅是将用户态缓冲区的数据拷贝到内核中的 TCP 发送缓冲区。至于这些数据何时真正发送到网络中，则由 TCP 协议栈的发送策略决定，例如：

• 滑动窗口（Sliding Window）
• Nagle 算法
• 拥塞控制等机制

因此，通过 send 写入的数据通常会暂存在 TCP 的发送缓冲区中等待后续发送。当写线程继续调用 send 发送更多数据时，新写入的数据会继续占用 TCP 发送缓冲区的空间。

一旦 TCP 发送缓冲区被填满，那么后续的 send 调用在默认情况下就会进入阻塞等待状态，即等待内核腾出新的缓冲区空间之后再继续写入数据。

如果写线程在此处发生阻塞，那么该写线程就会被动挂起并进入 IO 等待阶段。而根据上文的设计目标，我们希望尽可能减少线程在 IO 上的等待时间，以提高服务器整体的并发处理能力和吞吐量。

因此，这里需要将 send 设置为非阻塞调用，而不是使用默认的阻塞行为。一种常见的做法是在调用 send 时添加 MSG_DONTWAIT 标志位，使 send 在条件不满足时不会阻塞线程。

在这种情况下，当 TCP 发送缓冲区已满时，send 会立即返回 -1，并将 errno 设置为 EAGAIN 或 EWOULDBLOCK。此时写线程便可以停止当前连接的发送操作，转而去处理其他已经写事件就绪的 fd，从而避免写线程因单个连接而长时间阻塞。

因此写线程通常采用 “write until EAGAIN” 的策略：只要 send 还能继续写入数据，就持续发送；一旦 send 返回 EAGAIN，则立即停止当前连接的发送，并等待下一次写事件再次触发。

---

这里还需要注意一个重要问题。TCP 是一种**面向字节流（Byte Stream）**的传输协议，这意味着每一次调用 send 接口时，并不一定能够发送完整的 HTTP 响应报文。

根据上文的设计，这里的 send 是非阻塞调用。一旦 TCP 发送缓冲区被写满，send 就会立即返回 EAGAIN，此时写线程不会继续处理当前 fd，而是等待下一轮写事件再次触发后再继续发送。

这就意味着在当前这一轮发送过程中，发送队列很可能没有被完全清空。换句话说，一个 HTTP 响应报文可能只发送了一部分数据，剩余的数据仍然保留在发送队列的队首节点中。

因此，这里就需要一种发送状态延续机制。也就是说，当下一次写线程再次处理同一个 fd 的写事件时，必须能够知道上一次已经发送到了响应报文的哪个位置，从而避免重复发送已经发送过的数据。

为了解决这个问题，Connection 对象中需要维护一个写偏移量（write_offset），用于记录当前发送队列队首节点已经发送的字节数。

当写线程调用 send 成功发送 n 个字节后，就需要将 write_offset 增加 n，表示当前响应报文已经向前推进了 n 个字节。

当 write_offset 增长到队首节点数据长度时，就说明该 HTTP 响应报文已经全部发送完成。此时便可以将该节点从发送队列中 pop 掉，然后继续处理发送队列中的下一个节点。

如果 write_offset 尚未达到队首节点的末尾，则说明当前响应报文还未完全发送完成，那么该节点需要继续保留在队首位置。等到下一轮写事件再次触发时，写线程将会从 write_offset 所指示的位置继续发送剩余的数据。

因此，在 Connection 对象中需要维护如下状态信息：

class Connection
{
public:
    Connection()
        : write_offset(0)
         //....
    {}

    std::queue<std::string> out_buffer;
    std::mutex mtx;
    //...
    int write_offset;
};

此外，当发送队列的队首节点被弹出之后，需要将 write_offset 重新重置为 0，以便为下一条 HTTP 响应报文重新记录发送进度。

通过这种方式，即使一次 send 无法发送完整个响应报文，写线程也可以在后续的写事件中从上一次发送的位置继续发送剩余数据，从而保证响应报文能够被完整且正确地发送到客户端。

---

根据上文分析，我们已经知道 send 调用失败有一种常见情况是由于非阻塞发送导致的条件未就绪，即 TCP 发送缓冲区已经被写满，此时 send 会返回 -1，并将 errno 设置为 EAGAIN 或 EWOULDBLOCK。在这种情况下，写线程只需要停止当前发送操作，并等待下一次写事件再次触发即可。

然而，send 调用失败并不一定都是由于发送缓冲区已满造成的。在某些情况下，send 可能是由于真正的写入错误而失败。例如，当对端已经关闭连接，而写线程仍然尝试向该连接写入数据时，send 就会返回错误。

对于这种情况，写线程显然没有必要继续尝试清空发送队列，而应该直接关闭该连接，并释放与该连接相关的资源。具体而言，需要执行以下操作：

• 释放 Connection 对象
• 从哈希表中删除对应节点
• 关闭该连接对应的套接字

然而，这里会引入一个关键的并发问题。

根据前文的设计，读线程和写线程是并发执行的。其中：

• 读线程负责读取完整的 HTTP 请求、解析请求并构建响应
• 写线程负责从发送队列中取出响应并调用 send 发送

当读线程构建好 HTTP 响应报文之后，会将该响应报文加入对应 Connection 对象的发送队列。这个过程实际上包含两个步骤：

1. 通过 fd 作为键值，在哈希表中定位对应的 Connection 对象（例如使用 [] 操作）
2. 调用该 Connection 对象中的发送队列 push 函数，将响应报文加入队列

假设发生如下执行序列：

1. 读线程已经完成第一个步骤，即通过 fd 在哈希表中获取到了对应的 Connection 指针
2. 此时线程发生上下文切换
3. 写线程开始执行，并检测到同一个 fd 的写事件就绪
4. 写线程调用 send 发送数据，但不幸的是 send 调用失败（例如对端关闭连接）
5. 写线程决定关闭该连接，于是释放 Connection 对象，并从哈希表中删除对应节点
6. 随后线程再次切换回读线程
7. 读线程继续执行第二个步骤，即向 Connection 的发送队列中 push 数据

然而此时 Connection 对象已经被写线程释放，读线程仍然持有一个已经失效的指针，因此会导致非法内存访问（Use-After-Free）。

除此之外，还存在另一种潜在的并发问题。

我们知道，当主线程检测到监听套接字的读事件就绪后，会调用 accept 获取新的连接，并为该连接创建一个新的 Connection 对象，同时将其插入到哈希表中。

而 std::unordered_map 在插入节点时，如果当前负载因子超过阈值，可能会触发哈希表扩容（rehash）。扩容的过程通常包括：

1. 创建新的桶数组
2. 将原有节点重新分布到新的桶中
3. 释放旧的桶数组

假设发生如下情况：

1. 写线程正在执行删除操作，已经根据 fd 计算出了对应的哈希索引
2. 在其继续遍历链表之前发生线程切换
3. 主线程接受新连接并插入新的节点
4. 由于负载因子增加，unordered_map 触发扩容
5. 哈希表重新分配桶数组
6. 线程再次切换回写线程

此时写线程仍然基于旧的哈希结构继续访问数据结构，就有可能导致访问已经失效的内存区域，从而产生未定义行为。

因此，为了保证哈希表在并发访问下的安全性，需要对整个哈希表的访问进行同步保护。

具体做法是引入一把全局互斥锁，用于保护整个哈希表结构。当任何线程需要访问哈希表时，都必须首先获取这把锁，加锁成功后才能执行相应操作，包括：

• 查找 Connection
• 插入新节点
• 删除节点

示例代码如下：

std::mutex send_queues_mtx;

通过这种方式，可以保证在任意时刻只有一个线程能够修改或访问哈希表结构，从而避免并发访问导致的数据结构不一致以及非法内存访问问题。

---

然而，如果简单地在整个哈希表上设置一把全局锁，又会引入新的效率问题。

对于写线程而言，其主要职责是清空发送队列，这意味着在处理某个连接时，写线程可能需要多次调用 send 接口。而在发送过程中，写线程需要不断访问 Connection 对象中的发送队列。

在当前设计中，Connection 对象是通过 fd 作为键值存储在哈希表中的，因此每一次访问 Connection 对象，本质上都需要通过键值在哈希表中定位对应节点。

如果每次访问都需要获取全局哈希锁，那么写线程在处理发送队列时就会频繁竞争这把锁。一旦锁竞争失败，线程就会进入阻塞等待，从而延迟当前 fd 的发送过程，并进一步影响后续 fd 的写事件处理效率。

不过在当前架构设计中，有一个非常关键的前提：

Connection 对象的释放以及哈希表节点的删除，完全由写线程负责。

也就是说，在整个系统中：

• 读线程不会删除 Connection
• 主线程只负责插入新的 Connection
• 只有写线程会在连接生命周期结束时或者写入出错时执行资源回收

因此，对于写线程来说，在访问某个哈希表节点时，并不需要担心该节点会被其他线程删除。换句话说，写线程不需要考虑节点是否仍然存在的问题，因为只有写线程自身才会执行删除操作。

写线程真正需要关注的问题其实只有一个：

哈希表的结构是否发生了变化。

例如，当主线程接受新连接并向 unordered_map 插入新的节点时，如果触发了哈希表扩容（rehash），那么底层桶数组就会被重新分配。

需要注意的是，rehash 的过程并不会释放链表节点本身，而只是重新分配桶数组，并将节点重新挂载到新的桶中，随后释放旧的桶数组。

因此，对于写线程而言，如果在访问哈希表节点时发生扩容，就可能导致基于旧桶数组计算出的索引失效，从而产生未定义行为。

基于这一点，可以对锁的使用方式进行优化。

写线程在处理写事件时，可以先短暂获取一次全局哈希锁，仅用于根据 fd 获取对应的 Connection 指针。在成功获取指针之后，便可以立即释放全局锁。

由于 Connection 对象的生命周期由写线程自身控制，并且其他线程不会删除该对象，因此在后续访问该指针时无需再次加锁，也无需担心哈希表扩容带来的问题。

这种方式可以显著缩短全局锁的持有时间，从而减少锁竞争对系统性能的影响。

具体实现示例如下：

static void write_thread_func(Pollmanager* write_arg, Pollmanager* read_arg)
{
    while (1)
    {
        std::vector<pollfd> poll_fds = write_arg->get_snapshot();

        if (poll_fds.empty())
        {
            usleep(10000);
            continue;
        }

        int n = poll(poll_fds.data(), poll_fds.size(), 10);

        if (n < 0)
        {
            lg.logmessage(Fatal, "Poll error");
            break;
        }
        else if (n == 0)
        {
            lg.logmessage(info, "write_thread poll timeout");
            continue;
        }

        for (auto& i : poll_fds)
        {
            if (i.revents & POLLOUT)
            {
                send_queues_mtx.lock();
                Connection* conn = send_queues[i.fd];
                send_queues_mtx.unlock();

                conn->mtx.lock();

                bool eagain_occurred = false;

                while (!conn->out_buffer.empty())
                {
                    std::string data = conn->out_buffer.front();

                    ssize_t send_bytes =
                        send(i.fd,
                             data.c_str() + conn->write_offset,
                             data.size() - conn->write_offset,
                             MSG_DONTWAIT);

                    // ...
                }

                // ...
            }
        }
    }
}

通过这种方式，全局哈希锁仅用于获取 Connection 指针，而不会在整个发送过程中持续持有，从而在保证线程安全的同时降低锁竞争带来的性能影响。

---

根据上文的分析，至此我们已经能够明确写事件的处理结果。总体而言，写事件完成后的状态可以归纳为以下四种情况：

1. 写入错误
2. 发送队列已经清空，但连接生命周期尚未结束（后续仍可能收到新的请求报文）
3. 非阻塞调用 send 时发送缓冲区已满（EAGAIN），发送队列尚未清空
4. 发送队列已经清空，并且连接生命周期结束

因此，写事件处理完成后的状态本质上只会落入上述四种情况之一。接下来分别分析这四种情况对应的处理方式。

---

第一种情况：写入错误 and 第四种情况：发送队列清空且连接生命周期结束

这两种情况的处理逻辑本质上是相同的：该连接已经不再需要继续存在，因此需要执行连接资源的回收。

具体处理流程如下：

1. 先从读事件集合和写事件集合中删除该 fd
2. 竞争全局哈希表锁
3. 先释放 Connection 对象内部的锁（conn->mtx）

这里需要特别注意锁的释放顺序。如果在删除 Connection 对象之后再尝试释放 conn->mtx，就会导致对已经释放内存的访问，从而产生非法内存访问（use-after-free）。

因此必须先释放 Connection 锁，再执行对象销毁。

1. 删除 Connection 对象
2. 从哈希表中删除对应节点
3. 释放全局哈希锁
4. 最后关闭套接字

---

第二种情况：发送队列清空，但连接尚未关闭

当发送队列已经被完全清空，而连接生命周期仍然有效时，说明当前响应已经发送完成，但连接仍然保持（例如 HTTP Keep-Alive）。

在这种情况下：

• 写线程不能释放 Connection 对象
• 也不能删除哈希表节点

但是此时继续监听该 fd 的写事件已经没有意义。

这是因为写事件就绪的语义是：发送缓冲区未满，可以继续发送数据。然而当前发送队列已经为空，即使写事件再次就绪，也没有任何数据可以发送，因此线程会被不断唤醒却无法执行有效工作，从而形成空转（busy loop）。

因此需要将该 fd 从写事件集合中删除，仅保留读事件监听，以等待新的请求到达。

完成删除写事件集合操作后，只需要释放 Connection 锁即可。

---

第三种情况：发送缓冲区已满（EAGAIN）

如果 send 返回 EAGAIN，说明当前套接字的发送缓冲区已经满，但这并不是错误，而是非阻塞 I/O 的正常行为。

此时发送队列尚未清空，因此该连接仍然需要在后续写事件中继续发送剩余数据。

因此：

• 不需要删除写事件
• 不需要释放 Connection
• 只需要释放连接锁

当下一轮写事件再次就绪时，写线程会继续清空发送队列。

---

在具体实现中，由于上述几种情况都会导致退出 while 循环，因此代码中设置了两个标志位用于区分不同情况：

• error_occurred：用于标识写入错误
• eagain_occurred：用于标识发送缓冲区已满

对应实现代码如下：

static void write_thread_func(Pollmanager* write_arg,Pollmanager* read_arg)
{
    while(1){
        std::vector<pollfd> poll_fds=write_arg->get_snapshot();
        if(poll_fds.empty())
        {
            usleep(10000);
            continue;
        }
        int n=poll(poll_fds.data(),poll_fds.size(),10);
        if(n<0)
        {
            lg.logmessage(Fatal,"Poll error");
            break;
        }else if(n==0)
        {
            lg.logmessage(info,"write_thread poll timeout");
            continue;
        }
        for(auto& i:poll_fds)
        {
            if(i.revents&POLLOUT)
            {
                send_queues_mtx.lock();
                Connection* conn=send_queues[i.fd];
                send_queues_mtx.unlock();

                conn->mtx.lock();

                bool error_occurred=false;
                bool eagain_occurred=false;

                while(!conn->out_buffer.empty())
                {
                    std::string data=conn->out_buffer.front();

                    ssize_t send_bytes=
                        send(i.fd,
                             data.c_str()+conn->write_offset,
                             data.size()-conn->write_offset,
                             MSG_DONTWAIT);

                    if(send_bytes<0)
                    {
                        if(errno==EAGAIN)
                        {
                            eagain_occurred=true;
                            break;
                        }

                        lg.logmessage(Fatal,"send error");
                        error_occurred=true;
                        break;
                    }
                    else
                    {
                        conn->write_offset+=send_bytes;

                        if(conn->write_offset==data.size())
                        {
                            conn->write_offset=0;
                            conn->out_buffer.pop();
                        }
                    }
                }

                if(error_occurred||(conn->should_close&&conn->out_buffer.empty()))
                {
                    read_arg->del_fd(i.fd);
                    write_arg->del_fd(i.fd);

                    send_queues_mtx.lock();
                    conn->mtx.unlock();
                    delete conn;
                    send_queues.erase(i.fd);
                    send_queues_mtx.unlock();

                    close(i.fd);
                }
                else if(eagain_occurred)
                {
                    conn->mtx.unlock();
                }
                else if(conn->should_close==false&&conn->out_buffer.empty())
                {
                    write_arg->del_fd(i.fd);
                    conn->mtx.unlock();
                }
            }
        }
    }
}

最后还需要补充一个问题：写线程如何判断当前连接的生命周期是否已经结束。

这一点实际上与读线程的处理逻辑密切相关。

当读线程读取到完整的 HTTP 请求报文之后，会进入请求处理阶段。在处理请求的过程中，需要解析请求头中的一个关键字段，即 Connection 字段。该字段用于指示客户端对连接生命周期的期望。

通常存在两种情况：

1. Connection: keep-alive

   如果请求头中的 Connection 字段为 keep-alive，说明客户端希望保持当前 TCP 连接。在这种情况下，当前请求处理完成之后，连接仍然保持打开状态，后续还可能继续接收新的请求报文。

2. Connection: close

   如果请求头中的 Connection 字段为 close，则说明客户端希望在当前请求处理完成后关闭连接。因此，当前读线程读取到的请求报文就是该连接上的最后一个请求报文，后续不会再接收到新的请求。

基于这一语义，在 Connection 对象中引入了一个标志位 should_close。当读线程在解析 HTTP 请求头时，如果检测到 Connection 字段为 close，就会将该标志位设置为 true。

之后，当写线程完成响应发送时，就可以根据 should_close 标志位判断当前连接的最终处理方式：

• 如果 should_close == false，说明连接仍然需要保持，用于后续请求
• 如果 should_close == true，则在发送队列清空之后需要关闭连接并释放相关资源

也就是说，读线程负责决定连接生命周期，而写线程负责在发送完成后执行最终的连接回收。这种职责划分可以避免多线程同时操作连接生命周期所带来的并发问题。

对应的 Connection 数据结构定义如下：

enum ParseState 
{
    PARSE_HEDER,
    PARSE_BODY
};

class Connection
{
public:
    Connection()
        : write_offset(0),
          should_close(false),
          parse_state(PARSE_HEDER),
          header_length(0),
          body_length(0),
          header("")
    {
    }

    std::queue<std::string> in_buffer;
    std::queue<std::string> out_buffer;

    std::mutex mtx;

    int write_offset;

    bool should_close;

    ParseState parse_state;

    size_t header_length;
    size_t body_length;

    std::string header;
};

std::unordered_map<int, Connection*> send_queues;

至此，终于说明白了Connection对象的完全体

---

Connection对象生命周期管理：引用计数+延迟删除

至此，对于读线程而言，我们就可以进一步完善 Get_HttpRequest 函数的实现逻辑。Get_HttpRequest 函数的主要职责是：读取一个完整的 HTTP 请求报文并完成反序列化解析。

然而，根据上文的分析，我们已经知道写线程在某些情况下会执行“收尸”操作，也就是释放 Connection 对象并删除哈希表节点。因此，对于读线程来说，该函数后续流程必然需要访问 Connection 对象中的接收队列，而读线程与写线程是并发执行的。这是因为 TCP 本身是全双工通信，读事件和写事件可以同时发生。

在这种情况下，就可能出现如下并发场景：

1. 读线程正在访问哈希表，通过 fd 获取对应的 Connection 对象；
2. 线程发生切换，调度到写线程；
3. 写线程正在处理该 fd 的写事件，并且在写入过程中发生错误；
4. 写线程执行清理逻辑，删除哈希表节点并释放 Connection 对象；
5. 随后线程再次切换回读线程；
6. 读线程继续访问已经被释放的 Connection 对象，从而导致非法内存访问（use-after-free）。

---

有些读者可能会想到这样一种解决方案。

根据之前的设计，写线程在处理写事件时，其加锁流程大致如下：

1. 先竞争全局哈希锁；
2. 根据 fd 获取 Connection 指针；
3. 释放哈希锁；
4. 再获取 Connection 内部锁（小锁）；
5. 完成写事件处理。

而在执行清理操作时，写线程的锁操作顺序如下：

1. 先释放 Connection 小锁；
2. 再竞争全局哈希锁；
3. 删除哈希表节点并释放 Connection 对象；
4. 最后释放哈希锁。

这里需要特别说明一点：写线程在清理阶段必须先释放 Connection 小锁。如果在删除 Connection 对象之后再释放该锁，就会对已经释放的内存进行访问，从而导致未定义行为。

因此，整体加锁流程大致如下：

读线程：哈希锁 → 获取指针 → conn->mtx → 释放哈希锁 → 工作中 → 释放 conn->mtx
写线程：哈希锁 → 获取指针 → 释放哈希锁 → conn->mtx → 工作

虽然读线程和写线程都会竞争全局锁和连接锁，但这里不会产生死锁。这是因为死锁通常需要满足多个必要条件，其中一个关键条件是：

多个线程以相反顺序获取相同的两把锁。

而在当前设计中，写线程会主动释放其中一把锁，从而破坏死锁产生所需的条件，因此不会形成死锁。

---

基于上述分析，有些读者可能会提出一个改进思路：

读线程在访问 Connection 时，可以先获取全局哈希锁，然后在哈希锁保护范围内再获取 Connection 小锁，随后释放哈希锁。

这样一来，当写线程尝试处理写事件时，必须先获取 Connection 小锁，而该锁已经被读线程持有，因此写线程会被阻塞。

由于该锁只保护当前 Connection 对象，因此不会影响其他连接的处理，也不会阻塞其他读线程或写线程。

表面上看，这似乎是一种合理的方案。

---

然而，这种方案实际上仍然存在问题。

首先是效率问题。如果读线程持有 Connection 锁的时间较长，就会阻塞写线程对该连接的写事件处理。这样不仅会延迟当前 fd 的发送过程，还会影响写线程后续处理其他 fd 的写事件。

但更关键的问题其实是安全性问题。

需要重点关注写线程的清理阶段。考虑如下执行时序：

1. 读线程首先获取全局哈希锁；
2. 读线程通过 fd 获取 Connection 指针；
3. 读线程尝试获取 conn->mtx，但此时该锁被写线程持有；
4. 写线程正在执行清理逻辑；
5. 写线程释放 conn->mtx；
6. 写线程随后尝试获取哈希锁；
7. 此时读线程成功获取 conn->mtx，随后释放哈希锁；
8. 写线程获得哈希锁并删除 Connection 对象。

执行顺序可以表示为：

读线程：哈希锁 → 取指针 → conn->mtx → 释放哈希锁 → 工作中...
写线程：释放 conn->mtx → 等待哈希锁 → 拿到哈希锁 → delete conn → 释放哈希锁

在这种情况下，读线程已经持有 conn->mtx 并继续访问 Connection 对象，而写线程随后删除了该对象。这就会导致读线程访问已经被释放的内存，从而产生非法内存访问问题。

因此，仅仅通过调整锁的获取顺序，并不能彻底解决 Connection 生命周期与并发访问之间的冲突问题。

---

所以这里就需要引入 shared_ptr 智能指针。那么此时哈希表中的节点不再是裸指针，而是由 shared_ptr 进行管理。shared_ptr 内部维护一个引用计数，其拷贝构造本质上是“浅拷贝”，多个 shared_ptr 实例共同指向同一块内存区域；在拷贝时，会先对引用计数进行递增。而在析构时，会先对引用计数进行递减，只有当引用计数为 0 时，才会真正释放所管理的对象。

std::unordered_map<int,std::shared_ptr<Connection>> send_queues;

因此，对于写线程而言，不再需要手动释放 Connection 节点并删除哈希表节点，而是只需调用 erase 方法即可。该方法会触发对应元素的析构，从而减少 Connection 对象的引用计数。换言之，写线程只需要关注哈希表结构本身的一致性（即结构是否被修改），因此只需要对哈希表加锁即可。

从职责上看，写线程是“收尾/清理线程”，其不需要关心节点在其他线程中是否仍被使用。

---

而对于读线程的 Get_HttpRequest 函数，这里也不再尝试使用细粒度锁，而是直接使用全局大锁，原因有两点：

1. 保证哈希表结构的一致性，避免在扩容或删除过程中发生非法内存访问；
2. 确保节点存在性，防止在访问前该节点已被写线程删除。

具体流程是：先加大锁，通过 find 判断节点是否存在；若存在，则获取对应的 shared_ptr（这一步会增加引用计数）；随后立即释放大锁。

这里在大锁内获取 shared_ptr 的意义在于：延长 Connection 对象的生命周期。即使后续写线程删除了该节点，也只是减少引用计数，只要当前读线程仍持有 shared_ptr，对象就不会被释放。

---

接下来读线程会执行“抽干 TCP 接收缓冲区”的操作：循环调用 recv（非阻塞模式），直到返回 EAGAIN 为止。

• 若 recv < 0 且 errno ≠ EAGAIN → 返回 RECV_ERROR
• 若 recv == 0 → 表示对端关闭连接，返回 CLIENT_CLOSE
• 若 recv > 0 → 将数据追加到当前队首节点

---

随后进入 HTTP 解析阶段：

1. 解析请求头阶段（PARSE_HEADER）

首先查找 \r\n\r\n：

• 若不存在 → 说明请求头尚未接收完整，返回 RECV_TIMEOUT
• 若存在 → 说明请求头完整，可以进行解析

解析内容包括：

• 提取请求头字符串
• 记录 header_length
• 查找 Content-Length 字段（大小写兼容）
  • 若存在 → 设置 body_length
  • 若不存在 → 默认为 0（通常为 GET 请求）

然后状态切换为 PARSE_BODY

---

2. 解析请求体阶段（PARSE_BODY）

计算：

remaining = 已接收总长度 - header_length

分情况：

• 若 remaining < body_length
  → 请求体未接收完整，返回 RECV_TIMEOUT
• 若 remaining >= body_length
  → 请求体已完整：
  • 提取请求体
  • 若存在多余数据（粘包）：
    • 将剩余部分放入新的队列节点（用于下一个请求）
  • 反序列化为 Http_Request 对象
  • 弹出当前节点
  • 重置解析状态（支持 keep-alive 场景）

这里特别需要强调一点：TCP 是字节流协议，一次 recv 可能包含多个 HTTP 请求，因此必须正确处理“粘包”问题。

返回值语义

• RECV_TIMEOUT
  表示当前未获取完整 HTTP 请求，仅需重新加入读事件集合即可
• RECV_ERROR
  表示读取出错。此时读线程不能直接清理资源，而是：
  • 设置错误响应（如 502）
  • 标记 should_close = true
  • 交由写线程完成最终清理
• CLIENT_CLOSE
  表示对端关闭连接：
  • 标记 should_close = true
  • 转交写线程清理发送队列并关闭连接

---

4. 关于锁顺序（关键细节）

这里有一个非常关键的并发细节：锁的获取顺序必须统一，否则会发生死锁。

当前设计是：

• 读线程：
  • 先加大锁 → 获取 shared_ptr → 释放大锁 → 再加小锁
• 写线程：
  • 先持有小锁 → 再尝试加大锁（用于 erase）

如果读线程在持有大锁时再去获取小锁，就可能形成：

读线程：持有大锁 → 等待小锁  
写线程：持有小锁 → 等待大锁  
→ 死锁

因此这里采用“读线程先释放大锁再加小锁”的策略，避免锁顺序反转。

---

当然，也可以调整写线程的加锁顺序（例如先释放小锁再获取大锁），这样读线程就可以在大锁内部获取小锁。但当前方案的优势在于：

👉 借助 shared_ptr 的引用计数机制，无需依赖锁来保证对象生命周期安全，从而可以主动降低锁粒度，避免死锁风险。

读线程实现细节

那么，当获取到一个完整的 HTTP 请求之后，接下来就进入业务逻辑处理阶段。此时会根据 HTTP 请求的方法类型进行分发判断，即判断是 GET 请求还是 POST 请求，从而进入不同的业务处理流程，分别调用 Http_Get_Handler 或 Http_Post_Handler 函数。函数的返回值即为构造好的 HTTP 响应报文。

随后进入响应发送阶段。首先仍然需要加全局哈希锁，检查当前节点是否存在；若存在，则创建一个 shared_ptr 指向对应的 Connection 对象，从而增加引用计数，然后释放全局锁。接着加 Connection 对象的小锁，将响应报文放入发送队列 out_buffer 中。

之后需要根据请求头中的 Connection 字段判断连接生命周期：

• 若为 close，则说明当前请求为该连接的最后一个请求，后续不会再有新的请求报文，此时直接将 should_close 标志位置为 true；
• 若不为 close，则表示连接逻辑上仍然保持，但不意味着客户端一定还会继续发送请求。

---

这里需要重点关注一个典型场景：

客户端在一个 TCP 长连接中连续发送请求 A、B、C，随后不再发送任何请求，并主动关闭连接。而服务端在处理该 fd 的读事件时，会“抽干” TCP 接收缓冲区，将 A、B、C 三个请求全部追加到 in_buffer 队首节点中。

此时读线程解析出第一个完整请求 A，完成业务处理并生成响应，然后将该 fd 添加到读/写集合中，交由写线程发送响应。

但是此时客户端已经不会再发送新的请求报文，这会导致：

• 主线程不会再检测到该 fd 的读事件（因为没有新数据到达）
• Connection 对象持续存在（未触发释放条件）
• in_buffer 中残留的请求 B、C 无法被继续处理

最终可能导致**逻辑上的“内存滞留”（类似泄漏）**以及请求积压。

---

为避免上述问题，读线程在处理完一个请求后，不能仅仅将 fd 加入写集合，还需要主动检查接收队列状态：

• 若 Connection 未关闭（should_close == false），且 in_buffer 队首仍存在未处理数据（即还有后续请求），则：
  • 需要主动创建新的任务对象 Task
  • 将其投递到线程池（环形缓冲区）
  • 由其他读线程继续处理剩余请求

也就是说，这里从“事件驱动”转为“事件 + 主动调度”的混合模型，以避免长连接场景下的请求滞留。

---

但是这里存在一个关键的并发陷阱：

读线程当前是持有 Connection 小锁的，而任务投递（push 到线程池队列）是一个共享资源操作（读线程以及主线程都会push），会涉及锁竞争。如果在持锁状态下进行 push：

• 一旦线程池队列锁竞争失败，读线程会阻塞
• 而读线程此时持有 Connection 锁
• 写线程可能正等待该锁以处理发送事件

从而导致写线程被长时间阻塞，甚至影响当前以及其他 fd 的写事件处理

---

因此这里必须严格遵守：

👉 在 push 任务之前，必须释放 Connection 小锁

同时注意：

• 小锁的作用仅限于：
  • 向 out_buffer 写入响应
  • 修改 should_close 标志位
• 而任务投递发生在上述操作之后，因此提前释放小锁是安全的

此外，在释放锁之前，还需要确保：

• 已将 fd 加入写集合（保证响应可以被发送）

---

另一种情况是：

• Connection 未关闭
• 且 in_buffer 已为空

说明当前场景是典型的“请求-响应串行模型”：

客户端发送一个请求 → 等待响应 → 再发送下一个请求

此时无需主动调度任务，只需：

• 将 fd 同时加入读集合和写集合
• 等待后续事件驱动即可

void Task::run()
{
     Http_Request hr;
        int get_result = Get_HttpRequest(socketfd, hr);
        if (get_result != 0)
        {
            if(get_result == RECV_TIMEOUT)
            {
                 send_queues_mtx.lock();
                  if(send_queues.find(socketfd)==send_queues.end())
            {
                send_queues_mtx.unlock();
                return;
            }
              std::shared_ptr<Connection> conn=send_queues[socketfd];
            send_queues_mtx.unlock();
            conn->mtx.lock();
            read_manager->add_fd(socketfd);
            conn->mtx.unlock();
            last_active_time[socketfd]=time(NULL);
            return;
            }
            if(get_result == CLIENT_CLOSE)
            {
                send_queues_mtx.lock();
                if(send_queues.find(socketfd)==send_queues.end())
                {
                    send_queues_mtx.unlock();
                    return;
                }
                std::shared_ptr<Connection> conn=send_queues[socketfd];
                send_queues_mtx.unlock();
                conn->mtx.lock();
                read_manager->del_fd(socketfd);
                conn->should_close=true;
                write_manager->add_fd(socketfd);
                conn->mtx.unlock();
                return;

            }
            lg.logmessage(Fatal, "get http request error");
            send_queues_mtx.lock();
            if(send_queues.find(socketfd)==send_queues.end())
            {
                send_queues_mtx.unlock();
                return;
            }
        std::shared_ptr<Connection> conn=send_queues[socketfd];
           send_queues_mtx.unlock();
           conn->mtx.lock();
           hr.headers["Connection"]="Close";
           conn->out_buffer.push(process_bad_request(hr));
           conn->should_close=true;
           read_manager->del_fd(socketfd);
           write_manager->add_fd(socketfd);
           conn->mtx.unlock();
            return;
        }
        std::string res;
        if (hr.method == "GET")
        {
            res = Http_Get_Handler(hr);
        }
        else if (hr.method == "POST")
        {
            res = Http_Post_Handler(hr);
        }
        else
        {
            lg.logmessage(warning, "unsupported method:%s", hr.method.c_str());
            send_queues_mtx.lock();
            if(send_queues.find(socketfd)==send_queues.end())
            {
                send_queues_mtx.unlock();
                return;
            }
           std::shared_ptr<Connection> conn=send_queues[socketfd];
           send_queues_mtx.unlock();
           conn->mtx.lock();
           hr.headers["Connection"]="Close";
           conn->out_buffer.push(process_bad_request(hr));
           conn->should_close=true;
           write_manager->add_fd(socketfd);
           conn->mtx.unlock();
            return;
        }
         send_queues_mtx.lock();
         if(send_queues.find(socketfd)==send_queues.end())
            {
                send_queues_mtx.unlock();
                return;
            }
            std::shared_ptr<Connection> conn=send_queues[socketfd];
            send_queues_mtx.unlock();
            conn->mtx.lock();
             conn->out_buffer.push(res);
        if(hr.headers["Connection"]=="close"||hr.headers["Connection"]=="Close")
        {
              conn->should_close=true;
        }else
        {
            if(!conn->in_buffer.empty()&&!conn->in_buffer.front().empty())
            {
                write_manager->add_fd(socketfd);
                conn->mtx.unlock();
                Task t(socketfd,read_manager,write_manager,last_active_time);
                last_active_time[socketfd]=time(NULL);
                threadpool::getinstance().push(t);
                return;
            }else
            {
                read_manager->add_fd(socketfd);
            }
        }
        write_manager->add_fd(socketfd);
        conn->mtx.unlock();
        last_active_time[socketfd]=time(NULL);
}

超时检测线程

而这里需要注意的是，由于当前场景是高并发服务器，服务器在任意时刻都会接收到大量连接以及对应的请求报文。因此，对于客户端而言，其行为可能是请求-响应的串行模型，而不是一次性连续发送多个 HTTP 请求。

也就是说，客户端在收到响应之后，往往需要较长时间来完成自身的业务逻辑处理。进一步考虑极端情况，客户端甚至可能是一个恶意客户端：其在接收到服务端返回的响应报文后，不再发送后续请求，但连接依然保持不关闭状态。在这种情况下，对于主线程而言，将无法再检测到该 fd 对应的读事件就绪，从而导致该文件描述符以及其对应的 Connection 对象无法被及时清理。因此，这里就引出了超时检测线程存在的必要性。

那么在实现上，我们会在 Poll_Server 对象中定义一个静态数组，长度为 65535，用于覆盖进程文件描述符表的上限。数组的下标直接对应 fd。每当读线程读取到完整的 HTTP 请求，完成解析并构建响应，随后将响应放入发送队列之后，还会同步更新该 fd 对应的时间戳。

而对于超时检测线程，其核心流程如下：首先获取当前时间戳，然后获取读集合的一个快照（snapshot）。之所以采用快照，是因为超时线程会与主线程以及读线程并发访问读集合，为了避免加锁带来的性能开销，这里会短暂持有锁。随后遍历该快照数组，筛选出需要检测读事件的 fd，并访问时间戳数组，通过“当前时间 - 最后活跃时间”计算时间间隔，再判断该值是否超过设定阈值（例如 60 秒）。之所以选择 60 秒，是基于通常情况下客户端处理速度与网络传输延迟都较低这一经验前提。

如果该时间间隔大于 60 秒，则认为该连接已经超时。此时将该 fd 从读集合中删除。由于写线程负责执行 Connection 对象的最终清理工作，因此这里的处理流程如下：首先加全局大锁，查询对应的 Connection 对象是否仍然存在；若存在，则构造一个 shared_ptr 指向该 Connection 对象以增加引用计数，随后释放大锁；接着加细粒度的小锁，修改其关闭标志位 should_close，再释放小锁；最后将该 fd 加入对应的写集合，由写线程完成后续资源回收与关闭操作。

  static void check_timeout_fun(Poll_Server* server)
        {
            while(1)
            {
                time_t now=time(NULL);
                std::vector<pollfd> poll_fds=server->read_pollmanger.get_snapshot();
                for(auto& read_active:poll_fds)
                {
                    if(read_active.fd!=server->listensock.fd())
                    {
                         if(now-server->last_active_time[read_active.fd]>60)
                         {
                            if(server->read_pollmanger.del_fd(read_active.fd)){
                            lg.logmessage(info,"client fd:%d timeout",read_active.fd);
                            send_queues_mtx.lock();
                            if(send_queues.find(read_active.fd)==send_queues.end())
                            {
                                send_queues_mtx.unlock();
                                continue;
                            }
                            std::shared_ptr<Connection> conn=send_queues[read_active.fd];
                            send_queues_mtx.unlock();
                            conn->mtx.lock();
                            conn->should_close=true;
                            conn->mtx.unlock();
                            server->write_pollmangers[read_active.fd%write_thread_num]->add_fd(read_active.fd);
                         }
                    }
                }
                }
                sleep(30);
        }
        }
       

此外需要注意，超时检测线程不能持续高频运行。由于每次检测都需要进行一次 O(N) 级别的遍历，如果频繁执行，会对 CPU 造成较大压力。因此，在每一轮遍历结束后，线程会主动让出 CPU，并进入休眠状态（例如 sleep 30 秒），以降低系统整体负载。这本质上是一种时间换空间、吞吐与延迟之间的权衡策略。

这里还需要补充一个关键细节，即读线程更新时间戳的时机问题。有读者可能会担心：当前实现中，时间戳是在 run 函数末尾才更新的，也就是在“读取 HTTP 请求 → 解析请求 → 构建响应 → 放入发送队列”这一整套流程全部完成之后才更新。如果中间业务处理耗时较长，或者由于锁竞争导致线程阻塞，那么在这段时间内，超时检测线程可能会误判该连接已超时，从而错误地关闭连接。

但这里的设计恰恰规避了这一问题。关键点在于：主线程在将读事件分发给读线程之前，会先将该 fd 从读集合中移除。从表面上看，这一操作的目的在于保证读线程对该 fd 的独占访问，避免多个读线程并发读取 TCP 接收缓冲区。但实际上，这一机制同时也“巧妙地”避免了超时误判问题。

原因在于：超时检测线程只会检测当前仍在读集合中的 fd。而一旦某个 fd 被主线程分配给读线程处理，它就已经从读集合中移除。因此，在读线程执行期间，该 fd 不会出现在超时检测线程的检测范围内。也就是说，即使业务处理时间较长，也不会被误判为超时。

进一步分析一个竞态场景：假设某一时刻，超时线程获取了读集合的快照，其中包含 fd5；随后，在超时线程尚未完成检测之前，主线程检测到 fd5 读事件就绪，并将其从读集合中移除，同时分配读线程处理。在这种情况下，可能存在一个极短的时间窗口，使得超时线程仍然基于旧快照判断 fd5 已超时。

例如，存在这样一种时序场景：主线程在第 59 秒时检测到某个 fd 的读事件就绪，并将其分配给读线程进行处理；而超时检测线程在第 60 秒执行检测时，基于时间戳判断该 fd 已经超时，从而将其标记并通知对应的写线程执行清理操作。

需要注意的是，这种情况本质上属于**竞态条件（race condition）**的一种表现。因此，在执行超时清理逻辑之前，必须再次校验该 fd 当前是否仍然处于读集合中；如果该 fd 已被主线程移除并交由读线程处理，则不应执行超时关闭操作，以避免误杀正在处理中的连接。

因此，这里必须增加一道额外的校验：在执行超时清理前，必须再次确认该 fd 是否仍然存在于读集合中。如果该 fd 已经被移除，则说明其已经进入读线程处理流程，不应进行超时清理；只有当 fd 仍然存在于读集合中且满足超时条件时，才执行连接关闭逻辑。

所以无论是哪种场景：

场景 A：主线程抢先。
客户端发送新数据后，主线程在 poll 返回后立即执行 read_manager->del_fd(fd)，并将该 fd 分配给读线程处理。此时，若超时检测线程尝试执行 del_fd，则会失败（返回 false）。该返回结果隐含地表明：当前连接已经被主线程接管，正处于活跃处理状态。因此，超时线程会据此放弃清理操作。

场景 B：超时线程抢先。
当连接确实在 60 秒内无任何活动时，超时检测线程会先一步成功执行 del_fd（返回 true）。此时，即使在下一瞬间主线程的 poll 返回，由于该 fd 已经从读集合中移除，主线程也不会再感知到该 fd 的读事件，从而不会对其进行处理。

上述两种场景共同构成了一种天然的互斥机制。本质上，这里利用了 Pollmanager 内部的那把锁和 vector/unordered_map 的操作结果，完成了一次无须额外标志位的“原子抢占”，就能够在主线程与超时线程之间建立正确的竞争关系，从而避免竞态条件带来的误判。

Accept()

而这里最后，我们来解答上文埋下的一个关键点，也就是主线程获取新连接的处理策略。

当主线程检测到监听套接字的读事件就绪时，便会执行获取新连接的操作。我们知道，监听套接字的读事件就绪，本质上意味着全连接队列（accept 队列）不为空。而调用 accept 接口时，其语义只是从全连接队列的队首取出一个已完成连接的套接字（即一个已建立连接的 fd）。

然而，在高并发服务器场景下，服务端会持续接收大量客户端的连接请求，底层 TCP 协议栈不断完成三次握手，将新连接放入全连接队列。如果每次仅调用一次 accept，只取出一个连接，那么在高并发压力下，全连接队列很可能在极短时间内再次被填满，甚至出现队列溢出（backlog 被打满）的情况。

因此，对于监听套接字的读事件处理，正确的做法并不是“处理一个连接”，而是尽可能清空当前全连接队列。也就是说，需要循环调用 accept，直到队列被取空为止。

---

但这里存在一个容易踩的关键问题：

当最后一个已连接套接字被取走后，全连接队列变为空，而 accept 的默认行为是阻塞等待新的连接到来。这就会导致主线程被挂起，无法继续处理其他已经就绪的 I/O 事件（例如已有连接的读写事件），从而影响整个事件循环的响应能力。

因此，这里必须将监听套接字设置为非阻塞模式，使得当全连接队列为空时，accept 能够立即返回，而不是阻塞等待。

实现方式是通过 fcntl 接口修改文件描述符的状态标志位：

• 系统调用：fcntl
• 头文件：<fcntl.h>
• 函数声明：int fcntl(int fd, int cmd, ... /* arg */ );
• 返回值：成功时返回值取决于 cmd（例如 F_GETFL 返回文件状态标志），失败时返回 -1 并设置 errno

在内核中，每个 fd 对应一个 file 结构体，其中包含 flags 字段，用于描述该文件描述符的行为属性（如阻塞/非阻塞、读写模式（比如O_RDONLY）等）。因此，这里的操作流程是：先获取当前标志位，再进行按位修改，最后写回。

代码示例：

 // 获取文件状态标志
int flags = fcntl(fd, F_GETFL, 0);

// 设置非阻塞标志
flags |= O_NONBLOCK;
fcntl(fd, F_SETFL, flags);

设置为非阻塞之后，当全连接队列为空时，accept 会立即返回 -1，并将 errno 设置为 EAGAIN 或 EWOULDBLOCK，表示“当前操作无法立即完成，需要稍后重试”。这正是我们用来判断“队列已经被取空”的依据。

---

结合上述机制，监听套接字的处理逻辑可以实现为：

void Accept(Pollmanager* read_arg)
{
    while(true){
        struct sockaddr_in Client;
        memset(&Client, 0, sizeof(Client));
        socklen_t Client_len = sizeof(Client);

        int connfd = listensock.accept(&Client,&Client_len);
        if(connfd < 0)
        {
            if(errno == EAGAIN || errno == EWOULDBLOCK)
            {
                // 全连接队列已取空
                break;
            }
            return;
        }

        last_active_time[connfd] = time(NULL);

        send_queues_mtx.lock();
        send_queues[connfd] = std::make_shared<Connection>(connfd);
        send_queues_mtx.unlock();

        read_arg->add_fd(connfd);
    }
}

源码

Poll_Server.hpp:

#pragma once
#include<iostream>
#include<unistd.h>
#include<signal.h>
#include"daemon.hpp"
#include"Socket.hpp"
#include"log.hpp"
#include<string>
#include<poll.h>
#include<sys/time.h>
#include<sys/types.h>
#include<thread>
#include<queue>
#include<unordered_map>
#include<vector>
#include"Task.hpp"
#include"Threadpool.h"
#include"Pollmanager.hpp"
#include<mutex>
#include<time.h>
#include<atomic>

extern  std::string _deafault;
extern log lg;
const int write_thread_num = 16;

/** Poll_Server 类实现了一个基于poll的多线程服务器*/
class Poll_Server
{
public:
        Poll_Server(uint16_t _port=8888, std::string _ip=_deafault)
                :port(_port)
                , ip(_ip)
            ,read_tp(threadpool::getinstance()) // 获取线程池单例
        {
        // 初始化最后活动时间数组
           for(int i=0;i<sizeof(last_active_time)/sizeof(time_t);i++)
           {
              last_active_time[i]=0;
           }
        }
        static void check_timeout_fun(Poll_Server* server)
        {
            while(1)
            {
                time_t now=time(NULL); // 获取当前时间
                std::vector<pollfd> poll_fds=server->read_pollmanger.get_snapshot();
            // 遍历所有文件描述符
                for(auto& read_active:poll_fds)
                {
                    if(read_active.fd!=server->listensock.fd())
                    {
                    // 检查是否超过60秒无活动
                         if(now-server->last_active_time[read_active.fd]>60)
                         {
                        // 删除超时的文件描述符
                               if(server->read_pollmanger.del_fd(read_active.fd)){
                            lg.logmessage(info,"client fd:%d timeout",read_active.fd);
                            server->read_pollmanger.del_fd(read_active.fd);
                            send_queues_mtx.lock();
                            // 检查发送队列中是否存在该连接
                            if(send_queues.find(read_active.fd)==send_queues.end())
                            {
                                send_queues_mtx.unlock();
                                continue;
                            }
                            // 获取连接并标记为需要关闭
                           std::shared_ptr<Connection> conn=send_queues[read_active.fd];
                            send_queues_mtx.unlock();
                            conn->mtx.lock();
                            conn->should_close=true;
                            conn->mtx.unlock();
                            // 将文件描述符添加到写轮询管理器
                            server->write_pollmangers[read_active.fd%write_thread_num]->add_fd(read_active.fd);
                         }
                      }
                    }
                }
                sleep(30); // 每30秒检查一次
        }
        }
    
  
   static void write_thread_func(Pollmanager* write_arg, Pollmanager* read_arg)
{
    while(1) {  // 无限循环处理发送任务
        // 获取当前需要监听的文件描述符快照
        std::vector<pollfd> poll_fds = write_arg->get_snapshot();
        
        // 如果没有需要监听的描述符，短暂休眠避免空转
        if(poll_fds.empty()) {
            usleep(10000); // 休眠10ms
            continue;
        }
        
        // 使用poll等待可写事件，超时时间为10ms
        int n = poll(poll_fds.data(), poll_fds.size(), 10);
        
        if(n < 0) {
            // poll调用出错，记录错误日志并退出循环
            lg.logmessage(Fatal, "Poll error");
            break;
        } else if(n == 0) {
            // poll超时，没有描述符可写
            lg.logmessage(info, "write_thread poll timeout");
            continue;
        }
        
        // 遍历所有监听的文件描述符
        for(auto& i : poll_fds) {
            // 检查是否有可写事件
            if(i.revents & POLLOUT) {
                // 加锁获取连接对象
                send_queues_mtx.lock();
                std::shared_ptr<Connection> conn = send_queues[i.fd];
                send_queues_mtx.unlock();
                
                // 加锁连接对象，确保线程安全
                conn->mtx.lock();
                
                bool error_occurred = false;
                bool eagain_occurred = false;
                
                // 循环发送缓冲区中的所有数据
                while(!conn->out_buffer.empty()) {
                    // 获取待发送数据
                    std::string data = conn->out_buffer.front();
                    
                    // 尝试发送数据（非阻塞模式）
                    ssize_t send_bytes = send(i.fd, 
                                             data.c_str() + conn->write_offset, 
                                             data.size() - conn->write_offset, 
                                             MSG_DONTWAIT);
                    
                    if(send_bytes < 0) {
                        // 发送出错
                        if(errno == EAGAIN) {
                            // 缓冲区满，暂时无法发送
                            eagain_occurred = true;
                            break;
                        }
                        // 其他错误，记录日志
                        lg.logmessage(Fatal, "send error");
                        error_occurred = true;
                        break;
                    } else {
                        // 更新已发送的字节数
                        conn->write_offset += send_bytes;
                        
                        // 如果当前数据块发送完毕
                        if(conn->write_offset == data.size()) {
                            conn->write_offset = 0;
                            conn->out_buffer.pop();
                        }
                    }
                }
                
                // 处理发送完成后的状态
                if(error_occurred || (conn->should_close && conn->out_buffer.empty())) {
                    // 发生错误或需要关闭连接
                    read_arg->del_fd(i.fd);  // 从读监听中移除
                    write_arg->del_fd(i.fd); // 从写监听中移除
                    
                    // 加锁移除连接
                    send_queues_mtx.lock();
                    conn->mtx.unlock();
                    send_queues.erase(i.fd);
                    send_queues_mtx.unlock();
                } else if(eagain_occurred) {
                    // 缓冲区满，暂时无法发送，保持连接在监听中
                    conn->mtx.unlock();
                } else if(conn->should_close == false && conn->out_buffer.empty()) {
                    // 数据发送完毕且不需要关闭连接，从写监听中移除
                    write_arg->del_fd(i.fd);
                    conn->mtx.unlock();
                }
            }
        }
    }
}

    
/**
 * 接受客户端连接的函数
 */
        void Accept(Pollmanager* read_arg)
        {
                while(true)  // 无限循环，持续接受新的连接
                {
                struct sockaddr_in Client;  // 客户端地址结构体
                memset(&Client, 0, sizeof(Client));  // 清空客户端地址结构体
                socklen_t Client_len = sizeof(Client);  // 客户端地址结构体长度
        // 接受新的连接，获取套接字描述符
                int connfd = listensock.accept(&Client,&Client_len);
                if(connfd<0)  // 如果接受连接失败
                {
            // 如果错误是EAGAIN或EWOULDBLOCK，表示没有更多连接需要处理，退出循环
                    if(errno==EAGAIN||errno==EWOULDBLOCK)
                    {
                        break;
                    }
                   
            // 其他错误，直接返回
                    return;
                }
        // 记录连接的最后活动时间为当前时间
                last_active_time[connfd]=time(NULL);
        // 加锁保护共享资源
                send_queues_mtx.lock();
        // 为新连接创建Connection对象并存入发送队列
                send_queues[connfd]=std::make_shared<Connection>(connfd);
                send_queues_mtx.unlock();
        // 将新的连接描述符添加到Pollmanager中管理
                read_arg->add_fd(connfd);
                 }
        }
/**
 * 分发函数，用于将文件描述符从一个poll管理器转移到任务队列中
 */
        void Distribute(int fd,Pollmanager* read_arg)
        {
    // 从读取poll管理器中删除该文件描述符
                   read_arg->del_fd(fd);
    // 创建一个任务对象，包含以下参数：
    // - fd: 文件描述符
    // - read_arg: 读取poll管理器指针
    // - write_pollmangers[fd%write_thread_num]: 根据fd计算得到的写入poll管理器
    // - last_active_time: 最后活跃时间
                   Task t(fd,read_arg,write_pollmangers[fd%write_thread_num],last_active_time);
    // 将创建的任务推入读取线程池的任务队列
                   read_tp.push(t);
        }
/**
 * 处理任务函数，根据poll事件进行相应的处理
 */
        void handlertask(Pollmanager* read_arg,const std::vector<pollfd>& poll_fds)
        {
    // 遍历所有的pollfd结构
               for(auto& i:poll_fds)
               {
        // 检查是否有可读事件发生
                if(i.revents&POLLIN)
                {
            // 如果是监听套接字有事件，则接受新连接
                     if(i.fd==listensock.fd())
                     {
                        Accept(read_arg);
                     }else{
                // 否则，将事件分发给对应的处理函数
                        Distribute(i.fd,read_arg);
                     }
                }
               }
        }
/**
 * 初始化函数，用于设置监听套接字
 */
         bool init()
        {
    // 创建套接字
                listensock.socket();
    // 绑定IP地址和端口号
                listensock.bind(ip, port);
    // 将套接字设置为非阻塞模式
                listensock.setnonblock();
    // 返回初始化成功标志
                return true;
        }
void start(bool isdaemon=false)
{
    // 忽略SIGPIPE信号，防止向已关闭的socket写入导致进程崩溃
    signal(SIGPIPE, SIG_IGN); 
    
    // 如果需要以守护进程方式运行
    if(isdaemon)
    {
        Daemon d;
        d.daemon();  // 将当前进程转换为守护进程
    }
    
    // 开始监听客户端连接
    listensock.listen();
    
    // 启动读线程池
    read_tp.start();
    
    // 启动超时检查线程，并分离
    std::thread(check_timeout_fun, this).detach();
    
    // 创建写线程池
    for(int i = 0; i < write_thread_num; i++)
    {
        // 为每个写线程创建一个Pollmanager对象
        write_pollmangers.push_back(new Pollmanager(2));
        // 启动写线程，传入写Pollmanager和读Pollmanager
        std::thread(write_thread_func, write_pollmangers.back(), &read_pollmanger).detach();
    }
    
    // 获取监听套接字的文件描述符
    int listensock_fd = listensock.fd();
    
    // 将监听套接字添加到读Pollmanager中
    read_pollmanger.add_fd(listensock_fd);
    
    // 主事件循环
    while (1)
    {
        // 获取当前需要监听的文件描述符快照
        std::vector<pollfd> poll_fds = read_pollmanger.get_snapshot();
        
        // 使用poll等待事件，超时时间为10ms
        int n = poll(poll_fds.data(), poll_fds.size(), 10);
        
        if(n < 0)
        {
            // poll调用出错，记录错误日志
            lg.logmessage(Fatal, "poll error");
            break;
        }
        else if(n == 0)
        {
            // poll超时，没有事件发生
            lg.logmessage(info, "poll timeout");
        }
        else
        {
            // 有事件发生，调用任务处理函数
            handlertask(&read_pollmanger, poll_fds);
        }
    }
}

private:
        sock listensock;
        uint16_t port;
        std::string ip;
        Pollmanager read_pollmanger{1};
        std::vector<Pollmanager*> write_pollmangers;
        threadpool& read_tp;
        std::atomic<time_t> last_active_time[65536];
};

Poll_Server.cpp:

#include"Poll_Server.hpp"
#include"log.hpp"
#include<stdlib.h>
#include<memory>
 log lg;
 std::string path="./wwwroot";
 std::string _deafault="0.0.0.0";
 std::mutex send_queues_mtx;
 std::unordered_map<int,std::shared_ptr<Connection>> send_queues;
/**
 * 主函数，负责初始化并启动服务器
 */
int main()
{
    // 定义一个字符数组用于存储解析后的绝对路径
      char resolved_path[1024];
    // 使用realpath函数获取文件的绝对路径，如果失败则记录错误日志并退出
      if(realpath(path.c_str(),resolved_path)==nullptr)
      {
        // 记录致命错误日志
        lg.logmessage(Fatal,"realpath error");
        // 返回错误状态码
        return -1;
      }
    // 将解析后的绝对路径赋值给path变量
      path=resolved_path;
    // 创建一个Poll_Server智能指针实例，监听8888端口
      std::unique_ptr<Poll_Server> server(new Poll_Server(8888));
    // 初始化服务器，如果失败则记录错误日志并退出
      if(!(server->init()))
      {
        // 记录致命错误日志
        lg.logmessage(Fatal,"server init error");
        // 返回错误状态码
        return -1;
      }
    // 启动服务器
      server->start();
}

Pollmanager.hpp:

#pragma once
#include<mutex>
#include<poll.h>
#include<sys/types.h>
#include<vector>
#include<memory>
#include<unistd.h>
#include<unordered_map>
/**
 *  Pollmanager类是一个基于poll的事件管理器，用于管理文件描述符(fd)及其事件
 */
class Pollmanager
{
    public:
    /**
     _event 指定要监听的事件类型，1表示POLLIN(可读)，2表示POLLOUT(可写)
     */
    Pollmanager(int _event)
    :event(_event)
    {

    }
    /**
     * 添加文件描述符到poll管理器
     */
    void add_fd(int fd)
    {
       // 使用std::lock_guard自动管理互斥锁
       std::lock_guard<std::mutex> lock(mtx);
       // 检查fd是否已经存在，存在则直接返回
       if(fd_index_map.find(fd)!=fd_index_map.end())
       {
             return;
       }
       // 创建新的pollfd结构体
       pollfd pfd;
       pfd.fd=fd;
       // 根据构造函数传入的event参数设置事件类型
       if(event==1)
       {
         pfd.events=POLLIN;  // 设置为可读事件
       }else if(event==2)
       {
        pfd.events=POLLOUT; // 设置为可写事件
       }
       pfd.revents=0;       // 初始化revents为0
       // 将pollfd结构体添加到vector中
       pollfds.push_back(pfd);
       // 更新fd到索引的映射关系
       fd_index_map[fd]=pollfds.size()-1;
    }
/**
 * 从pollfds集合中删除指定的文件描述符
 */
    bool del_fd(int fd)
    {
    // 使用互斥锁确保线程安全
       std::lock_guard<std::mutex> lock(mtx);
    // 检查文件描述符是否存在
       if(fd_index_map.find(fd)==fd_index_map.end())
       {
        // 文件描述符不存在，返回false
         return false;
       }
    // 获取文件描述符在pollfds中的索引位置
       int index=fd_index_map[fd];
    // 如果要删除的是最后一个元素
       if(index==pollfds.size()-1)
       {
        // 直接删除最后一个元素
           pollfds.pop_back();
        // 从映射表中删除该文件描述符
           fd_index_map.erase(fd);
           return true;
       }
    // 如果要删除的不是最后一个元素
    // 获取最后一个元素的文件描述符
       int last_fd=pollfds.back().fd;
    // 将最后一个元素移动到要删除的位置
       pollfds[index]=pollfds.back();
    // 删除最后一个元素
       pollfds.pop_back();
    // 更新被移动元素的索引映射
       fd_index_map[last_fd]=index;
    // 删除原文件描述符的映射
       fd_index_map.erase(fd);
       return true;

    }
/**
 * 获取当前pollfd集合的快照
 * 该函数会获取当前pollfd集合的副本，线程安全
 */
    std::vector<pollfd> get_snapshot()
    {
        std::lock_guard<std::mutex> lock(mtx); // 使用互斥锁确保线程安全
        return pollfds; // 返回pollfd集合的副本
    }
    private:
    std::mutex mtx;
    std::vector<pollfd> pollfds;
    std::unordered_map<int,int> fd_index_map;
    int event;
};

Socket.hpp:

#pragma once
#include<arpa/inet.h>
#include<netinet/in.h>
#include<fcntl.h>
#include<unistd.h>
#include<string>
#include<cstring>
#include<cstdlib>
#include"log.hpp"
extern log lg;
const int MAXCONNNUM =500;
enum
{
    Socket_Error = 1,
    Bind_Error,
    Listen_Error,
    Accept_Error,
    Connect_Error,
    Usage_Error,
    Setnonblock_Error,
};

class sock
{
public:
    sock()
        :socketfd(-1)
    {

    }
    ~sock()
    {
        if (socketfd >= 0)
        {
            ::close(socketfd);
        }
    }
    int fd()
    {
        return socketfd;
   }
/**
 * 创建一个TCP套接字并设置重用地址和端口选项
 * 如果创建失败，记录错误日志并退出程序
 */
    void socket()
    {
    // 创建TCP套接字，使用IPv4地址族和流式套接字
        socketfd = ::socket(AF_INET, SOCK_STREAM, 0);
        
    // 检查套接字是否创建成功
        if (socketfd < 0)
        {
           
        // 记录致命错误日志
            lg.logmessage(Fatal, "socket error");
        // 设置套接字描述符为无效值
            socketfd = -1;
        // 退出程序，错误码为Socket_Error
            exit(Socket_Error);
        }
    // 设置套接字选项，允许重用地址和端口
        int opt = 1;
        setsockopt(socketfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
      
    // 记录信息日志，表明套接字创建成功
        lg.logmessage(info, "socket successfully");
    }

/**
 * 绑定IP地址和端口号到套接字
 */
    void bind(std::string ip, uint16_t port)
    {
        
    // 检查套接字是否有效
        if (socketfd < 0) 
        {
            lg.logmessage(Fatal, "socket not created");    // 记录致命错误：套接字未创建
            exit(Socket_Error);   // 退出程序，套接字错误
        }

      
    // 创建服务器地址结构体
        struct sockaddr_in server;  
        memset(&server, 0, sizeof(server));    // 将地址结构体清零

       
    // 设置地址族为IPv4
        server.sin_family = AF_INET;  
    // 将端口号从主机字节序转换为网络字节序
        server.sin_port = htons(port);  
    // 处理IP地址
        if (ip == "0.0.0.0")  
        {
          
        // 绑定到所有可用的网络接口
            server.sin_addr.s_addr = INADDR_ANY; 
        }
        else if (inet_pton(AF_INET, ip.c_str(), &server.sin_addr) <= 0)  
        {
           

        // 将IP地址字符串转换为网络地址格式失败
            lg.logmessage(Fatal, "inet_pton fail");    // 记录致命错误：IP地址转换失败
            ::close(socketfd);    // 关闭套接字
            socketfd = -1;    // 重置套接字描述符
            exit(Bind_Error);    // 退出程序，绑定错误
        }
    // 获取服务器地址结构体的长度
        socklen_t serverlen = sizeof(server);  
    // 绑定套接字到指定地址和端口
        int n = ::bind(socketfd, (struct sockaddr*)&server, serverlen);  
    // 检查绑定是否成功
        if (n < 0)  
        {
            lg.logmessage(Fatal, "bind error");    // 记录致命错误：绑定失败
            ::close(socketfd);    // 关闭套接字
            socketfd = -1;    // 重置套接字描述符
            exit(Bind_Error);    // 退出程序，绑定错误

        }
    // 记录绑定成功信息
        lg.logmessage(info, "bind successfully");  
    }
   
/**
 * 监听函数，用于使套接字进入监听状态，准备接受客户端连接
 * 如果套接字未创建或监听失败，会记录错误信息并退出程序
 */
    void listen()
    {
       
    // 检查套接字是否有效（是否已创建）
        if (socketfd < 0)
        {
        // 记录致命错误：套接字未创建
            lg.logmessage(Fatal, "socket not created");
        // 退出程序，错误代码为Socket_Error
            exit(Socket_Error);
        }
    // 调用系统listen函数开始监听，最大连接数为MAXCONNNUM
        int n = ::listen(socketfd, MAXCONNNUM);
    // 检查listen函数是否成功执行
        if (n < 0)
        {
        // 记录致命错误：监听失败
            lg.logmessage(Fatal, "listen error");
        // 关闭套接字
            ::close(socketfd);
        // 将套接字描述符重置为无效值
            socketfd = -1;
        // 退出程序，错误代码为Listen_Error
            exit(Listen_Error);
        }
    // 记录信息：监听成功
        lg.logmessage(info, "listen successfully");
    }

      /**
 *  将socketfd设置为非阻塞模式
 * 
 * 该函数通过fcntl系统调用修改文件描述符的标志位，
 * 添加O_NONBLOCK标志，使socket操作变为非阻塞模式。
 * 在非阻塞模式下，I/O操作不会阻塞进程，而是立即返回。
 */
void setnonblock()
{
    // 获取当前文件描述符的状态标志
    int flags = fcntl(socketfd, F_GETFL, 0);
    
    // 检查获取标志是否成功
    if(flags < 0)
    {
        // 记录错误日志
        lg.logmessage(Fatal, "fcntl error");
        // 退出程序，返回错误码
        exit(Setnonblock_Error);
    }
    
    // 添加非阻塞标志
    flags |= O_NONBLOCK;
    
    // 设置新的文件描述符标志
    if(fcntl(socketfd, F_SETFL, flags) < 0)
    {
        // 记录错误日志
        lg.logmessage(Fatal, "fcntl error");
        // 退出程序，返回错误码
        exit(Setnonblock_Error);
    }
}


    /**接受客户端连接请求
 * 
 * 该函数封装了系统调用accept，用于接受客户端的连接请求。
 * 它会检查socket状态，处理非阻塞模式下的EAGAIN错误，
 * 并记录连接建立或失败的日志。
 */
int accept(struct sockaddr_in* client, socklen_t* clientlen)
{
    // 检查监听socket是否有效
    if (socketfd < 0)
    {
        // 记录错误日志：socket未创建
        lg.logmessage(Fatal, "socket not created");
        // 退出程序，返回socket错误码
        exit(Socket_Error);
    }

    // 调用系统accept函数接受客户端连接
    int client_fd = ::accept(socketfd, (struct sockaddr*)client, clientlen);
    
    // 检查accept是否成功
    if (client_fd < 0)
    {
        // 处理非阻塞模式下的EAGAIN/EWOULDBLOCK错误
        // 这表示当前没有连接可接受，不是真正的错误
        if(errno == EAGAIN || errno == EWOULDBLOCK){
            return -1;  // 返回-1表示暂时没有连接
        }
        
        // 记录accept错误日志
        lg.logmessage(Fatal, "accept error");
        return -1;  // 返回-1表示accept失败
    }

    // 记录成功接受连接的日志
    lg.logmessage(info, "accept successfully");
    
    // 返回新的客户端socket文件描述符
    return client_fd;
}

/**
 * 连接到服务器的函数
 */
    void connect(struct sockaddr_in* server, socklen_t serverlen)
    {
    // 检查socket是否已创建
        if (socketfd < 0)
        {
        // 记录错误日志并退出程序
            lg.logmessage(Fatal, "socket not created");
            exit(Socket_Error);
        }
    // 尝试连接到服务器
        int n = ::connect(socketfd, (struct sockaddr*)server, serverlen);
    // 检查连接是否成功
        if (n < 0)
        {
        // 记录错误日志，关闭socket，重置socketfd，并退出程序
            lg.logmessage(Fatal, "connect error");
            ::close(socketfd);
            socketfd = -1;
            exit(Connect_Error);
        }
    // 记录连接成功的日志
        lg.logmessage(info, "connect successfully");
    }
    void close()
    {
        if (socketfd >= 0)
        {
            ::close(socketfd);
            socketfd = -1;
        }
    }
    sock(const sock&) = delete;
    sock& operator=(const sock&) = delete;
private:
    int socketfd;
};

Task.hpp:

#pragma once
#include<iostream>
#include<fstream>
#include<string>
#include<sys/types.h>
#include<unistd.h>
#include<time.h>
#include<unordered_map>
#include"protocol.hpp"
#include"log.hpp"
#include"Pollmanager.hpp"
#define BUFFER_SIZE 1024
#define RECV_TIMEOUT 1
#define RECV_ERROR -1
#define CLIENT_CLOSE 2
#include<queue>
#include<mutex>
#include<atomic>
extern log lg;
extern std::string path;
extern std::mutex send_queues_mtx;

class threadpool;

/**
 * 枚举类型：解析状态
 * 用于表示当前解析过程所处的阶段状态
 */
enum ParseState 
{
    PARSE_HEDER,  // 解析头部状态，表示当前正在解析数据头部部分
    PARSE_BODY    // 解析主体状态，表示当前正在解析数据主体部分
};

/**
 * Connection类，表示一个网络连接
 * 负责管理连接的读写缓冲区、解析状态以及文件描述符等
 */
class Connection
{
   public:
   /**
    * Connection类的构造函数
    *  _fd 文件描述符，用于标识网络连接
    */
Connection(int _fd)
:write_offset(0)      // 写偏移量，初始化为0
,should_close(false)  // 是否应该关闭连接，初始化为false
,parse_state(PARSE_HEDER)  // 解析状态，初始化为解析头部状态
,header_length(0)     // 头部长度，初始化为0
,body_length(0)       // 数据体长度，初始化为0
,header("")           // 头部内容，初始化为空字符串
,fd(_fd)              // 文件描述符，使用传入的参数初始化
{

}
   /**
    * Connection类的析构函数
    * 如果文件描述符有效（大于0），则关闭连接
    */
~Connection()
    {
        if(fd>0)  // 检查文件描述符是否有效
        {
          close(fd);  // 关闭文件描述符，释放资源
               }
   }
   // 输入缓冲区，用于存储接收到的数据
   std::queue<std::string> in_buffer;
   // 输出缓冲区，用于存储待发送的数据
   std::queue<std::string> out_buffer;
   // 互斥锁，用于保护共享资源的访问
   std::mutex mtx;
   // 写偏移量，记录当前写入位置
   int write_offset;
   // 标记连接是否应该被关闭
   bool should_close;
   // 解析状态，指示当前解析的阶段（头部、数据体等）
   ParseState parse_state;
   // 已解析的头部长度
   size_t header_length;
   // 数据体长度
   size_t body_length;
   // 存储头部内容
   std::string header;
   // 文件描述符，标识网络连接
   int fd;
};

extern std::unordered_map<int,std::shared_ptr<Connection>> send_queues;

/**
 *  从socket接收数据并解析HTTP请求
 * 
 * 该函数负责从指定的socket接收数据，解析HTTP请求头和请求体，
 * 并将解析结果存储在Http_Request对象中。它处理非阻塞I/O，
 * 支持分块接收数据，并维护连接的解析状态。
 * 
 */
int Get_HttpRequest(size_t socketfd, Http_Request& hr)
{
    // 加锁查找对应的连接对象
    send_queues_mtx.lock();
    if(send_queues.find(socketfd) == send_queues.end())
    {
        send_queues_mtx.unlock();
        return RECV_ERROR;  // 连接不存在，返回错误
    }
    
    // 获取连接对象并解锁
    std::shared_ptr<Connection> conn = send_queues[socketfd];
    send_queues_mtx.unlock();
    
    // 确保输入缓冲区至少有一个节点
    if(conn->in_buffer.empty())
    {
        conn->in_buffer.push("");
    }
    
    // 获取缓冲区前端的引用
    std::string& front_node = conn->in_buffer.front();
    
    // 临时缓冲区用于接收数据
    char temp_buffer[BUFFER_SIZE];
    
    // 循环接收数据，直到遇到EAGAIN错误
    while(true)
    {
        // 非阻塞接收数据
        int read_bytes = recv(socketfd, &temp_buffer[0], BUFFER_SIZE - 1, MSG_DONTWAIT);
        
        if(read_bytes < 0)
        {
            // 检查是否是暂时没有数据可读
            if(errno == EAGAIN)
            {
                break;  // 退出接收循环
            }
            // 其他错误，记录日志并返回错误
            lg.logmessage(Fatal, "recv error");
            return RECV_ERROR;
        }
        else if(read_bytes == 0)
        {
            // 客户端关闭了连接
            lg.logmessage(info, "client closed connection, fd:%d", socketfd);
            return CLIENT_CLOSE;
        }
        
        // 将接收到的数据追加到缓冲区
        front_node.append(temp_buffer, read_bytes);
    }
    
    // 解析HTTP请求头
    if(conn->parse_state == PARSE_HEDER)
    {
        // 查找请求头结束标记
        size_t head_end_pos = front_node.find("\r\n\r\n");
        
        if(head_end_pos == std::string::npos)
        {
            // 请求头不完整，等待更多数据
            return RECV_TIMEOUT;
        }
        else
        {
            // 提取请求头
            std::string head = front_node.substr(0, front_node.find("\r\n\r\n") + 4);
            conn->header = head;
            conn->header_length = head.size();
            
            // 查找Content-Length字段
            size_t pos = head.find("Content-Length:");
            if(pos == std::string::npos)
            {
                // 尝试小写形式
                pos = head.find("content-length:");
            }
            
            if(pos != std::string::npos)
            {
                // 提取Content-Length值
                size_t endpos = head.find("\r\n", pos);
                std::string content_length_str = head.substr(pos + 15, endpos - pos - 15);
                size_t content_length = std::stoi(content_length_str);
                conn->body_length = content_length;
            }
            else
            {
                // 没有Content-Length字段，表示没有请求体
                conn->body_length = 0;
            }
            
            // 更新解析状态
            conn->parse_state = PARSE_BODY;
        }
    }
    
    // 解析HTTP请求体
    std::string body;
    if(conn->parse_state == PARSE_BODY)
    {
        // 计算剩余数据量
        size_t remaining = front_node.size() - conn->header_length;
        std::string extra_data = "";
        
        if(remaining < conn->body_length)
        {
            // 请求体不完整，等待更多数据
            return RECV_TIMEOUT;
        }
        else
        {
            // 提取请求体
            body = front_node.substr(conn->header_length, conn->body_length);
        }
        
        // 处理多余的数据（可能包含下一个请求的开始）
        if(remaining > conn->body_length)
        {
            extra_data = front_node.substr(conn->header_length + conn->body_length);
        }
        
        // 将多余的数据放回缓冲区
        conn->in_buffer.push(extra_data);
        
        // 设置请求体并反序列化请求头
        hr.text = body;
        hr.Deserialization(conn->header);
        hr.debugprint();
        
        // 移除已处理的数据
        conn->in_buffer.pop();
        
        // 重置解析状态
        conn->parse_state = PARSE_HEDER;
        conn->header_length = 0;
        conn->body_length = 0;
        conn->header = "";
    }
    
    return 0;  // 成功解析请求
}

/**
 * 读取文件内容并返回字符串
 *  file_path 文件路径
 * return 文件内容字符串，如果文件不存在则返回空字符串
 */
std::string read_file(std::string file_path)
{
    // 以二进制模式打开文件
    std::ifstream file(file_path, std::ios::binary);
    // 检查文件是否成功打开
    if (!file.is_open())
    {
        // 记录文件未找到的日志
        lg.logmessage(info, "file not found:%s", file_path.c_str());
        return "";
    }
    // 获取文件起始位置
    std::streampos start = file.tellg();
    // 将文件指针移动到末尾
    file.seekg(0, std::ios::end);
    // 获取文件结束位置
    std::streampos end = file.tellg();
    // 计算文件大小
    size_t file_size = end - start;
    // 创建字符串并调整大小以容纳文件内容
    std::string content;
    content.resize(file_size);
    // 将文件指针重置到文件开头
    file.seekg(0, std::ios::beg);
    // 读取文件内容到字符串
    file.read(&content[0], file_size);
    // 关闭文件
    file.close();
    return content;
}

std::string Http_Get_Handler(Http_Request& hr);

/**
 * 处理HTTP错误请求的函数，返回一个400 Bad Request的HTTP响应
 *  hr HTTP请求对象，包含请求头等信息
 * return 返回一个格式化的HTTP响应字符串，包含状态行、头部和内容
 */
std::string process_bad_request(Http_Request& hr)
{
    // 记录错误日志，级别为Fatal
    lg.logmessage(Fatal, "bad request body");
    // 设置HTTP响应状态行
    std::string headler_line = "HTTP/1.0 400 Bad Request\r\n";
    // 初始化响应头字符串
    std::string header;
    // 根据请求中的Connection头决定保持连接还是关闭连接
    if(hr.headers["Connection"]=="close"||hr.headers["Connection"]=="Close")
    {
        // 设置Connection头为Close
           header="Connection: Close\r\n";
    }else if(hr.headers["Connection"]=="keep-alive"||hr.headers["Connection"]=="Keep-Alive")
    {
        // 设置Connection头为Keep-Alive
         header="Connection: Keep-Alive\r\n";
    }
    // 设置响应内容
    std::string content = "Bad Request";
    // 添加Content-Length头，指示响应内容的长度
    header += "Content-Length: " + std::to_string(content.size()) + "\r\n";
    // 添加Content-Type头，指示响应内容的类型为纯文本
    header += "Content-Type: text/plain\r\n";
    // 添加空行，表示头部结束
    header += "\r\n";
    // 返回完整的HTTP响应：状态行 + 头部 + 内容
    return headler_line + header + content;
}
/**
 * 执行简单的四则运算计算
 * 
 * 该函数从输入的参数映射中提取操作数和运算符，执行相应的数学运算，
 * 并将结果存储在输出参数中。支持处理URL编码的运算符。
 * 
 *  val 包含计算参数的键值对映射，键包括"a"、"b"和"op"
 *  result 用于存储计算结果的引用参数
 * return bool 计算成功返回true，失败返回false
 */
bool process_calculation(std::unordered_map<std::string, std::string>& val, int& result)
{
    // 从参数映射中提取操作数a和b，并转换为整数
    int a = std::stoi(val["a"]);
    int b = std::stoi(val["b"]);

    // 获取运算符
    std::string op = val["op"];
    
    // 处理URL编码的运算符，统一转换为标准运算符
    if (op == "+" || op == "%2B" || op == "%2b") 
    {
        op = "+";  // 加号
    }
    else if (op == "-" || op == "%2D" || op == "%2d")
    {
        op = "-";  // 减号
    }
    else if (op == "*" || op == "%2A" || op == "%2a") 
    {
        op = "*";  // 乘号
    }
    else if (op == "/" || op == "%2F" || op == "%2f")
    {
        op = "/";  // 除号
    }
    else
    {
        // 不支持的运算符，记录错误日志并返回false
        lg.logmessage(Fatal, "unsupported operator:%s", op.c_str());
        return false;
    }

    // 根据运算符执行相应的计算
    switch (op[0]) 
    {
        case '+': 
            result = a + b;  // 加法
            break;
        case '-': 
            result = a - b;  // 减法
            break;
        case '*': 
            result = a * b;  // 乘法
            break;
        case '/': 
            // 检查除数是否为零
            if (b == 0)
            {
                lg.logmessage(warning, "division by zero");
                return false;
            }
            result = a / b;  // 除法
            break; 
    }

    // 计算成功，返回true
    return true;
}

/**
 * 处理HTTP POST请求的函数
 *  hr HTTP请求对象，包含请求的URL、头部信息和请求体等
 * return 返回处理后的HTTP响应字符串
 */
std::string Http_Post_Handler(Http_Request& hr) {
    std::string res; // 用于存储最终的HTTP响应
    std::unordered_map<std::string, std::string>val; // 存储解析后的键值对
    size_t start = 0; // 用于记录字符串查找的起始位置
    // 处理"/calc"路径的POST请求
    if (hr.url == "/calc")
    {
        std::string body = hr.text; // 获取HTTP请求体
        size_t pos1 = body.find("&"); // 查找第一个分隔符"&"
        if (pos1 == std::string::npos) // 如果没有找到分隔符，返回错误响应
        {
            return process_bad_request(hr);
        }
        // 解析第一个键值对
        std::string expression = body.substr(start, pos1);
        size_t pos2 = expression.find("="); // 查找键值分隔符"="
        if (pos2 == std::string::npos) // 如果没有找到分隔符，返回错误响应
        {
            return process_bad_request(hr);
        }
        std::string result_key_str = expression.substr(start, pos2); // 提取键
        std::string result_value_str = expression.substr(pos2 + 1); // 提取值
        val[result_key_str] = result_value_str; // 存储到map中
        start = pos1 + 1; // 更新查找起始位置

        
        // 解析第二个键值对
        pos1 = body.find("&", start); // 查找下一个分隔符
        if (pos1 == std::string::npos) // 如果没有找到分隔符，返回错误响应
        {
            return process_bad_request(hr);
        }
        pos2 = body.find("=", start); // 查找键值分隔符
        if (pos2 == std::string::npos || pos2 > pos1) // 如果格式错误，返回错误响应
        {
            return process_bad_request(hr);
        }
        result_key_str = body.substr(start, pos2 - start); // 提取键
        result_value_str = body.substr(pos2 + 1, pos1 - pos2 - 1); // 提取值
        val[result_key_str] = result_value_str; // 存储到map中
        start = pos1 + 1; // 更新查找起始位置

        
        // 解析第三个键值对
        pos2 = body.find("=", start); // 查找键值分隔符
        if (pos2 == std::string::npos) // 如果没有找到分隔符，返回错误响应
        {
            return process_bad_request(hr);
        }
        result_key_str = body.substr(start, pos2 - start); // 提取键
        result_value_str = body.substr(pos2 + 1); // 提取值
        val[result_key_str] = result_value_str; // 存储到map中
        int calc_result; // 存储计算结果
        if (process_calculation(val, calc_result) == false) // 如果计算失败，返回错误响应
        {
            return process_bad_request(hr);
        }

        
        // 构建HTTP响应头
        std::string headler_line = "HTTP/1.0 200 OK\r\n"; // 状态行
        std::string header; // 响应头
        // 根据请求头中的Connection字段决定连接方式
        if(hr.headers["Connection"]=="close"|| hr.headers["Connection"]=="Close")
        {
             header ="Connection: Close\r\n"; // 关闭连接
        }else if(hr.headers["Connection"]=="keep-alive"|| hr.headers["Connection"]=="Keep-Alive")
        {
           header="Connection: Keep-Alive\r\n";    // 保持连接
        }
        header += "Content-Type: text/html\r\n"; // 内容类型为HTML
        // 构建HTML响应体
        std::string content = "<html><head><meta charset='UTF-8'></head><body>";
        content += "<h2>计算结果展示</h2>";
        content += "<p style='font-size:24px;'>结果为: " + std::to_string(calc_result) + "</p>";
        content += "<a href='/'>返回首页</a>";
        content += "</body></html>";

        
        // 完成响应头
        header += "Content-Length: " + std::to_string(content.size()) + "\r\n"; // 内容长度
        header += "\r\n"; // 空行，表示头部结束
        // 组合完整的HTTP响应
        res = headler_line + header + content;
        return res;
    }
    else // 如果URL不是"/calc"，记录错误并返回错误响应
    {
        lg.logmessage(Fatal, "unsupported post url:%s", hr.url.c_str()); // 记录错误日志
        return process_bad_request(hr); // 返回错误响应
    }
}
class Task
{
public:
    Task()
        :socketfd(-1)
        ,read_manager(nullptr)
        ,write_manager(nullptr)
    {

    }
    /**
 *  Task类的构造函数
 * 
 * 初始化一个任务对象，用于处理网络连接相关的操作。
 * 该构造函数接收socket文件描述符、读写Pollmanager对象和最后活跃时间指针，
 * 并将它们存储在类的成员变量中，供后续使用。
 * 
 *  _socketfd 客户端连接的socket文件描述符
 *  _read_manager 用于管理读事件的Pollmanager对象指针
 *  _write_manager 用于管理写事件的Pollmanager对象指针
 *  _last_active_time 指向原子时间变量的指针，用于记录连接的最后活跃时间
 */
Task(int _socketfd, Pollmanager* _read_manager, Pollmanager* _write_manager, std::atomic<time_t>* _last_active_time)
    : socketfd(_socketfd)          // 初始化socket文件描述符成员变量
    , read_manager(_read_manager)  // 初始化读事件管理器成员变量
    , write_manager(_write_manager) // 初始化写事件管理器成员变量
    , last_active_time(_last_active_time) // 初始化最后活跃时间成员变量
{
    // 构造函数体为空，所有初始化已在初始化列表中完成
}

/**
 * 处理文件后缀名，返回对应的MIME类型
 *  suffix 文件后缀名，例如 ".html", ".css" 等
 * return 对应的MIME类型字符串，如果找不到则返回默认的 "text/html"
 */
    static std::string suffix_handler(std::string suffix)
    {
    // 在map中查找给定的后缀名
        auto pos = map.find(suffix);
    // 如果找不到指定的后缀名
        if (pos == map.end())
        {
        // 返回默认的HTML MIME类型
            return map[".html"];
        }
    // 返回找到的对应MIME类型
        return map[suffix];

    }
 void run();
private:
    int socketfd;
    static std::unordered_map<std::string, std::string> map;
    Pollmanager* read_manager;
    Pollmanager* write_manager;
   std::atomic<time_t>* last_active_time;
};
std::unordered_map<std::string, std::string> Task::map = {
    {".html","text/html"},
    {".css","text/css"},
    {".png","image/png"},
    {".jpg","image/jpeg"}
};

/**
 * 处理HTTP GET请求并生成响应
 * 
 * 该函数根据客户端请求的URL路径，读取相应的文件内容，
 * 并生成符合HTTP协议的响应报文。支持处理文件类型判断、
 * 404错误页面和连接管理（Keep-Alive/Close）。
 * 
 *  hr 包含HTTP请求信息的对象
 * return std::string 生成的HTTP响应报文
 */
std::string Http_Get_Handler(Http_Request& hr)
{
    // 存储请求的文件路径、内容类型和最终响应
    std::string file_path;
    std::string content_type;
    std::string res;
    
    // 处理根路径和首页请求
    if (hr.url == "/" || hr.url == "/index.html")
    {
        file_path = path + "/index.html";  // 设置为首页路径
        content_type = "text/html";         // 设置内容类型为HTML
    }
    else
    {
        // 处理其他路径请求
        file_path = path + hr.url;          // 拼接完整文件路径
        
        // 查找文件扩展名以确定内容类型
        ssize_t pos = file_path.rfind(".");
        if (pos == std::string::npos)
        {
            // 没有找到扩展名，默认为HTML
            content_type = "text/html";
        }
        else {
            // 提取扩展名并获取对应的内容类型
            std::string suffix = file_path.substr(pos);
            content_type = Task::suffix_handler(suffix);
        }
    }
    
    // 读取请求的文件内容
    std::string body = read_file(file_path);
    
    // HTTP响应状态行和头部
    std::string headler_line;
    std::string header;
    
    // 处理文件不存在的情况
    if (body.empty())
    {
        // 设置404状态行
        headler_line = "HTTP/1.0 404 Not Found\r\n";
        
        // 处理连接管理头部
        if(hr.headers["Connection"] == "close")
        {
            header += "Connection: Close\r\n";
        }
        else if(hr.headers["Connection"] == "keep-alive" || hr.headers["Connection"] == "Keep-Alive")
        {
            header += "Connection: Keep-Alive\r\n";
        }
        
        // 读取404错误页面内容
        std::string content = read_file(path + "/404.html");
        
        // 设置响应头部
        header += "Content-Length: " + std::to_string(content.size()) + "\r\n";
        header += "Content-Type: text/html\r\n";
        header += "\r\n";
        
        // 组合完整的响应
        res = headler_line + header + content;
    }
    else
    {
        // 文件存在，设置200状态行
        headler_line = "HTTP/1.0 200 OK\r\n";
        
        // 设置内容长度头部
        header += "Content-Length: " + std::to_string(body.size()) + "\r\n";
        
        // 处理连接管理头部
        if(hr.headers["Connection"] == "close" || hr.headers["Connection"] == "Close")
        {
            header += "Connection: Close\r\n";
        }
        else if(hr.headers["Connection"] == "keep-alive" || hr.headers["Connection"] == "Keep-Alive")
        {
            header += "Connection: Keep-Alive\r\n";
        }
        
        // 设置内容类型头部
        header += "Content-type: " + content_type + "\r\n";
        header += "\r\n";
        
        // 组合完整的响应
        res = headler_line + header + body;
    }
    
    // 返回完整的HTTP响应
    return res;
}

#include"Threadpool.h"

/**
 *  执行HTTP请求处理任务
 * 
 * 该函数是Task类的核心方法，负责处理HTTP请求的整个流程：
 * 1. 接收并解析HTTP请求
 * 2. 根据请求方法(GET/POST)调用相应的处理函数
 * 3. 生成HTTP响应并发送给客户端
 * 4. 管理连接状态(Keep-Alive/Close)
 * 5. 更新连接的最后活跃时间
 * 
 * 该函数处理了各种错误情况，包括接收超时、客户端关闭连接、
 * 不支持的请求方法等，并确保线程安全地访问共享资源。
 */
void Task::run()
{
    // 创建HTTP请求对象
    Http_Request hr;
    
    // 获取HTTP请求
    int get_result = Get_HttpRequest(socketfd, hr);
    
    // 检查获取请求结果
    if (get_result != 0)
    {
        // 处理接收超时情况
        if(get_result == RECV_TIMEOUT)
        {
            // 加锁保护发送队列
            send_queues_mtx.lock();
            // 检查socket是否在发送队列中
            if(send_queues.find(socketfd) == send_queues.end())
            {
                send_queues_mtx.unlock();
                return;
            }
            // 获取连接对象
            std::shared_ptr<Connection> conn = send_queues[socketfd];
            send_queues_mtx.unlock();
            
            // 加锁保护连接对象
            conn->mtx.lock();
            // 将socket添加到读取管理器，继续等待数据
            read_manager->add_fd(socketfd);
            conn->mtx.unlock();
            
            // 更新最后活动时间
            last_active_time[socketfd] = time(NULL);
            return;
        }
        
        // 处理客户端关闭连接情况
        if(get_result == CLIENT_CLOSE)
        {
            send_queues_mtx.lock();
            if(send_queues.find(socketfd) == send_queues.end())
            {
                send_queues_mtx.unlock();
                return;
            }
            std::shared_ptr<Connection> conn = send_queues[socketfd];
            send_queues_mtx.unlock();
            
            conn->mtx.lock();
            // 从读取管理器移除socket，不再监听读事件
            read_manager->del_fd(socketfd);
            // 标记连接需要关闭
            conn->should_close = true;
            // 将socket添加到写入管理器，发送剩余数据后关闭
            write_manager->add_fd(socketfd);
            conn->mtx.unlock();
            return;
        }
        
        // 记录获取HTTP请求错误的严重日志
        lg.logmessage(Fatal, "get http request error");
        send_queues_mtx.lock();
        if(send_queues.find(socketfd) == send_queues.end())
        {
            send_queues_mtx.unlock();
            return;
        }
        std::shared_ptr<Connection> conn = send_queues[socketfd];
        send_queues_mtx.unlock();
        
        conn->mtx.lock();
        // 设置连接头为关闭
        hr.headers["Connection"] = "Close";
        // 将错误的请求处理结果添加到输出缓冲区
        conn->out_buffer.push(process_bad_request(hr));
        // 标记连接需要关闭
        conn->should_close = true;
        // 从读取管理器移除socket
        read_manager->del_fd(socketfd);
        // 将socket添加到写入管理器
        write_manager->add_fd(socketfd);
        conn->mtx.unlock();
        return;
    }
    
    // 定义响应字符串
    std::string res;
    
    // 处理GET请求
    if (hr.method == "GET")
    {
        res = Http_Get_Handler(hr);
    }
    // 处理POST请求
    else if (hr.method == "POST")
    {
        res = Http_Post_Handler(hr);
    }
    // 处理不支持的请求方法
    else
    {
        lg.logmessage(warning, "unsupported method:%s", hr.method.c_str());
        send_queues_mtx.lock();
        if(send_queues.find(socketfd) == send_queues.end())
        {
            send_queues_mtx.unlock();
            return;
        }
        std::shared_ptr<Connection> conn = send_queues[socketfd];
        send_queues_mtx.unlock();
        
        conn->mtx.lock();
        // 设置连接头为关闭
        hr.headers["Connection"] = "Close";
        // 将错误的请求处理结果添加到输出缓冲区
        conn->out_buffer.push(process_bad_request(hr));
        // 标记连接需要关闭
        conn->should_close = true;
        // 将socket添加到写入管理器
        write_manager->add_fd(socketfd);
        conn->mtx.unlock();
        return;
    }
    
    // 加锁保护发送队列
    send_queues_mtx.lock();
    if(send_queues.find(socketfd) == send_queues.end())
    {
        send_queues_mtx.unlock();
        return;
    }
    std::shared_ptr<Connection> conn = send_queues[socketfd];
    send_queues_mtx.unlock();
    
    conn->mtx.lock();
    // 将响应结果添加到输出缓冲区
    conn->out_buffer.push(res);
    
    // 检查连接是否需要关闭
    if(hr.headers["Connection"] == "close" || hr.headers["Connection"] == "Close")
    {
        // 标记连接需要关闭
        conn->should_close = true;
    }
    else
    {
        // 连接保持活跃，检查输入缓冲区是否有剩余数据
        if(!conn->in_buffer.empty() && !conn->in_buffer.front().empty())
        {
            // 有剩余数据，需要继续处理
            write_manager->add_fd(socketfd);
            conn->mtx.unlock();
            
            // 创建新任务处理剩余数据
            Task t(socketfd, read_manager, write_manager, last_active_time);
            last_active_time[socketfd] = time(NULL);
            threadpool::getinstance().push(t);
            return;
        }
        else
        {
            // 没有剩余数据，继续监听读事件
            read_manager->add_fd(socketfd);
        }
    }
    
    // 将socket添加到写入管理器，准备发送响应
    write_manager->add_fd(socketfd);
    conn->mtx.unlock();
    
    // 更新最后活动时间
    last_active_time[socketfd] = time(NULL);
}

protocol.hpp:

#pragma once
#include"log.hpp"
#include<iostream>
#include<vector>
#include<string>
#include<sstream>
#include<unordered_map>
extern log lg;
class Http_Request
{
public:
/**
 *  反序列化HTTP头信息
 *  head 包含HTTP头信息的字符串
 * return 反序列化成功返回true，失败返回false
 */
    bool Deserialization(std::string& head)
    {
       
    // 初始化起始位置
        size_t start = 0;
    // 存储解析出的HTTP头信息行
        std::vector<std::string>_header;
    // 循环解析HTTP头信息的每一行
        while (true)
        {
            std::string line;      // 存储当前行的内容
        // 查找行结束符"\r\n"的位置
            size_t end = head.find("\r\n", start);
        // 如果找不到行结束符，说明格式错误
            if (end == std::string::npos)
            {
                return false;
            }
        // 提取当前行的内容
            line = head.substr(start, end - start);
        // 如果遇到空行，说明HTTP头部分结束
            if (line.empty())
            {
                break;
            }
        // 更新起始位置到下一行的开始
            start = end + 2;
        // 将当前行添加到头信息列表中
            _header.push_back(line);
        }
    // 检查至少有一行头信息（请求行）
        if (_header.size() < 1)
        {
            return false;
        }
    // 解析头信息的每一行（除了第一行请求行）
        for (size_t i = 1; i < _header.size(); i++)
        {
            std::string line = _header[i];    
        // 查找键值分隔符":"的位置
            ssize_t pos = line.find(":");
        // 如果找不到分隔符，说明格式错误
            if (pos == std::string::npos)
            {
                return false;
            }
        // 提取键
            std::string key = line.substr(0, pos);
        // 跳过分隔符后的空白字符
            size_t val_start = pos + 1;
            while (val_start < line.size() && std::isspace(line[val_start]))
            {
                val_start++;
            }
        // 提取值
            std::string value = line.substr(val_start);
        // 将键值对存储到headers映射中
            headers[key] = value;
        }
    // 解析第一行（请求行）
        std::string first_line = _header[0];
        std::stringstream ss(first_line);
    // 从请求行中提取方法、URL和HTTP版本
        ss >> method >> url >> http_version;
        return true;
    }
   
/**
 *  调试打印函数，用于输出HTTP请求的详细信息
 * 该函数会打印HTTP方法、URL、HTTP版本、头部信息和请求体文本
 */
    void debugprint()
    {
        
    // 打印一个空行，用于分隔不同部分的输出
        std::cout << std::endl;
    // 打印HTTP方法、URL和HTTP版本
        std::cout << method << " " << url << " " << http_version << std::endl;
    // 遍历并打印所有的HTTP头部信息
        for (auto it = headers.begin(); it != headers.end(); it++)
        {
            std::cout << it->first << ": " << it->second << std::endl;
        }
    // 打印一个空行，用于分隔头部和请求体
        std::cout << std::endl;
    // 打印请求体的文本内容
        std::cout << text << std::endl;
    // 打印一个空行，作为输出结束的标志
        std::cout << std::endl;
    }
public:
    std::unordered_map<std::string, std::string> headers;
    std::string text;
    std::string method;
    std::string url;
    std::string http_version;
};

daemon.hpp:

#pragma once
#include<unistd.h>
#include<string>
#include<cstdlib>
#include<signal.h>
#include<fcntl.h>
#include"log.hpp"
extern log lg;
std::string filename = "/dev/null";
std::string workingdirectory="/";
class Daemon
{
public:
        /**
 *  将当前进程转换为守护进程
 * 
 * 该函数执行创建守护进程的标准步骤：
 * 1. 忽略SIGCHLD和SIGHUP信号
 * 2. 创建子进程并退出父进程
 * 3. 创建新会话
 * 4. 重定向标准输入、输出和错误到指定文件
 * 5. 更改工作目录
 * 
 * 守护进程是在后台运行的特殊进程，独立于控制终端，
 * 通常用于执行系统服务或后台任务。
 */
void daemon()
{
    // 忽略SIGCHLD信号，防止子进程成为僵尸进程
    signal(SIGCHLD, SIG_IGN);
    // 忽略SIGHUP信号，防止终端关闭时影响守护进程
    signal(SIGHUP, SIG_IGN);
    
    // 创建子进程
    pid_t id = fork();
    if(id < 0)
    {
        // fork失败，记录错误日志并退出
        lg.logmessage(Fatal, "fork fail");
        exit(1);
    }
    else if(id > 0)
    {
        // 父进程退出，使子进程成为孤儿进程，被init进程收养
        exit(0);
    }
    
    // 创建新会话，使进程脱离控制终端
    int n = setsid();
    if(n < 0)
    {
        // 创建会话失败，记录错误日志并退出
        lg.logmessage(Fatal, "setsid fail");
        exit(1);
    }
    
    // 打开指定文件，用于重定向标准输入、输出和错误
    int fd = open(filename.c_str(), O_RDWR);
    if (fd < 0)
    {
        // 打开文件失败，记录错误日志并退出
        lg.logmessage(Fatal, "open dev/null fail");
        exit(1);
    }
    
    // 重定向标准输入到指定文件
    dup2(fd, STDIN_FILENO);
    // 重定向标准输出到指定文件
    dup2(fd, STDOUT_FILENO);
    // 重定向标准错误到指定文件
    dup2(fd, STDERR_FILENO);
    
    // 更改工作目录到指定目录
    if(chdir((workingdirectory).c_str()) < 0)
    {
        // 更改工作目录失败，记录错误日志并退出
        lg.logmessage(Fatal, "chdir fail");
        close(fd);
        exit(1);
    }
    
    // 记录成功转换为守护进程的日志
    lg.logmessage(info, "demonize successfully");
}

 };

log.hpp:

#pragma once
#include<iostream>
#include<string>
#include<time.h>
#include<unistd.h>
#include<stdarg.h>
#include<fcntl.h>
#define SIZE 1024
#define screen 0
#define File 1
#define ClassFile 2
enum
{
    info,
    debug,
    warning,
    Fatal,
};
class log
{
private:
    std::string memssage;
    int method;
public:
    log(int _method = screen)
        :method(_method)
    {

    }
    void logmessage(int leval, const char* format, ...)
    {
        const char* _leval;
        switch (leval)
        {
        case info:
            _leval = "info";
            break;
        case debug:
            _leval = "debug";
            break;
        case warning:
            _leval = "warning";
            break;
        case Fatal:
            _leval = "Fatal";
            break;
        default:
            _leval= "unknow";
            break;
        }
        char timebuffer[SIZE];
        time_t t = time(NULL);
        struct tm* localTime = localtime(&t);
        snprintf(timebuffer, SIZE, "[%d-%d-%d-%d:%d]", localTime->tm_year + 1900, localTime->tm_mon + 1, localTime->tm_mday, localTime->tm_hour, localTime->tm_min);
        char rightbuffer[SIZE];
        va_list arg;
        va_start(arg, format);
        vsnprintf(rightbuffer, SIZE, format, arg);
        va_end(arg);
        char finalbuffer[2 * SIZE];
        int len = snprintf(finalbuffer, sizeof(finalbuffer), "[%s]%s:%s\n", _leval, timebuffer, rightbuffer);
        int fd=-1;
        switch (method)
        {
        case screen:
            std::cout << finalbuffer;
            break;
        case File:
            fd = open("log.txt", O_WRONLY | O_CREAT | O_APPEND, 0666);
            if (fd >= 0)
            {
                write(fd, finalbuffer, len);
                close(fd);
            }
            break;
        case ClassFile:
            switch (leval)
            {
            case info:
                fd = open("log/info.txt", O_WRONLY | O_CREAT | O_APPEND, 0666);
                break;
            case debug:
                fd = open("log/debug.txt", O_WRONLY | O_CREAT | O_APPEND, 0666);
                break;
            case warning:
                fd = open("log/Warning.txt", O_WRONLY | O_CREAT | O_APPEND, 0666);
                break;
            case Fatal:
                fd = open("log/Fatal.txt", O_WRONLY | O_CREAT | O_APPEND, 0666);
                break;
            }
            if (fd >= 0)
            {
                write(fd, finalbuffer, len);
                close(fd);
            }
        }
    }
};
extern log lg;
           

运行截图：

压测报告显示，我们的 PollServer 展现出了极其剽悍的性能：在 100 并发 的极限施压下，跑出了 6113 QPS 的优异成绩。

更令人欣慰的是，失败请求数为 0。这意味着我们为解决 检查时刻与使用时刻的时间差 误杀漏洞所设计的‘原子抢夺’机制，以及为防止野指针崩溃所引入的 shared_ptr 防线，在 10,000 次实战冲击中展现出了钢铁般的稳定性。每一份字节的流动，都在我们的逻辑掌控之中。

结语

那么这就是本篇文章的全部内容，带你全面认识以及掌握IO多路复用以select以及Poll，并且手写了一个高并发的HTTP服务器，接下来我会更新epoll，我会持续更新，希望你能够多多关注，如果本文有帮助到你的话，还请三连加关注，你的支持就是我创作的最大动力！

转载自CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2301_80260194/article/details/159171726