1、什么是 epoll

​ epoll 是 Linux 特有的一种高效的多路复用 IO 机制，可以用来处理大规模并发连接。它在实现和使用上与 select、poll 有很大差异，它使用一组函数来完成任务，而不是单个函数。其次，epoll 通过内核级别的事件通知机制，可以高效地管理和监控大量的文件描述符，同时还能够避免一些传统多路复用方式如 select、poll 的一些需要遍历和状态切换等缺点，但是 epoll 需要使用一个额外的文件描述符，来唯一标识内核中的这个事件表！

​ 在 epoll 编程中，应用程序先创建一个 epoll 对象（使用 epoll_create()），然后将需要监控的文件描述符添加到 epoll 对象中（使用 epoll_ctl()），当文件描述符就绪时，内核会将该事件添加到 epoll 实例的事件表中。之后应用程序可以调用 epoll_wait() 来等待事件的发生，一旦事件到达，epoll_wait() 就会返回，并将就绪的文件描述符及其事件类型告诉应用程序，应用程序再根据事件类型来进行相应的 IO 操作。

​ epoll 也提供了两种工作模式：电平触发模式**Level Triggered（LT）和 边缘触发模式Edge Triggered（ET**）。

• 在 LT 模式下，当文件描述符处于就绪状态时，epoll_wait() 会一直返回可读、可写等事件，直到应用程序对其进行了 IO 操作，这是默认的工作模式。
• 在 ET 模式下，只有文件描述符状态改变时，才会触发事件通知，这样可以避免频繁的事件通知和处理，提高效率。

2、epoll 相对于 select 和 poll 有什么优势

相对于 select 和 poll，epoll 在性能和扩展性方面具有一些优势。以下是 epoll 相对于 select 和 poll 的几个主要优势：

1. 大规模并发支持：epoll 是 Linux 特有的多路复用机制，针对大规模并发连接的场景进行了优化。它使用事件驱动的方式，可以支持非常大数量的文件描述符，能够高效地处理成千上万的并发连接。
2. 高效的事件通知机制：epoll 使用基于事件的通知机制，当文件描述符就绪时，内核会主动通知应用程序，而不需要像 select 和 poll 那样需要轮询遍历整个描述符集合。这样可以避免不必要的遍历操作，提高了效率。
3. 内核管理的事件表：与 poll 相比，epoll 使用内核管理的事件表，无需将文件描述符集合从用户空间拷贝到内核空间，减少了内存拷贝开销，提高了速度。
4. 支持 Edge Triggered 模式：除了 Level Triggered 模式，epoll 还支持 ET 模式。在 ET 模式下，只有当文件描述符状态发生变化时才会触发事件通知，而不仅仅是处于就绪状态。这样可以减少事件通知的次数，降低开销。

​ 此外需要注意的是，虽然 epoll 的全称是 event poll，但它和 poll 基本是没有关系的！

​ 设想一个场景：有 100 万用户同时与一个进程保持着 TCP 连接，而每一时刻只有几十个或几百个 TCP 连接是活跃的(接收 TCP 包)，也就是说在每一时刻进程只需要处理这 100 万连接中的一小部分连接。那么，如何才能高效的处理这种场景呢？进程是否在每次询问操作系统收集有事件发生的 TCP 连接时，把这 100 万个连接告诉操作系统，然后由操作系统找出其中有事件发生的几百个连接呢？实际上，在 Linux2.4 版本以前，那时的 select 或者 poll 事件驱动方式是这样做的。

​ 这里有个非常明显的问题，即在某一时刻，进程收集有事件的连接时，其实这 100 万连接中的大部分都是没有事件发生的。因此如果每次收集事件时，都把 100 万连接的套接字传给操作系统(这首先是用户态内存到内核态内存的大量复制)，而由操作系统内核寻找这些连接上有没有未处理的事件，将会是巨大的资源浪费，然后 select 和 poll 就是这样做的，因此它们最多只能处理几千个并发连接。而 epoll 不这样做，它在 Linux 内核中申请了一个简易的文件系统，把原先的一个 select 或 poll 调用分成了 3 部分：

1. 调用 epoll_create 建立一个 epoll 对象(在 epoll 文件系统中给这个句柄分配资源)；
2. 调用 epoll_ctl 向 epoll 对象中添加这 100 万个连接的套接字；
3. 调用 epoll_wait 收集发生事件的连接。

​ 这样只需要在进程启动时建立 1 个 epoll 对象，并在需要的时候向它添加或删除连接就可以了，因此，在实际收集事件时，epoll_wait 的效率就会非常高，因为调用 epoll_wait 时并没有向它传递这 100 万个连接，内核也不需要去遍历全部的连接。

Ⅱ. epoll系统调用

​ 一个客户端和使用了 epoll 的服务端的交互过程如下图所示：

​ 我们先来介绍系统调用接口，再来介绍其底层原理！

1、epoll_create()

​ 上面我们提到过，epoll 需要使用一个额外的文件描述符，来唯一标识内核中的这个事件表，而这个文件描述符就要使用如下函数来创建：

#include <sys/epoll.h>
int epoll_create(int size);

• 参数 size 现在并不起作用，只是给内核一个提示，告诉内核这个事件表需要多大。
• 返回值：
  • 成功返回一个文件描述符，其唯一标识这个新创建的 epoll 模型也就是上面提到的内核中的事件表，它将用作其它 epoll 函数的第一个参数，以指定要访问的内核事件表！
  • 失败返回 -1，并且设置错误信息如 EINVAL、EMFILE 等。
• 需要注意的是，使用完之后必须要调用 close() 关闭该文件描述符！

参数size应该传多大比较合适

​ epoll_create() 函数中的参数 size 用于指定事件轮询器（eventpoll）内部维护的文件描述符表的初始大小。文件描述符表是一个哈希表，用于保存已经被注册到事件轮询器的文件描述符及其相应的事件项。

​ 一般来说，我们可以根据预计要监视的文件描述符数量来设置 size 参数。如果我们需要监视的文件描述符较少，可以将 size 设置为一个适当的值，例如 128 或 256。如果我们需要监视的文件描述符很多，则可以将 size 参数设得更大一些。但是，如果 size 设置得过大，可能会浪费内存空间，因为文件描述符表是在 eventpoll 中动态增长的。

