133｜Redis 单线程与多线程：事件循环、I/O 多线程和高性能原因

┌─────────────────────────────────────────────────────────────────┐
│                    Redis 主线程（单线程）                          │
│                                                                   │
│  ┌──────────┐  ┌──────────┐  ┌──────────────┐  ┌──────────┐    │
│  │ 接收连接  │→│ 解析命令  │→│ 执行命令(读写) │→│ 返回响应  │    │
│  │ accept() │  │  RESP    │  │ 内存操作      │  │ write()  │    │
│  └──────────┘  └──────────┘  └──────────────┘  └──────────┘    │
│                                                                   │
│  后台线程（BIO，非核心路径，不参与命令执行）：                       │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐           │
│  │ close 文件    │  │ AOF fsync    │  │ UNLINK 异步删除│          │
│  │ (BIO_CLOSE)  │  │ (BIO_FSYNC)  │  │ (BIO_LAZY)   │           │
│  └──────────────┘  └──────────────┘  └──────────────┘           │
└─────────────────────────────────────────────────────────────────┘

typedef struct aeEventLoop {
    int maxfd;                  // 当前注册的最大文件描述符（用于 epoll 边界判断）
    int setsize;                // 最大可监听的文件描述符数（默认 10128，即 maxclients + 128）
    long long timeEventNextId;  // 下一个时间事件的 ID（自增）
    aeFileEvent *events;        // 已注册的文件事件数组，下标为 fd
    aeFiredEvent *fired;        // 已触发的文件事件数组（epoll_wait 的结果）
    aeTimeEvent *timeEventHead; // 时间事件双向链表头
    int stop;                   // 事件循环停止标志
    void *apidata;              // 多路复用库的私有数据（epoll fd / kqueue fd 等）
    aeBeforeSleepProc *beforesleep; // 每次进入 epoll_wait 前的回调
    aeBeforeSleepProc *aftersleep;  // 每次 epoll_wait 返回后的回调
    int flags;                  // 标志位
} aeEventLoop;

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
        aeProcessEvents(eventLoop, AE_ALL_EVENTS|
                                   AE_CALL_BEFORE_SLEEP|
                                   AE_CALL_AFTER_SLEEP);
    }
}

int aeProcessEvents(aeEventLoop *eventLoop, int flags) {
    int processed = 0;

    // 1. 找到最近的时间事件，计算最短睡眠时间
    long long shortest = getShortestTimeEvent(eventLoop);
    long long ms_until_timer = shortest - now_ms();

    // 2. 如果没有任何事件（闲时），阻塞等待
    if (ms_until_timer > 0) {
        tvp.tv_sec = ms_until_timer / 1000;
        tvp.tv_usec = (ms_until_timer % 1000) * 1000;
    } else {
        tvp.tv_sec = 0;
        tvp.tv_usec = 0;  // 不阻塞，立即返回
    }

    // 3. beforesleep 回调
    if (flags & AE_CALL_BEFORE_SLEEP)
        eventLoop->beforesleep(eventLoop);

    // 4. 等待文件事件
    numevents = aeApiPoll(eventLoop, &tvp);

    // 5. aftersleep 回调
    if (flags & AE_CALL_AFTER_SLEEP)
        eventLoop->aftersleep(eventLoop);

    // 6. 处理触发的文件事件
    for (int j = 0; j < numevents; j++) {
        aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
        int mask = eventLoop->fired[j].mask;
        int fd = eventLoop->fired[j].fd;

        // 先处理读事件，再处理写事件
        if (fe->mask & mask & AE_READABLE)
            fe->rfileProc(eventLoop, fd, fe->clientData, mask);
        if (fe->mask & mask & AE_WRITABLE)
            fe->wfileProc(eventLoop, fd, fe->clientData, mask);
        processed++;
    }

    // 7. 处理到期的时间事件
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);

    return processed;
}

> **图 3**：Redis 事件循环的完整执行流程——6 个步骤构成了 Redis 的运行心跳   

typedef struct aeFileEvent {
    int mask;                  // AE_READABLE(1) | AE_WRITABLE(2) | AE_BARRIER(4)
    aeFileProc *rfileProc;    // 可读回调（readQueryFromClient）
    aeFileProc *wfileProc;    // 可写回调（sendReplyToClient）
    void *clientData;         // 指向 client 结构体
} aeFileEvent;

typedef struct aeTimeEvent {
    long long id;              // 事件 ID（自增）
    monotime when;             // 触发时间戳（毫秒级单调时钟）
    aeTimeProc *timeProc;      // 回调函数，返回下次触发的毫秒间隔
    aeEventFinalizerProc *finalizerProc; // 清理回调（删除事件时调用）
    void *clientData;
    struct aeTimeEvent *prev;  // 双向链表
    struct aeTimeEvent *next;
} aeTimeEvent;

