第4周：Socket 编程深入

目标：掌握非阻塞 I/O、I/O 多路复用，使用 epoll 实现高并发服务器。

---

1. Socket 基础 API

1.1 TCP Socket 完整流程

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/in.h>

// === 服务端 ===
int tcp_server(const char *host, uint16_t port) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) { perror("socket"); return -1; }

    // SO_REUSEADDR：允许快速重启（TIME_WAIT 后立即绑定）
    int opt = 1;
    setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_port   = htons(port),
        .sin_addr.s_addr = INADDR_ANY
    };

    if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
        perror("bind"); close(fd); return -1;
    }

    if (listen(fd, 128) < 0) {
        perror("listen"); close(fd); return -1;
    }

    printf("Server listening on %s:%d (fd=%d)\n", host, port, fd);
    return fd;
}

// === 客户端 ===
int tcp_client(const char *host, uint16_t port) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0) { perror("socket"); return -1; }

    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_port   = htons(port),
    };
    inet_pton(AF_INET, host, &addr.sin_addr);

    if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
        perror("connect"); close(fd); return -1;
    }

    return fd;
}

// === 通用收发 ===
ssize_t tcp_send(int fd, const void *buf, size_t len) {
    const char *ptr = buf;
    size_t sent = 0;

    while (sent < len) {
        ssize_t n = write(fd, ptr + sent, len - sent);
        if (n <= 0) {
            if (n < 0 && errno == EINTR) continue;  // 被信号中断，重试
            return -1;  // 错误或连接关闭
        }
        sent += n;
    }
    return sent;
}

ssize_t tcp_recv(int fd, void *buf, size_t len) {
    ssize_t n = read(fd, buf, len);
    if (n < 0 && errno == EINTR) return 0;  // 被中断，不算错误
    return n;
}

1.2 UDP Socket

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>

// UDP 服务端
int udp_server(uint16_t port) {
    int fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (fd < 0) { perror("socket"); return -1; }

    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_port   = htons(port),
        .sin_addr.s_addr = INADDR_ANY
    };

    bind(fd, (struct sockaddr *)&addr, sizeof(addr));
    printf("UDP server on port %d\n", port);
    return fd;
}

// UDP 收发（无连接）
void udp_demo() {
    int fd = udp_server(9000);

    char buf[1024];
    struct sockaddr_in client_addr;
    socklen_t addrlen = sizeof(client_addr);

    ssize_t n = recvfrom(fd, buf, sizeof(buf) - 1, 0,
                         (struct sockaddr *)&client_addr, &addrlen);
    if (n > 0) {
        buf[n] = '\0';
        char client_ip[INET_ADDRSTRLEN];
        inet_ntop(AF_INET, &client_addr.sin_addr, client_ip, sizeof(client_ip));

        printf("From %s:%d: %s\n", client_ip,
               ntohs(client_addr.sin_port), buf);

        // 回复
        const char *reply = "UDP reply";
        sendto(fd, reply, strlen(reply), 0,
               (struct sockaddr *)&client_addr, addrlen);
    }

    close(fd);
}

---

2. I/O 模型对比

2.1 五种 I/O 模型

阻塞 I/O (Blocking I/O):
  调用 read/write → 阻塞直到数据就绪/完成
  每个连接需要一个线程 → 并发差

非阻塞 I/O (Non-blocking I/O):
  read/write 立即返回 EAGAIN/EWOULDBLOCK
  需要轮询检查 → 浪费 CPU

I/O 多路复用 (I/O Multiplexing):
  select/poll/epoll 等待多个 fd 就绪
  单线程管理多连接 → 高并发

信号驱动 I/O (Signal-driven I/O):
  SIGIO 信号通知 I/O 就绪
  实现复杂，使用较少

异步 I/O (AIO / io_uring):
  内核完成 I/O 后通知
  性能最好但复杂度高

2.2 阻塞 vs 非阻塞对比

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

// === 设置非阻塞模式 ===
int set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    if (flags < 0) return -1;
    return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

// === 非阻塞读写循环 ===
ssize_t nb_read(int fd, void *buf, size_t len) {
    ssize_t n;
    while (1) {
        n = read(fd, buf, len);
        if (n > 0) return n;           // 读到数据
        if (n == 0) return 0;          // EOF
        if (errno == EAGAIN || errno == EWOULDBLOCK) {
            // 数据还没准备好，稍后再试
            return -1;
        }
        if (errno == EINTR) continue;   // 被信号打断，重试
        return -1;                      // 真正的错误
    }
}

