第6周：高性能服务模型

字数统计: 2.8k字   |   阅读时长≈ 15分

第6周：高性能服务模型

目标：掌握 Reactor/Proactor 模型，实现多线程高性能网络服务。

1. I/O 模型对比

1.1 Reactor vs Proactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
Reactor（反应器模式）：
  - 事件多路复用：等事件就绪
  - 同步 I/O：读写操作在就绪后自己完成
  - 代表：epoll, kqueue, IOCP 同步模式

  ┌─────────────┐
  │  Event Loop │ ← 等待 I/O 事件
  │  (epoll)    │
  └──────┬──────┘
         │ EPOLLIN 就绪
         ▼
  ┌─────────────┐
  │  read()     │ ← 同步读取（阻塞但 fd 已就绪）
  │  process    │
  │  write()    │ ← 同步写入
  └─────────────┘

Proactor（前摄器模式）：
  - 异步 I/O：提交操作，完成时回调
  - 不需要关心"何时就绪"，关心"何时完成"
  - 代表：Windows IOCP, Linux io_uring

  ┌─────────────┐
  │  Submit I/O │ ← 提交异步读写请求
  │  (io_submit)│
  └──────┬──────┘
         │
  ┌──────▼──────┐
  │  Callback  │ ← 完成时被通知
  │  (io_getevents)│
  └─────────────┘

1.2 五种主要并发模型

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
// === 模型 1：每连接一线程（Thread-per-Connection）===

void thread_per_connection() {
    while (1) {
        int conn = accept(listen_fd, ...);
        pthread_create(&tid, NULL, handle_client, (void *)(long)conn);
    }
}

// 优点：简单直观
// 缺点：线程数 = 连接数，10k 连接 = 10k 线程，上下文切换开销巨大


// === 模型 2：线程池 + 每任务一线程 ===

void thread_pool() {
    ThreadPool pool = tp_create(16);
    while (1) {
        int conn = accept(listen_fd, ...);
        Task task = {.fd = conn};
        tp_submit(&pool, handle_client, &task);
    }
}


// === 模型 3：单线程 Reactor ===

void single_thread_reactor() {
    int epfd = epoll_create1(0);
    register_listen_fd(epfd, listen_fd);

    while (1) {
        int n = epoll_wait(epfd, events, MAX, -1);
        for (int i = 0; i < n; i++) {
            if (is_listen_fd(events[i]))
                accept_and_register(epfd);
            else
                handle_io(events[i].data.fd);  // 单线程处理
        }
    }
}

// 优点：无锁，简单
// 缺点：单核，无法利用多核


// === 模型 4：多线程 Reactor（One Loop Per Thread）===

void multi_thread_reactor() {
    // 主线程（Acceptor）：接受连接
    // 工作线程：每个线程有自己的 epoll loop
    // 连接通过 hash 绑定到指定线程

    pthread_t workers[N];
    for (int i = 0; i < N; i++) {
        pthread_create(&workers[i], NULL, worker_loop, NULL);
    }

    // 主线程
    while (1) {
        int conn = accept(listen_fd, ...);
        // Round-robin 或 hash 分发
        int worker_id = conn_fd % N;
        dispatch_to_worker(worker_id, conn);
    }
}

// 优点：多核利用，每线程独立无锁
// 缺点：跨线程通信需要队列


// === 模型 5：主从 Reactor（Master-Worker）===

void master_worker_reactor() {
    // 主 Reactor：只处理 accept 事件
    // 从 Reactor：多个线程各自运行 epoll，处理 I/O

    // 主线程
    int main_epfd = epoll_create1(0);
    register_listen_fd(main_epfd, listen_fd);

    // 工作线程
    int worker_epfd[N];
    for (int i = 0; i < N; i++) {
        worker_epfd[i] = epoll_create1(0);
        start_worker(i, worker_epfd[i]);
    }

    int next_worker = 0;
    while (1) {
        int conn = accept(listen_fd, ...);
        // 转移到 worker 的 epoll
        transfer_fd(worker_epfd[next_worker], conn);
        next_worker = (next_worker + 1) % N;
    }
}

2. Reactor 模式完整实现

2.1 单线程 Reactor

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// reactor.h
#ifndef REACTOR_H
#define REACTOR_H

#include <sys/epoll.h>
#include <stdint.h>

#define MAX_EVENTS 4096
#define MAX_FD     10240

// 事件回调类型
typedef void (*event_callback)(int fd, uint32_t events, void *user_data);

// Reactor 实例
typedef struct Reactor {
    int epfd;
    int listen_fd;
    event_callback callbacks[MAX_FD];
    void *user_data[MAX_FD];
} Reactor;