// aeProcessEvents 中的饥饿防护逻辑
int maxIterations = 1024*1024;  // 防止无限循环
while (numevents--) {
    // 处理文件事件...
    if (--maxIterations == 0) break;  // 安全阀
}

// networking.c 中的简化逻辑
ssize_t _writevToClient(client *c, struct iovec *iov, int iovcnt) {
    ssize_t nwritten = 0;
    while (nwritten < totlen) {
        nwritten = writev(c->fd, iov, iovcnt);
        // writev 将多个不连续的缓冲区一次性写入 socket
        // 避免了将多个缓冲区拼接成一个大缓冲区的拷贝操作
    }
}

        方案 A：阻塞 I/O                      方案 C：多路复用
   ┌──────────┐                         ┌──────────┐
   │  read()  │ 阻塞等待                  │epoll_wait│ 等待任意就绪
   │  client1 │────────────────         │          │
   └──────────┘                         └────┬─────┘
         │  client2 被饿死                     │
         ▼                                    ▼
   (永远不会处理 client2)              ┌───────┴───────┐
                                      │ 只处理就绪的 fd │
                                      │ client2,client5│
                                      └───────────────┘

// ae_epoll.c 中的核心封装

// 创建 epoll 实例
static int aeApiCreate(aeEventLoop *eventLoop) {
    aeApiState *state = zmalloc(sizeof(aeApiState));
    state->epfd = epoll_create(1024);  // 初始大小 1024，内核会动态扩容
    eventLoop->apidata = state;
    return 0;
}

// 注册 fd 事件（ADD 或 MOD）
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0};
    int op = eventLoop->events[fd].mask == AE_NONE
             ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    ee.events = 0;
    mask |= eventLoop->events[fd].mask;  // 合并新旧 mask
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    ee.data.fd = fd;
    epoll_ctl(state->epfd, op, fd, &ee);
    return 0;
}

// 删除 fd 事件
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask) {
    aeApiState *state = eventLoop->apidata;
    struct epoll_event ee = {0};
    int mask = eventLoop->events[fd].mask & (~delmask);
    ee.events = 0;
    if (mask & AE_READABLE) ee.events |= EPOLLIN;
    if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
    ee.data.fd = fd;
    if (mask != AE_NONE) {
        epoll_ctl(state->epfd, EPOLL_CTL_MOD, fd, &ee);
    } else {
        epoll_ctl(state->epfd, EPOLL_CTL_DEL, fd, &ee);
    }
}

// 等待事件
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    aeApiState *state = eventLoop->apidata;
    int retval = epoll_wait(state->epfd, state->events, eventLoop->setsize,
                            tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
    for (int j = 0; j < retval; j++) {
        int mask = 0;
        if (state->events[j].events & EPOLLIN)  mask |= AE_READABLE;
        if (state->events[j].events & EPOLLOUT) mask |= AE_WRITABLE;
        if (state->events[j].events & EPOLLERR) mask |= AE_WRITABLE;  // 错误也当作可写处理
        if (state->events[j].events & EPOLLHUP) mask |= AE_WRITABLE;  // 挂起也当作可写处理
        eventLoop->fired[j].fd = state->events[j].data.fd;
        eventLoop->fired[j].mask = mask;
    }
    return retval;
}

                     Redis 6.0 多线程 I/O 架构

  ┌──────────────────────────────────────────────────────────────┐
  │                      主线程（事件循环）                         │
  │                                                              │
  │  ┌─────────┐   ┌─────────┐   ┌──────────┐   ┌───────────┐  │
  │  │epoll    │   │分配客户端│   │执行命令   │   │分配客户端  │  │
  │  │收集就绪  │→  │到 I/O   │→  │(单线程)   │→  │到 I/O     │  │
  │  │连接     │   │线程读   │   │           │   │线程写     │  │
  │  └─────────┘   └────┬────┘   └──────────┘   └────┬──────┘  │
  │                     │                             │          │
  └─────────────────────┼─────────────────────────────┼──────────┘
                        │                             │
         ┌──────────────┼──────────────┐              │
         │              │              │              │
         ▼              ▼              ▼              ▼
   ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐
   │ I/O线程1  │  │ I/O线程2  │  │ I/O线程3  │  │ I/O线程4  │
   │          │  │          │  │          │  │          │
   │ read()   │  │ read()   │  │ read()   │  │ write()  │
   │ 解析协议  │  │ 解析协议  │  │ 解析协议  │  │ 发送响应  │
   └──────────┘  └──────────┘  └──────────┘  └──────────┘