​ 其中，对于 Linux 2.6.8 及之后的内核版本，size 参数的意义有了一些改变。在这些内核版本中，size 的值将被用于分配一块能够容纳 size 个 struct epoll_event 结构体的内存空间。struct epoll_event 结构体用于存储事件信息，因此，如果需要同时处理大量事件，应该将 size 设为相应的值，以确保能够为所有事件分配足够的内存。

​ 一般来说，以下是一些可能的参考值：

• 如果你只需要监视少量的文件描述符（例如不超过 100 个），你可以将 size 设置为 128 或 256。这样会为文件描述符表分配适当的空间，并且不会浪费太多内存。
• 如果你需要同时监视大量的文件描述符（例如超过 1000 个），那么你可能需要将 size 设置为更大的值，以确保能够容纳所有的文件描述符和相关的事件项。你可以根据实际情况尝试增加 size 的值，比如设置为 1024、2048 或更高。

​ 请注意，size 参数只是一个初始值，eventpoll 在运行时会动态调整文件描述符表的大小，以适应实际的需求。因此，即使稍微低估了 size 的值，eventpoll 也会自动进行扩展。

2、epoll_ctl()

​ 有了内核事件表及其文件描述符之后，我们就需要来操作它，通过以下函数来实现：

#include <sys/epoll.h>
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

• 参数：

  • epfd

    • 就是上面调用 epoll_create() 后的返回值。

  • op

    • 指定操作的类型，用三个宏来表示，如下所示：

      1. EPOLL_CTL_ADD：表示注册新的 fd 到 epfd 中。
      2. EPOLL_CTL_MOD：表示修改已经注册的 fd 的监听事件。
      3. EPOLL_CTL_DEL：表示从 epfd 中删除一个 fd。

      /* Valid opcodes ( "op" parameter ) to issue to epoll_ctl().  */
      #define EPOLL_CTL_ADD 1	/* Add a file descriptor to the interface.  */
      #define EPOLL_CTL_DEL 2	/* Remove a file descriptor from the interface.  */
      #define EPOLL_CTL_MOD 3	/* Change file descriptor epoll_event structure.  */

  • fd

    • 表示要监听的文件描述符。

  • event

    • 表示告诉内核需要监听该文件描述符上的哪些事件。它是一个 epoll_event 结构指针类型，其定义如下所示：

      struct epoll_event
      {
          uint32_t events;	/* epoll事件 */
          epoll_data_t data;	/* 用户数据 */
      } __EPOLL_PACKED;

    • 其中 events 成员描述事件类型，其支持的事件类型和 poll 基本相同，都是用一些宏，只不过在前面加上了 E 而已！但是 epoll 还有两个额外的事件类型：EPOLLET 和 EPOLLONESHOT，它们对于 epoll 的高效运作非常关键，这些宏如下所示：

      事件类型	功能
      EPOLLIN	表示对应的文件描述符可读（包括对端 socket 正常关闭）
      EPOLLOUT	表示对应的文件描述符可写
      EPOLLPRI	表示对应的文件描述符有紧急的数据可读（这里应该表示有带外数据的到来）
      EPOLLERR	表示对应的文件描述符发生错误
      EPOLLHUP	表示对应的文件描述符被挂断
      EPOLLET	将 epoll 设为边缘触发 ET 模式，这是相对于水平触发 LT 模式来说的
      EPOLLONESHOT	只监听一次事件，当监听完这次事件之后，如果还需要继续监听 这个 socket 的话，需要再次把这个 socket 加入到 epoll 队列里
      EPOLLRDHUP	表示读关闭、本端调用 shutdown、对端关闭连接（注意这里的 不能读的意思内核不能再往内核缓冲区中增加新的内容。已经在内核缓冲区中的内容，用户态依然能够读取到。）

    • 此外成员变量 data 用于存储用户数据，其类型 epoll_data_t 的定义如下：

      typedef union epoll_data    
      {
          void *ptr;
          int fd;
          uint32_t u32;
          uint64_t u64;
      } epoll_data_t;

      • 这是一个联合体，其中用的最多的成员是 fd，它指定事件所从属的目标文件描述符。
      • ptr 成员可用来指定与 fd 相关的用户数据。
        • 但由于 epoll_data_t 是一个联合体，我们不能同时使用 ptr 和 fd 成员，因此，如果要将文件描述符和用户数据关联起来，实现快速的数据访问，只能使用其它手段，比如放弃 epoll_data_t 的 fd 成员，而在 ptr 指向的用户数据中包含 fd。

• 返回值：

  • 成功返回 0，失败则返回 -1 并且设置 errno。

3、epoll_wait()

​ 上面两个函数调用完之后，只是我们告诉了操作系统需要关心哪个文件描述符的何种事件，但是我们还需要知道操作系统告诉我们哪些事件就绪了，就得使用 下面这个函数来实现！

​ 它的作用就是在一段超时时间内等待一组文件描述符上的事件，也就是收集在 epoll 监控的事件表中已经发送的事件，其函数原型如下：

#include <sys/epoll.h>
int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

• 参数：
  • epfd
    • 就是上面调用 epoll_create() 后的返回值。
  • events
    • 如果 epoll_wait() 函数检测到事件，就将所有就绪的事件从内核事件表（即 epfd 指定的）中复制到该参数 events 数组中，所以它是一个输出型参数！
    • 要注意的是，events 数组不可以为空，因为内核只负责把数据复制到这个 events 数组中，不会去帮助我们在用户态中分配内存。
  • maxevents
    • 该参数告诉内核用户空间事件数组的大小，即最多可以返回多少个就绪的事件。这是用于底层方便遍历和排序 events 数组的。要注意的是这个 maxevents 的值不能大于创建 epoll_create() 时的 size。
    • 如果实际就绪事件的数量超过了 maxevents，那么只会填充 maxevents 个事件到数组中，并返回 maxevents 作为函数的返回值。剩余的事件将保持在内核中等待下次的 epoll_wait() 调用。
  • timeout
    • 该参数和 poll 函数的 timeout 参数含义是一样的！表示函数的超时时间，单位是毫秒。
      • timeout = -1，则 poll 函数将永远阻塞，直到某个事件发生。
      • timeout = 0，则 poll 函数将立刻返回，也就是非阻塞等待。
      • timeout 为具体时间，则如果在该时间段内没有就绪事件的话，则会超时返回 0。
    • 一般如果网络主循环是单独的线程的话，可以用 -1 进行阻塞式来等待，这样可以保证一些效率；如果是和主逻辑在同一个线程的话，则可以用 0 进行非阻塞等待来保证主循环的效率。