Reactor *reactor_create(int port);
void reactor_run(Reactor *r);
void reactor_add(Reactor *r, int fd, uint32_t events,
                 event_callback cb, void *user_data);
void reactor_remove(Reactor *r, int fd);
void reactor_destroy(Reactor *r);

#endif
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
// reactor.c
#include "reactor.h"
#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

Reactor *reactor_create(int port) {
    Reactor *r = calloc(1, sizeof(Reactor));

    r->epfd = epoll_create1(0);
    if (r->epfd < 0) { perror("epoll_create1"); return NULL; }

    r->listen_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(r->listen_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
    };
    bind(r->listen_fd, (struct sockaddr *)&addr, sizeof(addr));
    listen(r->listen_fd, 1024);
    set_nonblocking(r->listen_fd);

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

    return r;
}

void reactor_add(Reactor *r, int fd, uint32_t events,
                 event_callback cb, void *user_data) {
    struct epoll_event ev = {.events = events, .data.fd = fd};
    epoll_ctl(r->epfd, EPOLL_CTL_ADD, fd, &ev);
    r->callbacks[fd] = cb;
    r->user_data[fd] = user_data;
}

void reactor_remove(Reactor *r, int fd) {
    epoll_ctl(r->epfd, EPOLL_CTL_DEL, fd, NULL);
    r->callbacks[fd] = NULL;
    r->user_data[fd] = NULL;
    close(fd);
}

void reactor_run(Reactor *r) {
    struct epoll_event events[MAX_EVENTS];

    while (1) {
        int n = epoll_wait(r->epfd, events, MAX_EVENTS, -1);
        for (int i = 0; i < n; i++) {
            int fd = events[i].data.fd;
            uint32_t evs = events[i].events;

            if (evs & (EPOLLERR | EPOLLHUP)) {
                close(fd);
                continue;
            }

            if (fd == r->listen_fd) {
                // 新连接（由用户注册的回调处理）
                continue;
            }

            if (r->callbacks[fd]) {
                r->callbacks[fd](fd, evs, r->user_data[fd]);
            }
        }
    }
}

void reactor_destroy(Reactor *r) {
    close(r->epfd);
    close(r->listen_fd);
    free(r);
}

2.2 多线程 Reactor（One Loop Per Thread）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
// multi_reactor.c
#include "reactor.h"
#include <pthread.h>

typedef struct {
    Reactor *reactor;
    int worker_id;
    int epfd;
} WorkerThread;

// 工作线程入口
void *worker_thread(void *arg) {
    WorkerThread *wt = (WorkerThread *)arg;
    printf("Worker %d started (TID=%lu)\n", wt->worker_id, pthread_self());

    struct epoll_event events[MAX_EVENTS];
    while (1) {
        int n = epoll_wait(wt->epfd, events, MAX_EVENTS, -1);
        for (int i = 0; i < n; i++) {
            int fd = events[i].data.fd;
            uint32_t evs = events[i].events;

            if (evs & (EPOLLERR | EPOLLHUP)) {
                close(fd);
                continue;
            }

            // 新连接：只负责转移 epoll 所属
            if (evs & EPOLLIN) {
                char buf[1];
                read(fd, buf, 1);  // 读一个字节触发 LT
                // 这里可以做协议解析，确认连接建立后转移
            }

            if (wt->reactor->callbacks[fd]) {
                wt->reactor->callbacks[fd](fd, evs, wt->reactor->user_data[fd]);
            }
        }
    }
    return NULL;
}

// 连接分发策略
typedef enum {
    DISPATCH_ROUND_ROBIN,
    DISPATCH_LEAST_CONN,
    DISPATCH_HASH
} DispatchStrategy;

int dispatch_fd(Reactor *reactor, int conn_fd, DispatchStrategy strategy,
                int n_workers, WorkerThread *workers) {
    int worker_id;

    switch (strategy) {
        case DISPATCH_ROUND_ROBIN:
            worker_id = conn_fd % n_workers;  // 简单 hash
            break;
        case DISPATCH_LEAST_CONN:
            worker_id = 0;  // TODO: 追踪每个 worker 的连接数
            break;
        case DISPATCH_HASH:
            // 基于源 IP 的 hash（同一客户端固定到同一 worker）
            struct sockaddr_in addr;
            socklen_t len = sizeof(addr);
            getpeername(conn_fd, (struct sockaddr *)&addr, &len);
            worker_id = (addr.sin_addr.s_addr ^ addr.sin_port) % n_workers;
            break;
    }

    // 从主 epoll 转移到 worker 的 epoll
    epoll_ctl(reactor->epfd, EPOLL_CTL_DEL, conn_fd, NULL);
    epoll_ctl(workers[worker_id].epfd, EPOLL_CTL_ADD, conn_fd, NULL);

    return worker_id;
}

