Linux epoll ET 模式如何用于多线程，局部 buf 是否安全？

2026-11-04数据阿明

结论：ET 不会自动创建线程，线程切换也不会破坏局部 buf

EPOLLET 只改变 epoll 的事件通知方式，既不会创建工作线程，也不会自动保证业务代码的线程安全。题中的程序没有调用线程创建接口，运行的是单线程事件循环。

多个线程分别调用同一个处理函数时，每次调用中的 char buf[5] 都是独立的局部对象。线程切换本身不会让一个线程覆盖另一个线程的 buf。

这段代码仍可能出现缓冲区内容异常，例如读取未初始化的数据、把非字符串当作字符串打印，或错误地判断写事件。这些问题在单线程中也会发生。

局部数组在什么情况下会发生并发问题？

要看其他线程是否拿到了同一个数组的地址，以及数组是否仍然有效。

多个线程分别进入处理函数，各自创建局部 buf：这些缓冲区相互独立。
把 buf 的地址传给工作线程：两个线程可能访问同一个数组，需要协调访问。
工作线程仍在使用 buf 时，创建数组的代码已经离开作用域：指针失效，继续访问会产生未定义行为。
缓冲区彼此独立，但多个线程同时读取同一个 socket：仍需协调连接处理，否则数据可能被不同线程分别取走，协议状态也可能发生竞争。

线程各自有栈，但栈内存并不是其他线程无法访问的隔离区域。只要传递了有效地址，其他线程就可能访问它。

如果采用线程池，可以把接收的数据复制到任务拥有的存储中，或者让工作线程独占连接及其缓冲区。不要把事件循环中临时数组的地址直接交给异步任务，然后继续复用该数组。

原代码需要先修正哪些问题？

1. 写事件判断用了错误的运算符

原代码是：