• 返回值：
  • 小于**0**，表示 epoll 函数出错。
  • 等于**0**，表示 epoll 函数等待超时。
  • 大于**0**，表示 epoll 函数由于监听的文件描述符就绪而返回，此时返回值表示返回就绪的文件描述符的个数。

Ⅲ. epoll底层原理

💥**深入理解 Linux 的 epoll 机制**

epoll源码解析翻译------说使用了mmap的都是骗子

epoll详解

​ 某一进程调用 epoll_create 方法时，Linux 内核会创建一个 eventpoll 结构体，这个结构体中有两个成员与 epoll 的使用方式密切相关，如下所示：（这里只列出部分重要字段）

struct eventpoll 
{
    /* 红黑树的根节点，这棵树中存储着所有添加到epoll中的事件，也就是这个epoll监控的事件 */
　　struct rb_root rbr;
    
    /* 双向链表，表示已就绪事件的列表，列表中保存着将要通过epoll_wait()返回给用户的、满足条件的事件 */
    struct list_head rdllist; 
 
    // ……
};

​ 我们在调用 epoll_create() 时，内核除了帮我们在 epoll 文件系统里创建了个 file 结构、在内核 cache 里建了个红黑树用于存储所有添加到 epoll 中的事件，还会再创建一个 rdllist 双向链表，用于存储准备就绪的事件。

​ 当 epoll_wait() 调用时，仅仅观察这个 rdllist 双向链表里有没有数据即可。有数据就返回，没有数据就 sleep，等到 timeout 时间到后即使链表没数据也返回。

​ 所以 epoll_wait() 不会去遍历所有事件判断是否就绪，这是非常高效的。

​ 而所有添加到 epoll 中的事件都会与设备（比如网卡）驱动程序建立回调关系，也就是说相应事件的发生时会调用这里的回调方法。这个回调方法在内核中叫做 ep_poll_callback，它会把这样的事件放到上面的 rdllist 双向链表中。

​ 在 epoll 中对于每一个事件都会建立一个 epitem 结构体，如下所示：

// 这个结构体包含每一个事件（即红黑树的节点）所对应的信息
struct epitem 
{   
    /* 在eventpoll结构体中形成链表 */
    struct epitem *next; 
    
    /* 通过事件时间排列的红黑树节点 */
　　struct rb_node rbn;
    
　　/* 双向链表的头节点，指向eventpoll结构体中rdllist的头节点 */
　　struct list_head rdllink;
    
　　/* 存放的是指向与此eventpoll相关联的文件的指针，以及该事件节点的文件描述符fd（节点本身就是一个文件） */
　　struct epoll_filefd ffd;
    
　　/* 指向其所属的eventepoll结构体对象 */
　　struct eventpoll *ep;
    
　　/* 用户期待的事件，通过epoll_ctl中的宏来设置 */
　　struct epoll_event event;
    
    /* 文件指针，指向该注册事件所属的文件 */
    struct file *file; 
};
 
// 下面是上面结构体中一些结构体的定义
struct list_head {
	struct list_head *next, *prev;
};
struct epoll_filefd {
	struct file *file;
	int fd;
};
struct epoll_event {
	__u32 events;
	__u64 data;
} EPOLL_PACKED;

​ 当调用 epoll_wait() 检查是否有发生事件的连接时，只是检查 eventpoll 对象中的 rdllist 双向链表是否有 epitem 元素而已，如果 rdllist 链表不为空，则这里的事件复制到用户态内存中（可以使用共享内存提高效率），同时将事件数量返回给用户。因此 epoll_wait() 效率非常高。

​ 此外 epoll_ctl（） 在向 epoll 对象中添加、修改、删除事件时，从 rbr 红黑树中查找事件也非常快，也就是说 epoll 是非常高效的，它可以轻易地处理百万级别的并发连接！

​ 并且上图提到，链表 rdllist 中的节点就是红黑树中已经就绪的节点，我们习惯性的以为一个对象只能被一个数据结构使用，其实一个对象是可以处于不同的数据结构中的，只需要在这个对象中存在不同数据结构的变量即可！

​ 比如每个红黑树节点 epitem 结构体中的 rbn 变量就是为了使得节点有在红黑树中存放的能力，而 next 变量，其实是为了让当前节点有在链表中存放的能力！

小总结

一颗红黑树，一张准备就绪句柄链表，少量的内核 cache，就帮我们解决了大并发下的 socket 处理问题。

• 执行 epoll_create() 时，创建了红黑树和就绪链表；
• 执行 epoll_ctl() 时，如果增加 socket 句柄，则检查在红黑树中是否存在，如果存在立即返回，不存在则添加到树干上，然后向内核注册回调函数，用于当中断事件来临时向准备就绪链表中插入数据；
• 执行 epoll_wait() 时返回准备就绪链表 rdllist 里的数据即可。

问题：链表 rdllist 是如何知道哪些事件就绪的呢❓❓❓

​ 这个问题，我们需要从硬件说起（硬件知识这里简略介绍），我们都知道冯诺依曼体系，控制器和输入输出设备之间是有控制信号的，当输入设备比如网卡，输入了信息之后，就会通过信号触发中断，此时控制器中的某个引脚会被点亮，而其中的电路会将其寄存器的值设置为输入设备的值，比如说将网卡对应的接口值为 6，那么寄存器中存放的就是 6。

​ 那么当内存去返回寄存器拿到 6 的时候，会去访问中断向量表（一种内存中的数据结构，其实就是一个函数指针数组），然后通过中断向量表中的函数指针调用驱动方法，将数据从外设拷贝到内存中的操作系统内部！

​ 接着网卡的数据拷贝到操作系统内部后，就要贯穿网络协议栈如链路层、网络层、传输层，最后到达应用层之后，会将数据拷贝放到应用层的缓冲区，就是我们学过的 struct file 结构体中。

​ 重点来了，struct file 中有一个 void* 类型的指针 private_data，它指向一个回调函数，当 struct file 收到数据的时候，就会去调用 private_data，就相当于调用了回调函数，这个回调方法在内核中叫做 ep_poll_callback，而回调函数的作用就是修改 epitem 结构体也就是红黑树节点中的 next 指针，使其节点置于 rdllist 中，也就是变成了就绪事件了！