3. HTTP 静态文件服务器

3.1 HTTP 请求解析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
#include <stdio.h>
#include <string.h>

// 简易 HTTP 请求结构
typedef struct {
    char method[16];
    char path[256];
    char version[16];
    int  content_length;
    int  headers_done;
} HttpRequest;

// 状态机解析 HTTP 请求行
typedef enum {
    PARSE_METHOD,
    PARSE_PATH,
    PARSE_VERSION,
    PARSE_CRLF,
    PARSE_DONE,
    PARSE_ERROR
} ParseState;

int parse_http_request(const char *data, size_t len, HttpRequest *req) {
    ParseState state = PARSE_METHOD;
    const char *p = data;
    const char *end = data + len;

    char *dst = NULL;
    char *dst_end = NULL;

    #define COPY_FIELD(field, max_len) \
        do { \
            dst = field; \
            dst_end = field + max_len - 1; \
        } while(0)

    while (p < end && state != PARSE_DONE && state != PARSE_ERROR) {
        switch (state) {
            case PARSE_METHOD:
                COPY_FIELD(req->method, sizeof(req->method));
                while (p < end && *p != ' ' && dst < dst_end) *dst++ = *p++;
                *dst = '\0';
                if (p < end && *p == ' ') { p++; state = PARSE_PATH; }
                else state = PARSE_ERROR;
                break;

            case PARSE_PATH:
                COPY_FIELD(req->path, sizeof(req->path));
                while (p < end && *p != ' ' && dst < dst_end) *dst++ = *p++;
                *dst = '\0';
                if (p < end && *p == ' ') { p++; state = PARSE_VERSION; }
                else state = PARSE_ERROR;
                break;

            case PARSE_VERSION:
                COPY_FIELD(req->version, sizeof(req->version));
                while (p < end && *p != '\r' && dst < dst_end) *dst++ = *p++;
                *dst = '\0';
                if (p + 1 < end && p[0] == '\r' && p[1] == '\n') {
                    p += 2; state = PARSE_CRLF;
                }
                break;

            case PARSE_CRLF:
                if (p + 1 < end && p[0] == '\r' && p[1] == '\n') {
                    state = PARSE_DONE;
                }
                // 跳过 header 行（简化：跳到空行）
                while (p < end - 1) {
                    if (p[0] == '\r' && p[1] == '\n') {
                        p += 2;
                        if (p < end - 1 && p[0] == '\r' && p[1] == '\n') {
                            state = PARSE_DONE;
                        }
                        break;
                    }
                    p++;
                }
                break;

            default: break;
        }
    }

    #undef COPY_FIELD
    return state == PARSE_DONE ? 0 : -1;
}

3.2 完整 HTTP 静态文件服务器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
// http_server.c
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "reactor.h"

#define DOC_ROOT "./www"
#define PORT     8080

// 每个连接的上下文
typedef struct {
    int fd;
    char buffer[8192];
    size_t buf_len;
    HttpRequest req;
    int parse_done;
} ConnContext;

// 初始化上下文
void ctx_init(ConnContext *ctx, int fd) {
    ctx->fd = fd;
    ctx->buf_len = 0;
    ctx->parse_done = 0;
    memset(&ctx->req, 0, sizeof(ctx->req));
}

// 发送 HTTP 响应
void send_response(int fd, int status_code, const char *content_type,
                   const char *body, size_t body_len) {
    char header[512];
    int header_len = snprintf(header, sizeof(header),
        "HTTP/1.1 %d %s\r\n"
        "Content-Type: %s\r\n"
        "Content-Length: %zu\r\n"
        "Connection: close\r\n"
        "\r\n",
        status_code,
        status_code == 200 ? "OK" :
        status_code == 404 ? "Not Found" : "Error",
        content_type, body_len);

    write(fd, header, header_len);
    if (body_len > 0) {
        write(fd, body, body_len);
    }
}