if (epevs[i].events | EPOLLOUT) {

| 是按位或。由于 EPOLLOUT 非零，即使当前事件没有包含 EPOLLOUT，这个条件也会成立。

应使用按位与检查标志：

if (epevs[i].events & EPOLLOUT) {

仅修改运算符还不够。socket 可写只说明当前可以尝试发送数据，不表示应用已经有数据可发送。

2. read() 返回的数据不一定是字符串

下面的代码不安全：

char buf[5];
len = read(curfd, buf, sizeof(buf));
printf("Recv data: %s\n", buf);

read() 不会自动补上 \0。即使读取成功，%s 也可能越过数组边界继续读取；读取失败或返回 0 时，更不能把数组当作有效的接收数据打印。

如果只是输出收到的字节，可以先检查返回值，再按实际长度输出：

ssize_t n = read(curfd, buf, sizeof(buf));
if (n > 0) {
    fwrite(buf, 1, (size_t)n, stdout);
}

这个示例只展示如何安全地使用读取长度，完整的 ET 处理还需要循环读取。

3. ET 模式下只读一次，可能留下无法及时处理的数据

非阻塞 socket 配合 ET 时，通常应持续读取，直到出现以下结果之一：

返回 -1，且 errno 为 EAGAIN 或 EWOULDBLOCK：当前可读数据已经取尽。
返回 0：对端关闭了发送方向，按协议处理连接后续状态。
发生其他读取错误：进入错误处理流程。

如果只读取 5 字节就回到 epoll_wait()，socket 接收缓冲区里可能还有数据，不能依赖 epoll 再次通知这些剩余数据。

EINTR 表示调用被信号中断，通常可以重试。

4. 发送长度必须来自有效数据

原代码直接执行：

write(curfd, buf, sizeof(buf));

如果只收到 2 字节，却发送 5 字节，就会把数组中其余无效内容也发送出去。若当前只有写事件而没有成功读取数据，整个数组都可能尚未初始化。

即使使用实际读取长度，非阻塞 write() 也可能只发送一部分，或返回 EAGAIN。需要为连接保存待发送数据和发送偏移，以便后续继续发送。

通常只在确有待发送数据、且发送因不可写而暂停时关注 EPOLLOUT；数据发送完毕后移除该关注。ET 写事件不能代替应用维护待发送队列。

非阻塞 ET 读取可以怎样组织？

这个辅助函数只负责读取，不包含完整的 echo 服务或发送队列：

#include <errno.h>
#include <stddef.h>
#include <unistd.h>

/*
 * 返回值：
 *  0：当前数据已读尽，等待后续事件
 *  1：读到 EOF
 * -1：读取出错，errno 保留错误原因
 *
 * on_bytes 必须同步消费或复制数据，不能保存 buf 指针供稍后使用。
 */
static int drain_input(
    int fd,
    void (*on_bytes)(const char *data, size_t size)
) {
    char buf[4096];

    for (;;) {
        ssize_t n = read(fd, buf, sizeof(buf));

        if (n > 0) {
            on_bytes(buf, (size_t)n);
            continue;
        }

        if (n == 0) {
            return 1;
        }

        if (errno == EINTR) {
            continue;
        }

        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            return 0;
        }

        return -1;
    }
}

调用前必须确认连接已成功设置为 O_NONBLOCK。回调收到的是一段字节流，不保证恰好对应一条应用消息；协议解析可能需要跨多次读取保留状态。

如果回调需要异步处理，应先复制数据。如果需要限制单次处理的数据量，还要安排后续主动继续处理，避免在尚未读到 EAGAIN 时直接等待新的边缘通知。

改成多线程时，怎样避免同一连接被同时处理？

一种常见做法是由事件线程接收通知，再把连接交给线程池处理，同时使用 EPOLLONESHOT 控制事件重新交付：

struct epoll_event ev = {0};
ev.data.fd = cfd;
ev.events = EPOLLIN | EPOLLET | EPOLLONESHOT;

if (epoll_ctl(epfd, EPOLL_CTL_ADD, cfd, &ev) == -1) {
    /* 处理注册失败 */
}

收到事件后，由一个工作线程处理该连接。工作线程处理完当前 I/O、保存连接状态并准备交还处理权时，通过 EPOLL_CTL_MOD 重新启用通知：

struct epoll_event ev = {0};
ev.data.fd = cfd;
ev.events = EPOLLIN | EPOLLET | EPOLLONESHOT;

/* 有待发送数据时，按发送状态决定是否加入 EPOLLOUT。 */
if (epoll_ctl(epfd, EPOLL_CTL_MOD, cfd, &ev) == -1) {
    /* 处理重新启用失败 */
}

重新启用后，新事件可能很快由其他线程取得。因此，交还处理权后，原工作线程不能继续无保护地修改连接状态。

EPOLLONESHOT 控制的是这个注册项的通知，不是 socket 的互斥锁。它不能阻止其他代码直接调用 read()、write() 或 close()。任务派发、连接状态访问和关闭操作仍要遵守统一的所有权规则。

多个线程也可以各自调用同一个 epoll 实例的 epoll_wait()，但每个调用线程应使用自己的事件输出数组。ET 下同一就绪通知的唤醒行为，不能保证整个连接处理过程互斥。

监听与连接清理还有哪些注意事项？

原代码给监听 socket 注册的是 EPOLLIN，因此监听端使用 LT；给已连接 socket 注册了 EPOLLET，这些连接才使用 ET。这种组合本身可以成立。

若把监听 socket 也改成 ET，应将其设置为非阻塞，并循环 accept()，直到返回 EAGAIN 或 EWOULDBLOCK。传给 accept() 的地址长度变量应使用 socklen_t：

socklen_t len = sizeof(cliaddr);
int cfd = accept(lfd, (struct sockaddr *)&cliaddr, &len);

还应检查 socket()、bind()、listen()、accept()、fcntl() 和 epoll_ctl() 的返回值。epoll_wait() 因 EINTR 返回时，通常可以继续等待。

读到 EOF 或发生连接错误后，要按协议决定是否完成剩余发送，并最终移除事件关注、关闭连接。多线程下还要确保没有未完成任务继续使用已关闭的文件描述符，尤其要防止描述符编号被新连接复用后，旧任务误操作新连接。