重新理解 epoll 接口

​ 我们将上面的回调机制以及前面的红黑树、链表结合起来，统称为一个 epoll 模型！下面让我们来重新理解一下 epoll 接口的调用，以及其底层原理的联系！

1. 首先就是调用 epoll_create()，创建一个 epoll 模型（包括空的红黑树、空的链表等等结构），而这整个 epoll 模型，其实就由一个 struct file 结构体来管理，所以返回值就是一个文件描述符 epfd。
2. 然后就是调用 epoll_ctl()，根据上面得到的 epoll 模型的文件描述符 epfd，再根据需要选择增删改操作，将感兴趣的事件以及要监听的文件描述符传入到 epoll_ctl() 中，其底层其实就是红黑树的插入、删除、修改操作，以及一些事件字段的设置罢了！
3. 最后就是调用 epoll_wait()，这个函数只关心 rdllist 就绪链表。其底层原理就是在我们传入的 timeout 时间后，去访问 rdllist 就绪链表，看看有没有事件已经就绪了，没有的话则返回继续 timeout 事件的阻塞；如果有的话则将就绪链表中的内容通过返回值、输出型参数以及文件缓冲区拷贝，反馈给用户！

​ 所以我们也能看出来，epoll 的底层其实就不需要去遍历所有的事件判断是否就绪，只需要通过 O(1) 时间复杂度，看看 rdllist 就绪链表中是否存在节点就能知道有没有事件就绪了，这是非常高效的！

​ 并且对于 epoll 的增删查是一个 O(logn) 的时间复杂度，也是非常优秀的！

epoll_wait() 的细节

​ 在 epoll_wait() 函数中，我们需要传入一个 maxevents，需要该参数的原因之一是用于指定用户空间事件数组的大小，即限制最多可以返回多少个就绪的事件。如果实际就绪事件的数量超过了 maxevents，那么只会填充 maxevents 个事件到数组中，并返回 maxevents 作为函数的返回值。剩余的事件将保持在内核中等待下次的 epoll_wait() 调用。

​ 所以正确设置 maxevents 参数非常重要。如果用户空间事件数组的大小不足以容纳所有就绪的事件，可能会导致事件丢失。如果将 maxevents 设置得过大，则可能会造成性能浪费。一般来说，建议根据实际需要预估需要处理的事件数量，并为 maxevents 分配足够的空间，以确保能够正确地处理所有就绪的事件。

​ 此外，epoll_wait() 返回就绪的文件描述符数量，之后我们从 0 到 maxevents 遍历传入的参数 epoll_event 数组，其中已经就绪的事件，内核已经帮我们做好内部的排序优化了。

​ 举一个例子，假设 epoll_event 队列中有 1000 个文件描述符，第一次调用 epoll_wait() 返回 5，那么表示队列前五个元素就绪了，如果不处理第三个就绪事件，其他的都处理。第二次调用 epoll_wait() 返回了 8，那么这 8 个 epoll_event 也是按顺序从 0 开始排列，而不是从 3 开始排列的。

​ 所以认为 epoll_wait() 这个函数做了内部的优化排序，返回给用户按顺序排好的 epoll_event 数组。

Ⅳ. epoll 的优点

• 接口使用方便：虽然 epoll 的使用拆分成了三个函数，但是反而使用起来更方便高效，因为不需要每次循环都设置关注的文件描述符，也做到了输入输出参数分离开。
• 数据拷贝轻量：只在需要的时候调用 EPOLL_CTL_ADD 将文件描述符结构拷贝到内核中，这个操作并不频繁，而 select/poll 都是每次循环重复地进行拷贝，消耗了资源。
• 事件回调机制：避免使用遍历，而是使用回调函数的方式，将就绪的文件描述符结构加入到就绪队列中，当 epoll_wait() 返回直接访问就绪队列就知道哪些文件描述符就绪，这个操作时间复杂度 O(1)，即使文件描述符数目很多，效率也不会受到影响。
• 文件描述符数量多：和 poll 一样，文件描述符数目都是 65535 个！

注意

​ 网上有些博客说，epoll 中使用了内存映射机制：即内核直接将就绪队列通过 mmap 的方式映射到用户态，避免了拷贝内存这样的额外性能开销。

​ 这种说法是不准确的，我们定义的 struct epoll_event 是我们在用户空间中分配好的内存，势必还是需要将内核的数据拷贝到这个用户空间的内存中的，也就是说这个过程是会涉及到数据拷贝和上下文切换等操作的。

Ⅴ. select/poll/epoll之间的比较

​ 首先 select 的参数类型 fd_set 没有将文件描述符和事件做到绑定，它仅仅是一个文件描述符的集合，因此使用 select 的时候需要提供三个 fd_set 类型的参数来分别关心可读、可写、异常事件，一方面这使得 select 无法处理更多类型的事件（处理更多的事件就得使用更多的 fd_set），另一方面就是每一个 fd_set 又同时作为输入和输出参数，所以 fd_set 的内容是会被修改的，这导致我们每次在调用 select 之前都需要重新设置每一个 fd_set，从而导致了接口使用的不方便！此外还有一个问题就是因为 select 使用的时候需要维护一个数组，其中维护的是就绪事件的文件描述符，可是因为这个数组中的元素是不连续，这导致我们每次都需要遍历整个数组去看看哪些事件就绪，时间复杂度为 O(n)，如果说一个数组是 1024 个空间，但没有事件就绪，我们还是需要去遍历，这就妥妥做了大量的无用功！

​ 而 poll 就对 select 的参数进行了优化，使用一个 pollfd 结构体，将文件描述符和事件绑定在一个结构中，用户可以通过按位与/按位或的操作来设置/获取事件，并且 pollfd 中是有两个变量，分别是 events 和 revents，它们分别代表输入和输出，这也就说明了 poll 调用无需每次调用前都去设置事件的集合，因为它们是互相独立的！但 poll 存在和 select 一样的问题，就是我们需要用一个数组来维护这些 pollfd 结构体，那么势必就需要每次都去遍历这个数组看看哪些事件就绪了，时间复杂度还是 O(n)，没有得到任何的优化。