// 处理 HTTP 请求
void handle_http_request(int fd, uint32_t events, void *user_data) {
    ConnContext *ctx = (ConnContext *)user_data;

    // 读取数据
    char buf[4096];
    ssize_t n = read(fd, buf, sizeof(buf));
    if (n <= 0) {
        close(fd);
        free(ctx);
        return;
    }

    // 追加到缓冲区
    memcpy(ctx->buffer + ctx->buf_len, buf, n);
    ctx->buf_len += n;

    // 解析请求
    if (!ctx->parse_done) {
        if (parse_http_request(ctx->buffer, ctx->buf_len, &ctx->req) == 0) {
            ctx->parse_done = 1;
        }
        return;  // 还没解析完，等下一轮
    }

    // 构建文件路径
    char filepath[512];
    snprintf(filepath, sizeof(filepath), "%s%s", DOC_ROOT, ctx->req.path);
    if (strcmp(ctx->req.path, "/") == 0) {
        snprintf(filepath, sizeof(filepath), "%s/index.html", DOC_ROOT);
    }

    // 打开文件
    int file_fd = open(filepath, O_RDONLY);
    if (file_fd < 0) {
        const char *not_found = "<h1>404 Not Found</h1>";
        send_response(fd, 404, "text/html", not_found, strlen(not_found));
    } else {
        struct stat st;
        fstat(file_fd, &st);

        // 确定 Content-Type
        const char *ext = strrchr(filepath, '.');
        const char *type = "application/octet-stream";
        if (ext) {
            if (strcmp(ext, ".html") == 0) type = "text/html";
            else if (strcmp(ext, ".css") == 0) type = "text/css";
            else if (strcmp(ext, ".js") == 0) type = "application/javascript";
            else if (strcmp(ext, ".png") == 0) type = "image/png";
            else if (strcmp(ext, ".jpg") == 0) type = "image/jpeg";
        }

        // 发送响应头
        char header[256];
        int header_len = snprintf(header, sizeof(header),
            "HTTP/1.1 200 OK\r\n"
            "Content-Type: %s\r\n"
            "Content-Length: %ld\r\n"
            "Connection: close\r\n"
            "\r\n", type, (long)st.st_size);
        write(fd, header, header_len);

        // 发送文件内容（sendfile 更高效）
        off_t offset = 0;
        while (offset < st.st_size) {
            ssize_t sent = sendfile(fd, file_fd, &offset, st.st_size - offset);
            if (sent <= 0) break;
        }

        close(file_fd);
    }

    close(fd);
    free(ctx);
}

int main() {
    Reactor *r = reactor_create(PORT);

    // 注册监听 socket 的回调（新连接）
    // 实际需要单独处理 accept

    printf("HTTP server on http://localhost:%d\n", PORT);
    printf("Document root: %s\n", DOC_ROOT);
    reactor_run(r);
    reactor_destroy(r);
    return 0;
}

4. sendfile：零拷贝传输

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#include <sys/sendfile.h>

// 从文件直接发送到 socket，无需经过用户空间
ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count);

// 使用示例
int file_fd = open("large_file.bin", O_RDONLY);
struct stat st;
fstat(file_fd, &st);

off_t offset = 0;
while (offset < st.st_size) {
    ssize_t sent = sendfile(conn_fd, file_fd, &offset,
                            st.st_size - offset);
    if (sent <= 0) break;
}

sendfile 优势：

1
2
3
4
5
6
7
8
9
传统方式：
  磁盘 → kernel buffer → user buffer → socket buffer → 网络
  (DMA)     (CPU copy)    (CPU copy)     (DMA)

sendfile：
  磁盘 → kernel buffer → socket buffer → 网络
  (DMA)                  (DMA)

零拷贝：减少一次 CPU 内存复制

5. 性能压测

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# === wrk（现代 HTTP 压测工具）===
wrk -t4 -c1000 -d30s http://localhost:8080/
# -t: 线程数
# -c: 并发连接数
# -d: 持续时间

# === ab（Apache Bench）===
ab -n 10000 -c 100 http://localhost:8080/

# === 自定义压测脚本 ===
# 写一个 C 程序创建 10000 个连接并发请求

# === 监控工具 ===
htop           # CPU/内存
pidstat -p $(pgrep server) 1  # 每 1 秒打印进程统计
sar -n DEV 1   # 网络接口统计

5.1 性能指标解读

1
2
3
4
5
6
7
8
9
QPS (Queries Per Second)：每秒请求数
Latency：延迟（平均 / P50 / P95 / P99）
Throughput：吞吐量（bytes/sec）
Error Rate：错误率

典型单机性能参考（优化后的 epoll 服务器）：
  - 静态小文件：10k-50k QPS
  - 静态大文件：吞吐受限于网卡（1Gbps ≈ 125MB/s）
  - 简单 echo：100k+ QPS

6. 常见性能瓶颈

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
1. 锁竞争
   - 全局数据结构用全局锁 → 多线程下争用
   - 解决：per-thread 数据结构，无锁队列

2. 内存分配
   - 频繁 malloc/free → 内存碎片
   - 解决：内存池（slab allocator）

3. 系统调用
   - 频繁 small write → 每次 syscall 有开销
   - 解决：writev（聚集写），缓冲合并

4. 协议解析
   - 字符串查找/解析慢
   - 解决：状态机，SIMD 指令

5. 上下文切换
   - 线程过多 → 内核调度开销
   - 解决：epoll + 适量工作线程（CPU 核数 × 2）
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发、Linux开发、
公众号开发、开源爱好者、Linux

联系QQ:562054870