// === 非阻塞 connect ===
int nb_connect(int fd, struct sockaddr *addr, socklen_t addrlen) {
    set_nonblocking(fd);

    int ret = connect(fd, addr, addrlen);
    if (ret == 0) return 0;  // 立即连接成功

    if (errno != EINPROGRESS) return -1;

    // 等待连接完成（用 select/poll/epoll 监听可写）
    struct pollfd pfd = {fd, POLLOUT, 0};
    poll(&pfd, 1, -1);

    // 检查结果
    int so_error;
    socklen_t len = sizeof(so_error);
    getsockopt(fd, SOL_SOCKET, SO_ERROR, &so_error, &len);
    return so_error ? -1 : 0;
}

---

3. select

3.1 使用方式

#include <sys/select.h>

int main() {
    int listen_fd = tcp_server("0.0.0.0", 8080);
    set_nonblocking(listen_fd);

    fd_set read_fds, master_fds;
    FD_ZERO(&master_fds);
    FD_SET(listen_fd, &master_fds);
    int max_fd = listen_fd;

    while (1) {
        read_fds = master_fds;  // select 会修改集合，每次要重置

        // 等待事件（阻塞）
        int nready = select(max_fd + 1, &read_fds, NULL, NULL, NULL);
        if (nready < 0) {
            if (errno == EINTR) continue;
            perror("select");
            break;
        }

        // 遍历所有 fd 检查是否就绪
        for (int fd = 0; fd <= max_fd; fd++) {
            if (!FD_ISSET(fd, &read_fds)) continue;

            if (fd == listen_fd) {
                // 新连接
                struct sockaddr_in client_addr;
                socklen_t addrlen = sizeof(client_addr);
                int conn_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &addrlen);
                if (conn_fd < 0) continue;

                set_nonblocking(conn_fd);
                FD_SET(conn_fd, &master_fds);
                if (conn_fd > max_fd) max_fd = conn_fd;
                printf("New connection fd=%d\n", conn_fd);
            } else {
                // 已有连接的数据
                char buf[4096];
                ssize_t n = read(fd, buf, sizeof(buf) - 1);

                if (n <= 0) {
                    // 连接关闭或错误
                    printf("Connection closed fd=%d\n", fd);
                    close(fd);
                    FD_CLR(fd, &master_fds);
                } else {
                    buf[n] = '\0';
                    printf("From fd=%d: %s", fd, buf);
                    write(fd, buf, n);  // echo
                }
            }
        }
    }

    close(listen_fd);
    return 0;
}

3.2 select 的局限

问题	说明
fd_set 上限	FD_SETSIZE 通常 1024
O(n) 扫描	每次 select 要遍历整个 fd_set
fd_set 复制	每次调用从用户态拷贝到内核态
fd 范围	只能管理 0 ~ max_fd

---

4. poll

4.1 poll 使用方式

#include <poll.h>

int main() {
    int listen_fd = tcp_server("0.0.0.0", 8080);
    set_nonblocking(listen_fd);

    struct pollfd fds[1024] = {0};
    fds[0].fd = listen_fd;
    fds[0].events = POLLIN;
    int nfds = 1;

    while (1) {
        int ret = poll(fds, nfds, -1);  // -1 = 无限等待
        if (ret < 0) {
            if (errno == EINTR) continue;
            perror("poll");
            break;
        }

        for (int i = 0; i < nfds; i++) {
            if (fds[i].revents == 0) continue;

            if (fds[i].fd == listen_fd) {
                // 新连接
                int conn_fd = accept(listen_fd, NULL, NULL);
                if (conn_fd < 0) continue;
                set_nonblocking(conn_fd);

                fds[nfds].fd = conn_fd;
                fds[nfds].events = POLLIN;
                nfds++;
            } else {
                // 数据就绪
                char buf[4096];
                ssize_t n = read(fds[i].fd, buf, sizeof(buf) - 1);

                if (n <= 0) {
                    close(fds[i].fd);
                    fds[i] = fds[nfds - 1];  // 用最后一个填充
                    nfds--;
                } else {
                    buf[n] = '\0';
                    write(fds[i].fd, buf, n);
                }
            }
        }
    }
    return 0;
}

4.2 poll vs select

特性	select	poll
fd 上限	1024 (FD_SETSIZE)	无硬性限制
数据结构	fd_set (bitmap)	pollfd 数组
每次调用	复制 fd_set	复制 pollfd 数组
事件类型	只读/写/异常	更丰富 (POLLIN/OUT/PRI/ERR/HUP)
性能	O(n) 扫描	O(n) 扫描