​ 此时 epoll 就诞生了，它采用和 select、poll 完全不同的机制来达到高效率。首先调用 epoll_create() 创建一个 epoll 模型，在内核中维护一棵红黑树以及一个就绪链表（也可以认为是队列），当我们需要操作关心事件如增删改的时候，就调用 epoll_ctl() 在红黑树中操作这些事件（具体如何找到这个红黑树，其实就是靠 eventpoll 结构体中的 rbr 成员，这个我们上面讲过），因为每个红黑树的节点都是独立的，所以不需要我们每次调用后去重新设置属性！然后我们只需要调用 epoll_wait() 函数，让其在特定时间去看看就绪链表中是否存在节点，如果说不存在节点的话就直接返回了，这是 O(1) 时间复杂度，非常高效，而如果存在节点的话，则遍历这些节点，将其拷贝到用户传入的就绪事件数组中即可！

​ 从实现原理上来说，select 和 poll 都是采用轮询的方式，即每次调用都要扫描整个文件描述符集合，判断是否有就绪事件，时间复杂度为 O(n)。而 epoll_wait 采用的是回调机制。当有事件就绪的时候，会触发回调函数，该回调函数会将其插入到就绪队列中去，而内核最后在适当时机将该就绪队列中的内容拷贝到用户空间即可，因此 epoll_wait 不需要轮询整个文件描述符集合，所以时间复杂度为 O(1)。

​ 但要注意的是，当活动连接比较多的时候，epoll_wait 的效率未必比 select 和 poll 高，因为此时回调函数被触发得过于频繁，所以 epoll 适合于连接数量多，但是活动连接较少的情况！

​ 还有，poll 和 epoll_wait 分别使用 nfds 和 maxevents 参数来指定最多监听多少个文件描述符和事件。这两个数值都能达到系统允许打开的最大文件描述符的数目，即 65535 个文件描述符，这可以通过 cat /proc/sys/fs/file-max 指令来查看当前系统的最大文件描述符个数。而 select 允许监听的最大文件描述符是有限制的，因为它的参数类型 fd_set 本质是一个位图，并且是系统固定的，这就导致了上限的问题！

​ 此外，select 和 poll 只能工作在相对低效的 LT 模式，而 epoll 是可以设置工作在 ET 高效模式的，并且 epoll 还支持 EPOLLONESHOT 事件，该事件能大大减少事件被触发的次数！

Ⅵ. 简单epoll代码编写

​ 这里我们直接使用前面 select/poll 中的几个头文件，如 err.hpp、log.hpp、sock.hpp，这几个都是一样的，这里就不再赘述了，核心代码都在服务器头文件中，也是我们的重点！

main.cc

​ 首先是主函数 main.cc，还是一样，只是修改了一下命名空间以及变量名而已：

#include "epoll_server.hpp"
#include <memory>
using namespace std;
using namespace epoll_space;
 
static void Usage(const string& proc)
{
    cerr << "\nUsage:\n\t" << proc << " port\n\n"; 
}
 
int main(int argc, char* argv[])
{
    if(argc != 2)
    {
        Usage(argv[0]);
        exit(USAGE_ERR);
    }
    unique_ptr<epoll_server> svr(new epoll_server(atoi(argv[1])));
    svr->run();
    return 0;
}

epoll_server.hpp

​ 接下来就是重点，服务器的头文件，先给出主体框架：

#pragma once
#include <iostream>
#include <sys/epoll.h>
#include "sock.hpp"
 
namespace epoll_space
{
    const static int epoll_size = 128;         // 在当前linux版本中，表示epoll_event的默认个数
    const static int defalult_num = 64;        // 就绪数组的默认大小
    const static uint16_t default_port = 8080; // 服务器默认端口号
    const static int buffer_size = 1024;       // 缓冲区大小
 
    class epoll_server
    {
    private:
        uint16_t _port;
        int _listensock;
        int _epfd;                   // epoll模型的文件描述符
        int _num;                    // 就绪事件数组的大小
        struct epoll_event* _events; // 就绪事件数组
 
    public:
        epoll_server(uint16_t port = default_port, int num = defalult_num) // 构造函数
            : _port(port), _num(num)
        {}
        ~epoll_server() // 析构函数
        {}
        void run() // 服务器启动函数
        {}
        void handler(int ready_num) // 处理业务主函数
        {}
    private:
        void Accepter() // 获取新链接函数
        {}
        void Receiver(int fd) // 读取数据函数
        {}
    };
}

构造函数

​ 这里我们就不像之前那样写 init() 函数初始化了，直接放在构造函数中一起做初始化操作！

​ 主要的步骤就是完成监听之后，创建一个 epoll 模型，然后添加最开始的 _listensock 到 epoll 模型中，作为监听描述符，所以要关心其可读事件。接着还有一步就是初始化一个就绪事件数组 _events，因为我们后面调用 epoll_wait() 的时候，需要有这样一个数组来存放就绪事件，而这个数组大小，这里设为 64。

const static int epoll_size = 128;         // 在当前linux版本中，表示epoll_event的默认个数
const static int defalult_num = 64;        // 就绪数组的默认大小
const static uint16_t default_port = 8080; // 服务器默认端口号
const static int buffer_size = 1024;       // 缓冲区大小
 
epoll_server(uint16_t port = default_port, int num = defalult_num)
    : _port(port), _num(num)
{
    // 1. 完成套接字基本流程
    _listensock = sock::Socket();
    sock::Bind(_listensock, _port);
    sock::Listen(_listensock);
 
    // 2. 创建epoll模型
    _epfd = epoll_create(epoll_size);
    if(_epfd == -1)
    {
        logMessage(Level::FATAL, "create epoll error");
        exit(EPOLL_CREATE_ERR);
    }
    logMessage(Level::NORMAL, "create epoll success");
 
    // 3. 添加_listensock到epoll中，并且关心可读事件
    struct epoll_event in;
    in.data.fd = _listensock;
    in.events = EPOLLIN;
    int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &in);
    if(ret == -1)
    {
        logMessage(Level::ERROR, "add _listensock error");
        exit(EPOLL_CTL_ERR);
    }
    logMessage(Level::NORMAL, "add _listensock success");
 
    // 4. 开辟就绪事件数组的空间
    _events = new struct epoll_event[_num];
    if(_events == nullptr)
    {
        logMessage(Level::ERROR, "new epoll_events error");
        exit(NEW_EVENTS_ERR);
    }
    logMessage(Level::NORMAL, "init server success");
}

