Logo

菜单

蔚淦丞
发布于 2026-09-07 / 9 阅读
0
0

epoll单线程最简模型 + 多线程多Reactor工业模型

一、epoll 核心概念

epoll 是 Linux 专属的高性能 IO 多路复用机制，核心作用是监听海量文件描述符的读写事件，是 C/C++ 高并发 TCP 服务的底层基石。

相较于 select/poll：epoll 采用事件通知机制，仅返回就绪的文件描述符，海量空闲连接不会产生性能损耗，高并发场景性能优势极强。

二、epoll 三大核心函数（最简用法）

epoll 所有功能均依赖三个核心函数，分工明确、解耦清晰。

1. epoll_create

int epoll_create(int size);

创建一个独立的 epoll 实例，返回 epollfd 文件描述符。

  • 新版 Linux 中 size 参数已废弃，无实际限制作用；

  • 支持多线程创建多个独立 epoll 实例，线程间监听资源完全隔离；

  • 使用完毕必须 close 释放，避免文件描述符泄露。

2. epoll_ctl

int epoll_ctl(int epfd, int op, int fd, struct epoll_event *ev);

epoll 注册管理核心接口，用于对指定 fd 进行增、删、改监听操作。

操作类型 op：

  • EPOLL_CTL_ADD：新增 fd 到 epoll 监听集合

  • EPOLL_CTL_MOD：修改已注册 fd 的监听事件

  • EPOLL_CTL_DEL：从 epoll 中移除 fd 监听

核心监听事件：

  • EPOLLIN：fd 可读就绪（新连接/客户端数据）

  • EPOLLOUT：fd 可写就绪（内核发送缓冲区空闲）

3. epoll_wait

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);

阻塞等待内核返回就绪事件，批量获取就绪的 fd 列表。

  • events：用户态数组，接收内核返回的就绪事件数据；

  • maxevents：单次调用最多返回的就绪事件数量；

  • timeout：-1 永久阻塞、0 非阻塞、正数阻塞指定毫秒；

  • 返回值：就绪事件个数，0 为超时，-1 为执行出错。

三、核心参数误区：MAX_EVENTS

核心结论：MAX_EVENTS 只限制单次 epoll_wait 返回的事件数，不限制 epoll 整体监听的 fd 总数。

epoll 最大可监听 fd 数量，仅由系统参数 ulimit -n 决定，与该参数无关。

场景举例：内核就绪队列存在 1000 个就绪事件，MAX_EVENTS=128

  • 本次 epoll_wait 仅返回 128 个事件；

  • 剩余事件留在内核队列，不会丢失；

  • 下一轮循环调用 epoll_wait 继续分批处理。

工程取舍：参数过小会频繁触发系统调用、性能下降；过大会占用栈内存，常规业务 128/256/512 为最优配置。

四、长连接核心：EPOLLIN / EPOLLOUT 事件区分

单个客户端 fd 可同时监听 EPOLLIN | EPOLLOUT，epoll 返回的 events 是位掩码，可通过位与运算精准判断事件类型，支撑长连接多轮交互。

// 事件判断核心代码
uint32_t rev = events[i].events;
if (rev & EPOLLIN) {
    // 可读事件：客户端发送数据，执行recv读取
}
if (rev & EPOLLOUT) {
    // 可写事件：内核缓冲区空闲，可发送残留数据
}

可写事件使用场景（关键）

TCP 发送缓冲区打满时，send 无法一次性发送全部数据，不能阻塞等待。标准处理流程：

  1. 数据未发送完毕，通过 EPOLL_CTL_MOD 开启 EPOLLOUT 监听；

  2. 内核缓冲区空闲后，触发 EPOLLOUT 事件，续发残留数据；

  3. 数据全部发送完成，立即关闭 EPOLLOUT，避免 CPU 空转 100%。

五、epoll 两种触发模式 LT / ET

  • LT 水平触发（默认）：只要 fd 存在未读完数据、缓冲区可写，就会持续触发事件。编程简单、稳定性高，适合绝大多数业务。

  • ET 边沿触发：仅在 fd 状态切换瞬间触发一次事件。必须搭配非阻塞 IO + 循环读写至 EAGAIN，性能更高，开发复杂度更高。

六、入门代码：单线程最简 epoll 模型

极简实现，仅保留核心逻辑，适合理解 epoll 基础运行流程。

#include <stdio.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <fcntl.h>

#define MAX_EVENTS 128
#define PORT 8888

// 设置非阻塞
int set_nonblock(int fd){
    int flag = fcntl(fd, F_GETFL);
    fcntl(fd, F_SETFL, flag | O_NONBLOCK);
    return fd;
}