# redis.conf
io-threads 4               # I/O 线程数（包含主线程，所以实际子线程数 = 4-1 = 3）
io-threads-do-reads yes     # 是否在读阶段也使用多线程（默认 yes，写阶段始终多线程）

int handleClientsWithPendingWritesUsingThreads(void) {
    // 如果待写客户端数量少于 I/O 线程数的 2 倍，直接用单线程写
    // 避免线程调度的开销大于并行收益
    if (server.clients_pending_write < server.io_threads_num * 2) {
        return handleClientsWithPendingWrites();  // 回退到单线程
    }

    // 1. 主线程将待写客户端均匀分配到各 I/O 线程
    listIter li;
    listNode *ln;
    int item_id = 0;
    listRewind(server.clients_pending_write, &li);
    while ((ln = listNext(&li)) != NULL) {
        client *c = listNodeValue(ln);
        int target_id = item_id % server.io_threads_num;
        listAddNodeTail(io_threads_list[target_id], c);
        item_id++;
    }

    // 2. 设置 I/O 线程的待处理数量，唤醒它们
    io_threads_op = IO_THREADS_OP_WRITE;
    for (int j = 1; j < server.io_threads_num; j++)
        io_threads_pending[j] = listLength(io_threads_list[j]);

    // 3. 主线程也处理自己的一份（id=0）
    listRewind(io_threads_list[0], &li);
    while ((ln = listNext(&li)) != NULL) {
        client *c = listNodeValue(ln);
        writeToClient(c, 0);
    }

    // 4. 自旋等待所有 I/O 线程完成
    while (1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += io_threads_pending[j];
        if (pending == 0) break;
    }

    // 5. 清理
    listRewind(server.clients_pending_write, &li);
    while ((ln = listNext(&li)) != NULL) {
        client *c = listNodeValue(ln);
        // 检查是否还有未发送完的数据，重新加入待写队列
        if (clientHasPendingReplies(c)) {
            installClientWriteHandler(c);
        }
    }
    listEmpty(server.clients_pending_write);
    return 1;
}

void *IOThreadMain(void *myid) {
    long id = (unsigned long)myid;
    while (1) {
        // 自旋等待主线程分配任务（忙等待 100 万次）
        for (int j = 0; j < 1000000; j++) {
            if (io_threads_pending[id] != 0) break;
        }
        if (io_threads_pending[id] == 0) {
            // 长时间无任务则休眠（通过互斥锁 + 条件变量等待）
            pthread_mutex_lock(&io_threads_mutex[id]);
            pthread_mutex_unlock(&io_threads_mutex[id]);
            continue;
        }

        // 处理分配给自己的客户端列表
        listIter li;
        listNode *ln;
        listRewind(io_threads_list[id], &li);
        while ((ln = listNext(&li)) != NULL) {
            client *c = listNodeValue(ln);
            if (io_threads_op == IO_THREADS_OP_WRITE) {
                writeToClient(c, 0);
            } else if (io_threads_op == IO_THREADS_OP_READ) {
                readQueryFromClient(c->conn);
            }
        }
        listEmpty(io_threads_list[id]);
        io_threads_pending[id] = 0;  // 通知主线程本线程完成
    }
}

  时间线：
  ────────────────────────────────────────────────────────────→

  主线程:  [epoll]  [分配]  [等待]  [执行命令]  [分配]  [等待]
                        │                         │
  I/O线程1:          [read]                    [write]
  I/O线程2:          [read]                    [write]
  I/O线程3:          [read]                    [write]

  阶段:    收集就绪   并行读    同步点   单线程执行   并行写   同步点

-- 这段简单的 Lua 脚本依赖单线程原子性
local old = redis.call('GET', KEYS[1])
if old < ARGV[1] then
    redis.call('SET', KEYS[1], ARGV[1])
end
return old

// bio.c 中的定义
#define BIO_CLOSE_FILE    0  // 关闭文件（close() 可能阻塞）
#define BIO_AOF_FSYNC     1  // AOF 文件刷盘（fsync() 可能阻塞数百毫秒）
#define BIO_LAZY_FREE     2  // 惰性删除（UNLINK，释放大对象内存）
#define BIO_NUM_OPS       3

// 主线程提交任务
void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3) {
    struct bio_job *job = zmalloc(sizeof(*job));
    job->arg1 = arg1;
    job->arg2 = arg2;
    job->arg3 = arg3;
    pthread_mutex_lock(&bio_mutex[type]);
    listAddNodeTail(bio_jobs[type], job);
    bio_pending[type]++;  // 计数器 +1
    pthread_cond_signal(&bio_newjob_cond[type]);  // 唤醒 BIO 线程
    pthread_mutex_unlock(&bio_mutex[type]);
}