析构函数

​ 析构函数自然不用说，就是一些文件描述符的关闭以及内存释放！

~epoll_server()
{
    if(_listensock)
        close(_listensock);
    if(_epfd)
        close(_epfd);
    if(_events)
        delete[] _events;
}

启动服务器 run()

​ 在这个函数中，我们要调用 epoll_wait() 函数进行事件的超时查询，判断是否有就绪事件，有的话就交给 handler() 去处理！可以看到这个代码写起来是非常简洁的！

void run()
{
    while(true)
    {
        // 5. 进行就绪事件的查询等待
        int n = epoll_wait(_epfd, _events, _num, 2000);
        if(n == -1)
            logMessage(Level::ERROR, "epoll_wait error, code: %d, strerror: %s", errno, strerror(errno));
        else if(n == 0)
            logMessage(Level::NORMAL, "timeout...");
        else
        {
            logMessage(Level::NORMAL, "一共有%d个事件就绪", n);
            handler(n);
        }
        sleep(1);
    }
}

处理业务主函数 handler()

​ 这里我们只关心获取新连接以及读取的事件，其它的事件比如可写、异常事件，等后面将 reactor 的时候一起写，因为现在读写其实是需要有自定义协议的，这里只是一个 demo 级别的样例！

void handler(int ready_num)
{
    for(int i = 0; i < ready_num; ++i)
    {
        // 这里遍历的事件都是就绪的！
        // 1. 首先获取事件类型以及文件描述符
        uint32_t event = _events[i].events;
        int fd = _events[i].data.fd;
 
        // 2. 根据不同的事件类型以及文件描述符来做不同的业务处理
        if(fd == _listensock && (event & EPOLLIN))
        {
            // 如果是_listensock并且是可读事件，则获取新链接
            Accepter();
        }
        else if(event & EPOLLIN)
        {
            // 如果是普通文件描述符的可读事件，我们就接收信息
            Receiver(fd);
        }
        else
        {
            // 如果是普通文件描述符的可写事件等等，我们这里不做处理，等后面将reactor的时候一起写
        }
    }
}

获取新连接函数 Accepter()

​ 获取新连接其实这个操作就是调用封装好的 Accept() 函数，然后将该新连接交给 epoll 模型管理即可！

void Accepter()
{
    // 1. 获取新连接
    std::string clientip;
    uint16_t clientport;
    int newfd = sock::Accept(_listensock, &clientip, &clientport);
 
    // 2. 将该新连接交给epoll模型管理
    struct epoll_event in;
    in.data.fd = newfd;
    in.events = EPOLLIN;
    int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, newfd, &in);
    if(ret == -1)
    {
        logMessage(Level::ERROR, "epoll_ctl error, fd: %d, errno: %d, why: %s", newfd, errno, strerror(errno));
        return;
    }
    logMessage(Level::NORMAL, "epoll_ctl success, fd: %d", newfd);
}

接收数据函数 Receiver()

​ 这里也是一个 demo 级别的读取样例，因为现在读写其实是需要有自定义协议的！

void Receiver(int fd)
{
    char buffer[1024];
    ssize_t n = recv(fd, buffer, sizeof(buffer) - 1, 0);
 
    if(n == -1)
    {
        // 读取发生错误，将该事件从epoll模型中去除，然后关闭
        epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
        close(fd);
        logMessage(Level::ERROR, "receive error, fd: %d, errno: %d, why: %s", fd, errno, strerror(errno));
    }
    else if(n == 0)
    {
        // 读到0表示请求断开连接，也是一样将该事件从epoll模型中去除，然后关闭
        epoll_ctl(_epfd, EPOLL_CTL_DEL, fd, nullptr);
        close(fd);
        logMessage(Level::NORMAL, "receive close, fd: %d", fd);
    }
    else
    {
        buffer[n] = 0;
        logMessage(Level::NORMAL, "receive success, fd: %d, 内容: %s", fd, buffer);
 
        // 这里做简单的回响处理
        std::string response = "响应: " + std::string(buffer);
        send(fd, response.c_str(), response.size(), 0);
    }
}

Ⅶ. epoll的两种触发模式💥

1、两种模式介绍

​ 首先我们要明白什么叫做事件就绪？事件就绪就是底层的 IO 条件满足之后，可以进行某种 IO 行为，就叫做事件就绪。而我们上面包括前面学过的 select/poll/epoll 系统调用其实都要等待事件就绪，那么就得有通知机制，而通知机制也是有策略的，主要体现在 epoll 上，因为 select/poll 都是工作在 LT 模式下的！

​ epoll 对文件描述符的操作有两种模式：水平触发 LT 模式（Level Trigger）和边缘触发 ET 模式（Edge Trigger）。

​ 其中 LT 模式就是 epoll 默认的工作模式，这种模式下 epoll 相当于一个效率较高的 poll，而当我们往 epoll 模型中注册一个文件描述符上的 EPOLLET 事件时，epoll 将以 ET 模式来操作该文件描述符，ET 模式是 epoll 的高效工作模式！

​ 我们先来讲个小故事，一个买家小王，在网上买东西后，快递最后交给了快递小哥张三，张三到了小王家楼下之后，打电话告诉小王下楼取快递，可此时小王正忙着和朋友打游戏，所以就让张三在楼下等着……然后张三就在楼下等着，每隔一会就打电话叫小王下楼拿快递，打了两个、三个……电话，最后小王终于下来了，把快递拿回家，而张三也就完成任务就回去了！这是第一种情况，也就是 LT 工作模式。

​ 假设第二种情况，还是买家小王买东西后，但是这次快递交给了快递小哥李四，李四脾气很臭，到了小王家楼下之后，打电话告诉小王快点下来拿快递，并且说了一句，如果还不快点下来，李四就直接走了，到时候快递不见了就不要怪他，说完之后就挂电话了，然后也没有继续打电话催小王，小王听了之后非常的无奈，怕自己的快递等会真被丢了，就马上下楼拿快递。这是第二种情况，也就是 ET 工作模式。

​ 上面的例子中，小王就是上层用户，快递就是数据，而快递小哥就是内核，从上面张三和李四的表现可以看出来，李四这种模式可以催促小王快速的下楼拿快递，并且还不需要打很多的电话，这是非常高效和节省资源的！这种模式对应的就是 ET 模式，我们也能看出来，打电话给小王的动作，其实本质就对应着内核通知用户事情已经就绪了！