---

5. epoll

5.1 epoll 核心 API

#include <sys/epoll.h>

// 创建 epoll 实例
int epoll_create1(int flags);  // flags=0 或 EPOLL_CLOEXEC

// 注册/修改事件
int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);

// 等待事件
int epoll_wait(int epfd, struct epoll_event *events,
               int maxevents, int timeout);

5.2 Level Triggered (LT) vs Edge Triggered (ET)

Level Triggered (LT，默认):
  [====可读====]------ 条件满足就一直通知
  只要 fd 可读，epoll_wait 就返回

Edge Triggered (ET):
  ----|------------------ 只在状态变化时通知一次
      [====可读====]
  只在 fd 从"不可读"变成"可读"时通知一次
  必须一次性读完，否则可能不再次通知

// ET 模式注册
struct epoll_event ev = {
    .events = EPOLLIN | EPOLLET,  // EPOLLET = Edge Triggered
    .data.fd = conn_fd
};
epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &ev);

// ET 模式下的读取（必须循环直到 EAGAIN）
void et_read(int fd) {
    char buf[4096];
    while (1) {
        ssize_t n = read(fd, buf, sizeof(buf));
        if (n > 0) {
            // 处理数据
            process_data(buf, n);
        } else if (n == 0) {
            // EOF，连接关闭
            close(fd);
            break;
        } else {
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // 已读完，等待下次可读事件
                break;
            }
            if (errno == EINTR) continue;
            // 错误
            close(fd);
            break;
        }
    }
}

ET 优点：减少 epoll_wait 返回次数，高并发下性能更好。
ET 缺点：必须用非阻塞 I/O + 循环读写，实现复杂。

5.3 完整 epoll 实现

// epoll_echo_server.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <arpa/inet.h>
#include <sys/epoll.h>
#include <sys/socket.h>

#define MAX_EVENTS 1024
#define LISTEN_PORT 8080
#define BUFFER_SIZE 4096

int set_nonblocking(int fd) {
    int flags = fcntl(fd, F_GETFL, 0);
    return fcntl(fd, F_SETFL, flags | O_NONBLOCK);
}

int main() {
    // 1. 创建监听 socket
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));

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

    // 2. 创建 epoll 实例
    int epfd = epoll_create1(0);
    if (epfd < 0) { perror("epoll_create1"); return 1; }

    // 3. 注册监听 socket
    struct epoll_event ev = {
        .events = EPOLLIN,
        .data.fd = listen_fd
    };
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);

    // 4. 事件循环
    struct epoll_event events[MAX_EVENTS];

    while (1) {
        int n = epoll_wait(epfd, events, MAX_EVENTS, -1);
        if (n < 0) {
            if (errno == EINTR) continue;
            perror("epoll_wait");
            break;
        }

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

            if (fd == listen_fd) {
                // 新连接
                while (1) {
                    struct sockaddr_in cli_addr;
                    socklen_t len = sizeof(cli_addr);
                    int conn_fd = accept(listen_fd,
                                         (struct sockaddr *)&cli_addr, &len);
                    if (conn_fd < 0) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                        perror("accept");
                        break;
                    }

                    set_nonblocking(conn_fd);

                    char ip[INET_ADDRSTRLEN];
                    inet_ntop(AF_INET, &cli_addr.sin_addr, ip, sizeof(ip));
                    printf("New connection: %s:%d (fd=%d)\n",
                           ip, ntohs(cli_addr.sin_port), conn_fd);

                    // 注册连接 socket（ET 模式示例）
                    struct epoll_event ce = {
                        .events = EPOLLIN | EPOLLET,
                        .data.fd = conn_fd
                    };
                    epoll_ctl(epfd, EPOLL_CTL_ADD, conn_fd, &ce);
                }
            } else {
                // 数据事件
                if (evs & EPOLLERR || evs & EPOLLHUP) {
                    printf("Connection closed (err/hup) fd=%d\n", fd);
                    close(fd);
                    continue;
                }

                if (evs & EPOLLIN) {
                    // ET 模式：循环读取直到 EAGAIN
                    char buf[BUFFER_SIZE];
                    while (1) {
                        ssize_t nread = read(fd, buf, sizeof(buf));
                        if (nread > 0) {
                            // echo 回写
                            ssize_t written = 0;
                            while (written < nread) {
                                ssize_t w = write(fd, buf + written,
                                                  nread - written);
                                if (w <= 0) {
                                    if (errno == EAGAIN) break;
                                    close(fd);
                                    goto next_event;
                                }
                                written += w;
                            }
                        } else if (nread == 0) {
                            printf("Client closed fd=%d\n", fd);
                            close(fd);
                            break;
                        } else {
                            if (errno == EAGAIN || errno == EWOULDBLOCK) break;
                            if (errno == EINTR) continue;
                            perror("read");
                            close(fd);
                            break;
                        }
                    }
                }
            }
        next_event:;
        }
    }

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