int main(){
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    set_nonblock(listen_fd);

    struct sockaddr_in addr = {0};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(PORT);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);

    bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr));
    listen(listen_fd, 128);

    int epfd = epoll_create(10);
    struct epoll_event ev, events[MAX_EVENTS];

    // 注册监听fd读事件
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);

    while(1){
        int n = epoll_wait(epfd, events, MAX_EVENTS, -1);
        for(int i = 0; i < n; i++){
            int fd = events[i].data.fd;
            // 新连接事件
            if(fd == listen_fd){
                int cli_fd = accept(listen_fd, NULL, NULL);
                set_nonblock(cli_fd);
                ev.events = EPOLLIN;
                ev.data.fd = cli_fd;
                epoll_ctl(epfd, EPOLL_CTL_ADD, cli_fd, &ev);
            }
            // 客户端数据可读事件
            else{
                char buf[1024] = {0};
                int ret = recv(fd, buf, sizeof(buf)-1, 0);
                if(ret <= 0){
                    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
                    close(fd);
                    continue;
                }
                printf("recv: %s\n", buf);
                send(fd, buf, ret, 0);
            }
        }
    }
    return 0;
}

七、工业级代码：多线程多Reactor 完整模型

核心模型：主线程仅负责 accept 接收新连接，每个工作线程维护独立的 epoll 实例，独立处理读写事件，支持长连接、半包续发、动态开关读写事件。

编译命令：gcc epoll.c -o epoll -lpthread

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <errno.h>

#define MAX_EVENTS        128
#define LISTEN_PORT       8888
#define WORKER_THREAD_NUM 4
#define BUF_SIZE          4096

// 连接上下文：保存单个连接的读写状态、半包数据
typedef struct ConnCtx {
    int fd;
    char recv_buf[BUF_SIZE];
    int  recv_len;

    char send_buf[BUF_SIZE];
    int  send_total;
    int  send_offset;
} ConnCtx;

// 工作线程参数：每个线程独立epoll
typedef struct {
    int epfd;
} Worker;

Worker workers[WORKER_THREAD_NUM];

// 设置文件描述符非阻塞
static int set_nonblock(int fd) {
    int flags = fcntl(fd, F_GETFL);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    return fd;
}

// 清空发送缓冲区状态
static void clear_send_ctx(ConnCtx *ctx) {
    ctx->send_total = 0;
    ctx->send_offset = 0;
}

// 工作线程主循环：独立处理读写事件
void *worker_loop(void *arg) {
    Worker *w = (Worker*)arg;
    int epfd = w->epfd;
    struct epoll_event events[MAX_EVENTS];

    while (1) {
        int nready = epoll_wait(epfd, events, MAX_EVENTS, -1);
        if (nready <= 0) continue;

        for (int i = 0; i < nready; i++) {
            int fd = events[i].data.fd;
            uint32_t rev = events[i].events;

            // 可读事件：接收客户端数据
            if (rev & EPOLLIN) {
                char buf[BUF_SIZE] = {0};
                int n = recv(fd, buf, sizeof(buf)-1, 0);
                if (n <= 0) {
                    epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
                    close(fd);
                    continue;
                }
                printf("thread %lu recv: %s\n", pthread_self(), buf);

                // 初始化发送上下文，准备回显数据
                ConnCtx ctx = {0};
                memcpy(ctx.send_buf, buf, n);
                ctx.send_total = n;
                ctx.send_offset = 0;

                // 尝试一次性发送数据
                int sent = send(fd, ctx.send_buf, ctx.send_total, MSG_NOSIGNAL);
                if (sent > 0) ctx.send_offset = sent;

                // 数据未发完：开启可写事件，等待缓冲区空闲续发
                if (ctx.send_offset < ctx.send_total) {
                    struct epoll_event ev;
                    ev.events = EPOLLIN | EPOLLOUT;
                    ev.data.fd = fd;
                    epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
                } 
                // 数据发完：仅保留可读事件
                else {
                    struct epoll_event ev;
                    ev.events = EPOLLIN;
                    ev.data.fd = fd;
                    epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
                }
            }

            // 可写事件：续发残留数据
            if (rev & EPOLLOUT) {
                // 发送完成后必须关闭可写事件，防止CPU空转
                struct epoll_event ev;
                ev.events = EPOLLIN;
                ev.data.fd = fd;
                epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);
            }
        }
    }
    return NULL;
}