// BIO 线程的执行循环
void *bioProcessBackgroundJobs(void *arg) {
    int type = (long)arg;
    while (1) {
        pthread_mutex_lock(&bio_mutex[type]);
        while (listLength(bio_jobs[type]) == 0) {
            pthread_cond_wait(&bio_newjob_cond[type], &bio_mutex[type]);
        }
        job = listFirst(bio_jobs[type]);
        pthread_mutex_unlock(&bio_mutex[type]);
        // 执行实际任务...
    }
}

              Redis 高性能原因全景

  ┌─────────────────────────────────────────────────────────────────┐
  │                         Redis 高性能                             │
  │                                                                 │
  │  ┌───────────┐  ┌───────────┐  ┌───────────────┐               │
  │  │ ①纯内存操作 │  │②高效数据结构│  │③ I/O 多路复用  │               │
  │  │           │  │           │  │               │               │
  │  │ · ns 级读写│  │ · SDS     │  │ · epoll/kqueue│               │
  │  │ · 无磁盘  │  │ · dict    │  │ · 事件驱动    │               │
  │  │   I/O 延迟│  │ · skiplist│  │ · 非阻塞 I/O  │               │
  │  │ · 零拷贝  │  │ · listpack│  │ · 水平触发    │               │
  │  └───────────┘  └───────────┘  └───────────────┘               │
  │                                                                 │
  │  ┌───────────┐  ┌───────────┐  ┌───────────────┐               │
  │  │④单线程执行  │  │⑤ 简洁协议  │  │ ⑥高效编码     │               │
  │  │           │  │           │  │               │               │
  │  │ · 无锁竞争 │  │ · RESP    │  │ · C 语言     │               │
  │  │ · 无上下文│  │ · 二进制安全│  │ · 精心优化   │               │
  │  │   切换    │  │ · 解析极快 │  │ · 极简代码   │               │
  │  │ · 天然原子│  │ · 无状态   │  │ · jemalloc   │               │
  │  └───────────┘  └───────────┘  └───────────────┘               │
  │                                                                 │
  │  ┌─────────────────────────────────────────────┐               │
  │  │ ⑦ Pipeline + 多线程 I/O (6.0+)              │               │
  │  │                                             │               │
  │  │ · Pipeline 批量发送，减少 RTT                │               │
  │  │ · 多线程 I/O 并行读写，多核利用               │               │
  │  │ · 两者互补：Pipeline 减少往返，多线程 I/O    │               │
  │  │   加速单次往返中的读写                       │               │
  │  └─────────────────────────────────────────────┘               │
  └─────────────────────────────────────────────────────────────────┘

+ 简单字符串   - 错误       : 整数
$ 批量字符串   * 数组       _ null
# 布尔        , 双精度     ( 大数
! 批量错误     = 逐字字符串  % 映射
~ 集合        > 推送

无 Pipeline：          有 Pipeline：
Client    Server       Client    Server
  │─ cmd1 ─→│             │─ cmd1 ─→│
  │←─ rsp1 ─│             │─ cmd2 ─→│
  │─ cmd2 ─→│             │─ cmd3 ─→│
  │←─ rsp2 ─│             │←─ rsp1 ─│
  │─ cmd3 ─→│             │←─ rsp2 ─│
  │←─ rsp3 ─│             │←─ rsp3 ─│
  3x RTT + 3x 系统调用     1x RTT + 1x 批量系统调用

你的 Redis QPS 是否超过 8 万？
├── 否 → 保持单线程（默认配置）
└── 是 → 是否大量使用 Pipeline？
    ├── 是 → Pipeline 已提供批量处理，多线程 I/O 收益有限，保持单线程
    └── 否 → 服务器 CPU 核心数？
        ├── < 4 核 → 多线程收益被线程切换抵消，保持单线程
        └── ≥ 4 核 → 开启多线程 I/O
            io-threads 4-6
            io-threads-do-reads yes

# redis.conf 推荐配置
io-threads 4                      # 4 核起步，设为 CPU 核心数的 1/2 到 3/4
io-threads-do-reads yes           # 读阶段也开启多线程

# 在 redis-cli 中验证配置
CONFIG GET io-threads
CONFIG GET io-threads-do-reads

# 压测验证效果
redis-benchmark -t set,get -n 1000000 -c 50 -d 256 --threads 4

lazyfree-lazy-eviction yes
lazyfree-lazy-expire yes
lazyfree-lazy-server-del yes

taskset -c 0-3 redis-server redis.conf

# 1. 开启多线程 I/O（4 核机器）
io-threads 4
io-threads-do-reads yes

# 2. 滚动重启 Redis 实例

客户端 A: GET stock → 10
客户端 B: GET stock → 10
客户端 A: SET stock 9  (10-1)
客户端 B: SET stock 9  (10-1)  ← 本应扣到 8！