gcc -o epoll_echo epoll_echo_server.c
./epoll_echo &

# 测试
curl http://localhost:8080
telnet localhost 8080

# 压测
wrk -t4 -c1000 -d30s http://localhost:8080/

5.4 epoll_event 详解

struct epoll_event {
    uint32_t     events;    /* Epoll events */
    epoll_data_t data;      /* User data variable */
};

// events 字段可用的标志：
// EPOLLIN      可读
// EPOLLOUT     可写
// EPOLLRDHUP   对端关闭或半关闭
// EPOLLERR     错误（总是被报告，无需注册）
// EPOLLHUP     挂起（总是被报告，无需注册）
// EPOLLET      Edge Triggered 模式
// EPOLLONESHOT 只触发一次，需用 epoll_ctl 重新注册

5.5 EPOLLONESHOT 用法

// 适用于每个 fd 只有一个线程处理的场景
// 防止一个 fd 被多个线程同时处理

// 设置
ev.events = EPOLLIN | EPOLLET | EPOLLONESHOT;
epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);

// 处理完事件后，重新注册
ev.events = EPOLLIN | EPOLLET | EPOLLONESHOT;
ev.data.fd = fd;
epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);

---

6. 连接管理：多进程 + SO_REUSEPORT

// 多进程 epoll 服务器，利用 SO_REUSEPORT 负载均衡
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>

#define N_WORKERS 4
#define PORT 8080

int run_worker(int worker_id) {
    int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;

    // SO_REUSEPORT：内核在多个监听 socket 之间负载均衡
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEPORT, &opt, sizeof(opt));

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

    // 每个 worker 独立的 epoll 实例
    int epfd = epoll_create1(0);
    set_nonblocking(listen_fd);

    struct epoll_event ev = {.events = EPOLLIN, .data.fd = listen_fd};
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);

    printf("Worker %d (PID=%d) started\n", worker_id, getpid());

    // ... 事件循环（同前）
    return 0;
}

int main() {
    pid_t pids[N_WORKERS];

    for (int i = 0; i < N_WORKERS; i++) {
        pids[i] = fork();
        if (pids[i] == 0) {
            // 子进程：运行 worker
            return run_worker(i);
        }
    }

    // 父进程：等待子进程
    for (int i = 0; i < N_WORKERS; i++) {
        waitpid(pids[i], NULL, 0);
    }
    return 0;
}

SO_REUSEPORT 优势：

• 内核级别的负载均衡（哈希或轮询分发新连接到不同 socket）
• 避免惊群（多个进程同时被 accept 唤醒）
• 多核利用，每个进程可以绑到不同 CPU

---

7. epoll 性能对比

连接数      | select | poll | epoll(LT) | epoll(ET)
----------- | ------ | ---- | --------- | ---------
100         | 快     | 快   | 快        | 快
1000        | 可接受 | 可接受 | 快      | 快
10000       | 慢     | 可接受 | 快      | 快
100000      | 很慢   | 慢   | 快        | 最快

原因：
- select/poll：每次都是 O(n) 遍历
- epoll：事件驱动，只返回就绪的 fd，O(1) 查找

---

8. 常见问题排查

// === EPOLLHUP 的处理 ===
// 当对端关闭连接，或 shutdown(SHUT_WR) 时触发
// ET 模式下只触发一次，需要关闭 fd

// === 惊群问题 ===
// 多个线程/进程同时被唤醒
// 解决：SO_REUSEPORT（内核负载均衡）
//      epoll 本身已优化，只有被事件驱动的线程被唤醒

// === 水平触发下的事件风暴 ===
// 如果接收缓冲区足够大，每次 epoll_wait 都返回 EPOLLIN
// 但数据可能没读完，下一轮又触发
// 解决：ET 模式 + 非阻塞循环读取

// === 半关闭问题 ===
// 对端 shutdown(SHUT_WR) 只关闭写方向
// 本地还能读，但读返回 0（EOF）不代表完全关闭
// 检查：shutdown 后 recv 返回 0，但 fd 还没 close

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： CoderSong
• 本文链接： https://jack-song-gif.github.io/2026/07/30/第4周：Socket 编程深入/
• 版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！