int main() {
    // 1. 创建监听socket
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    set_nonblock(listen_fd);

    struct sockaddr_in addr = {0};
    addr.sin_family = AF_INET;
    addr.sin_port = htons(LISTEN_PORT);
    addr.sin_addr.s_addr = htonl(INADDR_ANY);
    bind(listen_fd, (struct sockaddr*)&addr, sizeof(addr));
    listen(listen_fd, 128);

    // 2. 初始化多工作线程，每个线程独立epoll
    for (int i = 0; i < WORKER_THREAD_NUM; i++) {
        workers[i].epfd = epoll_create(10);
        pthread_t tid;
        pthread_create(&tid, NULL, worker_loop, &workers[i]);
    }

    // 3. 主线程epoll：仅监听新连接
    int main_epfd = epoll_create(10);
    struct epoll_event ev, events[MAX_EVENTS];
    ev.events = EPOLLIN;
    ev.data.fd = listen_fd;
    epoll_ctl(main_epfd, EPOLL_CTL_ADD, listen_fd, &ev);

    int idx = 0;
    while (1) {
        int n = epoll_wait(main_epfd, events, MAX_EVENTS, -1);
        if (n <= 0) continue;

        // 接收新连接，轮询分发到工作线程
        int cli_fd = accept(listen_fd, NULL, NULL);
        if (cli_fd <= 0) continue;
        set_nonblock(cli_fd);

        int worker_epfd = workers[idx % WORKER_THREAD_NUM].epfd;
        idx++;

        // 将新客户端fd注册到工作线程epoll
        ev.events = EPOLLIN;
        ev.data.fd = cli_fd;
        epoll_ctl(worker_epfd, EPOLL_CTL_ADD, cli_fd, &ev);
    }

    return 0;
}

八、关键答疑：C语言连接上下文存储方案

1. 核心前提

C语言标准库无 map 容器，网络开发中「fd映射存储上下文」是通俗叫法，并非使用哈希map。同时 epoll 是纯事件通知器，不保存任何连接业务状态，半包数据、发送偏移、连接状态必须开发者手动维护。

2. 为什么不推荐普通数组

直接定义固定结构体数组 ConnCtx ctx[65536] 存在致命问题：

  • Linux fd 随机递增、可复用、不连续；

  • 连接断开后 fd 复用，会导致数组下标脏数据覆盖；

  • 大量未使用下标造成严重内存空洞、资源浪费。

3. C语言工业级标准方案：fd指针索引数组

利用 fd 是纯数字的特性，以 fd 为下标、存储结构体指针，替代map，实现 O(1) 极速查询，无哈希冲突、实现极简。

4. 完整定义与使用代码

// 连接上下文结构体
typedef struct ConnCtx {
    int fd;
    char recv_buf[4096];
    int  recv_len;

    char send_buf[4096];
    int  send_total;
    int  send_offset;
} ConnCtx;

// 全局指针数组：适配系统最大fd限制
#define MAX_FD_LIMIT 65536
ConnCtx *conn_map[MAX_FD_LIMIT] = {NULL};

新连接初始化上下文：

int cli_fd = accept(listen_fd, NULL, NULL);
if (cli_fd > 0) {
    ConnCtx *ctx = (ConnCtx*)malloc(sizeof(ConnCtx));
    memset(ctx, 0, sizeof(ConnCtx));
    ctx->fd = cli_fd;
    conn_map[cli_fd] = ctx;
}

事件触发读取上下文：

int fd = events[i].data.fd;
ConnCtx *ctx = conn_map[fd];
if (!ctx) continue;

连接断开释放资源（防内存泄露、防脏数据）：

epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
close(fd);
free(conn_map[fd]);
conn_map[fd] = NULL;

5. 方案核心优势

  • 性能最优：纯数组下标访问，O(1) 速度，无哈希计算；

  • 零依赖：纯原生C语言实现，无需第三方库；

  • 规避脏数据：连接断开即时释放、置空，适配fd复用机制；

  • 无业务上限：仅受系统 ulimit -n 和机器内存限制。

九、全文核心总结

  1. epoll 三大函数各司其职：epoll_create 创建实例、epoll_ctl 管理监听、epoll_wait 批量获取就绪事件。

  2. MAX_EVENTS 仅控制单次返回事件数，不限制全局监听 fd 总量，内核事件不会丢失。

  3. 通过位与运算区分 EPOLLIN/EPOLLOUT，动态开关可写事件，是长连接多轮收发、半包续发的核心。

  4. 多Reactor模型：主线程负责连接接收，多工作线程独立epoll处理读写，并发性能更强。

  5. C语言无map，工业级采用「fd指针索引数组」维护连接上下文，高效且稳定。

  6. epoll 仅负责事件通知，所有业务状态、数据拼接、残留发送均需用户态手动维护。

Linux 线程等待 / 唤醒相关...
Select/Poll使用总结

评论