所以可以总结一下两种模式的区别：

• 水平触发 LT 模式：
  • 只要底层还有数据没读完，epoll_wait() 就会一直通知用户层（通过回调机制实现）要读取数据。
  • LT 模式支持文件描述符的阻塞读写和非阻塞读写。
  • 当 epoll 检测到 socket 上事件就绪的时候，可以不立刻进行处理，或者只处理一部分，但是由于只读了 1K 数据，缓冲区中还剩 1K 数据，在第二次调用 epoll_wait 时，epoll_wait 仍然会立刻返回并通知 socket 读事件就绪，直到缓冲区上所有的数据都被处理完，epoll_wait 才不会立刻返回。传统的 select/poll 都是这种模型的代表。
• 边缘触发 ET 模式：
  • 即使底层还有数据没读完，epoll 也不会再通知用户，除非底层的数据变化的时候，才会再一次通知用户。
  • ET 模式只支持非阻塞的读写。
  • 在它检测到有 IO 事件时，通过 epoll_wait 调用会得到有事件通知的文件描述符，对于每一个被通知的文件描述符，如可读，则必须将该文件描述符一直读到空，让 errno 返回 EAGAIN 为止，否则下次的 epoll_wait 不会返回余下的数据，会丢掉事件。

​ 两种模式最直观的现象，其实我们见过了其中的 LT 模式，就是当我们在写 epoll 包括之前的 select/poll 代码的时候，如果内核通知我们去获取新连接或者读取内容的时候，我们没去获取或者读取的话，就会不停的打印就绪事件的消息，这就是 LT 模式不停通知的效果！

​ 而设置 ET 模式其实很简单，只需要在传入的事件中添加上 EPOLLET 事件即可。

​ 这里我们演示在 _listensock 文件描述符中添加 ET 模式，并且将获取新连接的操作去掉，如下所示：

// 3. 添加_listensock到epoll中，并且关心可读事件
struct epoll_event in;
in.data.fd = _listensock;
in.events = EPOLLIN | EPOLLET; // 同时设为ET模式
int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, _listensock, &in);

​ 然后来看看它和 LT 模式的运行区别：

2、理解 ET 模式以及非阻塞文件描述符的重要性

​ 从上面的介绍可见，ET 模式在很大程度上降低了同一个 epoll 事件被重复触发的次数，因此效率要比 LT 模式高，但是效率高的原因不仅仅如此！下面我们从头来梳理一下整个思路。

​ 首先 ET 模式在事件就绪时候通知用户一次，而往后不再重复通知，只有当该事件的数据变化的时候比如数据增多了，才会再次通知用户，此时如果在代码中没有尽可能的将所有可读数据读取上来的话，会发生以下情况：

1. 数据丢失：在 ET 模式下，当有新数据到达时，只会触发一次事件通知。如果应用程序没有及时读取并处理所有的可读数据，那么下一次事件通知到来时，之前未读取的数据将会丢失，因为内核只关心是否有新的数据到达，而不会保存之前未读取的数据。
2. 堵塞：如果应用程序没有完全读取缓冲区中的数据，而且没有设置为非阻塞模式，那么下一次尝试读取时可能会因为缓冲区已满而被阻塞住，进而影响其他事件的处理。
3. 资源浪费：如果应用程序反复触发可读事件，并尝试读取数据，但每次读取都不完整，会造成 CPU 和内存资源的浪费。这是因为不断触发可读事件会导致应用程序在忙等待数据上花费大量时间，而实际上却没有有效地处理数据。

​ 需要注意的是，即使使用了 ET 模式并正确处理了数据的读取，仍然存在一些特殊情况下可能导致数据丢失，例如网络异常或者连接断开等。因此，开发应用程序时应该考虑到这些情况，并采取适当的处理方式，如重新建立连接等，以保证数据的完整性和可靠性。

​ 因为有以上的这些问题，所以也就自然的倒逼我们要尽量在代码中将就绪的可读数据都读取上来，那么可以采取以下措施来解决这些问题：

1. 以循环的方式读取数据，直到返回的读取结果表明没有更多数据可读，此时要注意的是需要将文件描述符设置为非阻塞模式，以便不会因为缓冲区已满而被阻塞住。
2. 使用合适大小的缓冲区来容纳接收的数据，避免频繁的内存分配和释放。

为什么 ET 模式必须将文件描述符设置为非阻塞模式❓❓❓

​ 如果 ET 模式不是非阻塞的，那这个一直读或一直写势必会在最后一次阻塞。因为如果使用阻塞式 IO 进行读取操作，在读取完所有可用数据后，如果没有更多的数据到达，读取函数将会阻塞等待数据的到达。由于 ET 模式只在状态发生变化时触发事件通知，所以没有新数据到达时不会再次触发事件通知，导致读取操作阻塞在最后一次读取上。

​ 同样地，如果使用阻塞式 IO 进行写入操作，在发送完所有数据后，如果无法立即发送更多数据（例如发送缓冲区已满），写入函数将会阻塞等待数据的发送。由于 ET 模式只在状态发生变化时触发事件通知，所以无法发送更多数据时不会再次触发事件通知，导致写入操作阻塞在最后一次写入上。

​ 所以要设置为非阻塞模式，可以使用 fcntl() 函数进行设置！而 LT 模式既可以是阻塞式读写也可以是非阻塞式读写。

​ 下面我们将前面的代码修改一下，将文件描述符设为非阻塞模式，并且将文件描述符设为 ET 模式，所以我们可以考虑将添加到 epoll 模型的操作和设置非阻塞的操作，设置成接口来使用，改动的地方如下所示：

public:
    epoll_server(uint16_t port = default_port, int num = defalult_num)
        : _port(port), _num(num)
    {
        // 1. 完成套接字基本流程
        _listensock = sock::Socket();
        sock::Bind(_listensock, _port);
        sock::Listen(_listensock);
 
        // 2. 创建epoll模型
        _epfd = epoll_create(epoll_size);
        if(_epfd == -1)
        {
            logMessage(Level::FATAL, "create epoll error");
            exit(EPOLL_CREATE_ERR);
        }
        logMessage(Level::NORMAL, "create epoll success");
 
        // 3. 直接调用函数，添加_listensock到epoll中，并且关心可读事件和启动ET模式
        AddFD(_listensock, true);
 
        // 4. 开辟就绪事件数组的空间
        _events = new struct epoll_event[_num];
        if(_events == nullptr)
        {
            logMessage(Level::ERROR, "new epoll_events error");
            exit(NEW_EVENTS_ERR);
        }
        logMessage(Level::NORMAL, "init server success");
    }
private:
    // 设置文件描述符为阻塞模式
    void SetNonBlocking(int fd)
    {
        // 1. 先获取文件描述符标记
        int old_option = fcntl(fd, F_GETFL);
 
        // 2. 设为非阻塞模式
        int new_option =old_option | O_NONBLOCK;
        fcntl(fd, new_option);
    }
 
    // 添加文件描述符到epoll模型中，如果有必要的话还可以设置为ET模式
    void AddFD(int fd, bool enable_ET)
    {
        // 1. 设置为可读事件
        struct epoll_event in;
        in.data.fd = fd;
        in.events = EPOLLIN;
 
        // 2. 看看是否需要设置为ET模式
        if(enable_ET)
            in.events |= EPOLLET;
 
        // 3. 添加到epoll模型中
        int ret = epoll_ctl(_epfd, EPOLL_CTL_ADD, fd, &in);
        if(ret == -1)
        {
            logMessage(Level::ERROR, "epoll_ctl error, fd: %d, errno: %d, why: %s", fd, errno, strerror(errno));
            exit(EPOLL_CTL_ERR);
        }
        logMessage(Level::NORMAL, "epoll_ctl success, fd: %d", fd);
 
        // 4. 设置为非阻塞模式
        SetNonBlocking(fd);
    }
 
    void Accepter()
    {
        // 1. 获取新连接
        std::string clientip;
        uint16_t clientport;
        int newfd = sock::Accept(_listensock, &clientip, &clientport);
 
        // 2. 将该新连接交给epoll模型管理
        AddFD(newfd, true);;
    }

​ 设置成接口之后，代码更简洁了！

Ⅷ. epoll 的使用场景

Apache与Nginx网络模型

​ epoll 的高性能是有一定的特定场景的，如果场景选择的不适宜的话，则 epoll 的性能可能适得其反。对于多连接，且多连接中只有一部分连接比较活跃时，比较适合使用 epoll。

​ 例如，典型的一个需要处理上万个客户端的服务器，例如各种互联网 APP 的入口服务器，这样的服务器就很适合 epoll。如果只是系统内部，服务器和服务器之间进行通信，只有少数的几个连接，这种情况下用 epoll 就并不合适。具体要根据需求和场景特点来决定使用哪种IO模型.

Ⅸ. epoll 中的惊群问题

1、epoll 惊群效应产生的原因

​ 很多朋友都在 Linux 下使用 epoll 编写过 socket 的服务端程序，在多线程环境下可能会遇到 epoll 的惊群效应。那么什么是惊群效应呢，其产生的原因是什么呢？

​ 在多线程或者多进程环境下，有些人为了提高程序的稳定性，往往会让多个线程或者多个进程同时在 epoll_wait 中监听文件描述符。当一个新的链接请求进来时，操作系统不知道选派那个线程或者进程处理此事件，则干脆将其中几个线程或者进程给唤醒，而实际上只有其中一个进程或者线程能够成功处理 accept 事件，其他线程都将失败，且 errno 错误码为 EAGAIN。这种现象称为惊群效应，结果是肯定的，惊群效应肯定会带来资源的消耗和性能的影响。

2、多线程环境下解决惊群解决方法

​ 这种情况，不建议让多个线程同时在 epoll_wait 监听文件描述符，而是让其中一个线程 epoll_wait 监听文件描述符，当有新的链接请求进来之后，由 epoll_wait 的线程调用 accept，建立新的连接，然后交给其他工作线程处理后续的数据读写请求，这样就可以避免了由于多线程环境下的 epoll_wait 惊群效应问题。

3、多进程下的解决方法

​ 目前很多开源软件，如 lighttpd、nginx 等都采用 master/workers 的模式提高软件的吞吐能力及并发能力，在 nginx 中甚至还采用了负载均衡的技术，在某个子进程的处理能力达到一定负载之后，由其他负载较轻的子进程负责 epoll_wait 的调用，那么具体 nginx 和 Lighttpd 是如何避免 epoll_wait 的惊群效用的。

​ lighttpd 的解决思路是无视惊群效应，仍然采用 master/workers 模式，每个子进程仍然管自己在监听的 socket 上调用 epoll_wait，当有新的链接请求发生时，操作系统仍然只是唤醒其中部分的子进程来处理该事件，仍然只有一个子进程能够成功处理此事件，那么其他被惊醒的子进程捕获 EAGAIN 错误，并无视。

​ nginx 的解决思路：利用锁机制，使得在同一时刻只有一个子进程在监听的 socket 上调用 epoll_wait，其做法是，创建一个全局的 pthread_mutex_t，在子进程进行 epoll_wait 前，先获取锁。

Ⅹ. EPOLLONESHOT 事件

​ 即使我们使用 ET 模式，一个 socket 上的某个事件还是可能被触发多次，这在并发程序中就会引起一个问题。

​ 比如一个线程（或进程，下同）在读取完某个 socket 上的数据后开始处理这些数据，而在数据的处理过程中该 socket 上又有新数据可读（**EPOLLIN**再次被触发)，此时另外一个线程被唤醒来读取这些新的数据。于是就出现了两个线程同时操作一个 socket 的局面。这当然不是我们期望的，我们期望的是一个 socket 连接在任一时刻都只被一个线程处理。这一点可以使用 epoll 的 EPOLLONESHOT 事件实现。

​ 对于注册了 EPOLLONESHOT 事件的文件描述符，操作系统最多触发其上注册的一个可读、可写或者异常事件，且只触发一次，除非我们使用 epoll_ctl 函数重置该文件描述符上注册的 EPOLLONESHOT 事件。这样，当一个线程在处理某个 socket 时，其他线程是不可能有机会操作该 socket 的。

​ 但反过来思考，注册了 EPOLLONESHOT 事件的 socket 一旦被某个线程处理完毕，该线程就应该立即重置这个 socket 上的 EPOLLONESHOT 事件，以确保这个 socket 下一次可读时，其 EPOLLIN 事件能被触发，进而让其他工作线程有机会继续处理这个 socket。