1
// Redis 事件循环（简化）
2
void aeMain(aeEventLoop *eventLoop) {
3
    eventLoop->stop = 0;
4
    while (!eventLoop->stop) {
5
        // 1. 处理到期的定时事件
6
        aeProcessTimeEvents(eventLoop);
7

8
        // 2. 等待并处理文件事件
9
        aeProcessEvents(eventLoop, AE_ALL_EVENTS);
10
    }
11
}

1
// 根据平台选择最优实现
2
#ifdef HAVE_EVPORT
3
#include "ae_evport.c"
4
#else
5
    #ifdef HAVE_EPOLL
6
    #include "ae_epoll.c"
7
    #else
8
        #ifdef HAVE_KQUEUE
9
        #include "ae_kqueue.c"
10
        #else
11
        #include "ae_select.c"
12
        #endif
13
    #endif
14
#endif

1
// epoll API
2
int epfd = epoll_create(1024);  // 创建 epoll 实例
3

4
// 添加监听事件
5
struct epoll_event ee;
6
ee.events = EPOLLIN;  // 监听可读事件
7
ee.data.fd = client_fd;
8
epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &ee);
9

10
// 等待事件
11
struct epoll_event events[1024];
12
int n = epoll_wait(epfd, events, 1024, timeout);
13
for (int i = 0; i < n; i++) {
14
    // 处理就绪的 fd
15
    handleEvent(events[i]);
16
}

1
// 文件事件类型
2
#define AE_READABLE 1   // 可读
3
#define AE_WRITABLE 2   // 可写
4

5
// 事件处理器结构
6
typedef struct aeFileEvent {
7
    int mask;           // 事件类型掩码
8
    aeFileProc *rfileProc;  // 读事件处理器
9
    aeFileProc *wfileProc;  // 写事件处理器
10
    void *clientData;   // 客户端数据
11
} aeFileEvent;

1
// Redis 客户端结构（简化）
2
typedef struct client {
3
    int fd;                 // 套接字描述符
4
    redisDb *db;            // 当前数据库
5
    robj *name;             // 客户端名称
6
    sds querybuf;           // 查询缓冲区
7
    int argc;               // 参数个数
8
    robj **argv;            // 参数数组
9
    struct redisCommand *cmd; // 要执行的命令
10

11
    // 输出缓冲区
12
    int bufpos;             // 缓冲区位置
13
    char buf[PROTO_REPLY_CHUNK_BYTES]; // 固定缓冲区
14
    list *reply;            // 可变缓冲区列表
15

16
    // 状态标志
17
    int flags;
18

19
    // 认证信息
20
    int authenticated;
21

22
    // 事务
23
    multiState mstate;
24

25
    // 订阅
26
    dict *pubsub_channels;
27
    list *pubsub_patterns;
28
} client;

1
*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n
2

3
解析为：
4
["SET", "key", "value"]

1
// 解析命令（简化）
2
int processMultibulkBuffer(client *c) {
3
    char *newline = NULL;
4
    int pos = 0, ok;
5
    long ll;
6

7
    // 解析数组长度 *<count>\r\n
8
    if (c->multibulklen == 0) {
9
        newline = strchr(c->querybuf, '\r');
10
        ok = string2ll(c->querybuf + 1, newline - c->querybuf - 1, &ll);
11
        c->multibulklen = ll;
12
        c->argc = ll;
13
        c->argv = zmalloc(sizeof(robj*) * c->argc);
14
        pos = newline - c->querybuf + 2;
15
    }
16

17
    // 解析每个参数
18
    while (c->multibulklen) {
19
        // 解析参数长度 $<len>\r\n
20
        // 解析参数数据 <data>\r\n
21
        // ...
22
    }
23

24
    return C_OK;
25
}

1
// 命令表（哈希表）
2
struct redisCommand redisCommandTable[] = {
3
    {"get", getCommand, 2, "r", 0, NULL, 1, 1, 1, 0, 0},
4
    {"set", setCommand, -3, "wm", 0, NULL, 1, 1, 1, 0, 0},
5
    {"del", delCommand, -2, "w", 0, NULL, 1, -1, 1, 0, 0},
6
    // ...
7
};
8

9
// 查找命令
10
struct redisCommand *lookupCommand(sds name) {
11
    return dictFetchValue(server.commands, name);
12
}

1
struct redisCommand {
2
    char *name;           // 命令名
3
    redisCommandProc *proc; // 命令实现函数
4
    int arity;            // 参数个数，-N 表示至少 N 个
5
    char *sflags;         // 字符串标志
6
    int flags;            // 整数标志
7
    redisGetKeysProc *getkeys_proc; // 获取键的函数
8
    int firstkey;         // 第一个键的位置
9
    int lastkey;          // 最后一个键的位置
10
    int keystep;          // 键的步长
11
    long long microseconds, calls; // 统计信息
12
};

1
// SET 命令实现（简化）
2
void setCommand(client *c) {
3
    robj *key = c->argv[1];
4
    robj *val = c->argv[2];
5

6
    // 解析选项：NX, XX, EX, PX
7
    int flags = 0;
8
    long long milliseconds = 0;
9

10
    for (int j = 3; j < c->argc; j++) {
11
        if (!strcasecmp(c->argv[j]->ptr, "nx")) {
12
            flags |= SET_NX;
13
        } else if (!strcasecmp(c->argv[j]->ptr, "xx")) {
14
            flags |= SET_XX;
15
        } else if (!strcasecmp(c->argv[j]->ptr, "ex")) {
16
            // EX 秒
17
        } else if (!strcasecmp(c->argv[j]->ptr, "px")) {
18
            // PX 毫秒
19
        }
20
    }
21

22
    // 设置键值
23
    genericSetKey(c->db, key, val, flags);
24

25
    // 设置过期时间
26
    if (milliseconds) {
27
        setExpire(c->db, key, mstime() + milliseconds);
28
    }
29

30
    // 发送响应
31
    addReply(c, shared.ok);
32
}

1
// 内存淘汰策略
2
#define MAXMEMORY_FLAG_LRU (1<<0)
3
#define MAXMEMORY_FLAG_LFU (1<<1)
4
#define MAXMEMORY_FLAG_ALLKEYS (1<<2)
5

6
// 淘汰策略
7
typedef enum {
8
    MAXMEMORY_VOLATILE_LRU,
9
    MAXMEMORY_VOLATILE_LFU,
10
    MAXMEMORY_VOLATILE_TTL,
11
    MAXMEMORY_VOLATILE_RANDOM,
12
    MAXMEMORY_ALLKEYS_LRU,
13
    MAXMEMORY_ALLKEYS_LFU,
14
    MAXMEMORY_ALLKEYS_RANDOM,
15
    MAXMEMORY_NO_EVICTION
16
} maxmemory_policy_t;
17

18
// 执行淘汰
19
int freeMemoryIfNeeded(void) {
20
    while (server.maxmemory &&
21
           zmalloc_used_memory() > server.maxmemory) {
22
        // 根据策略选择要删除的键
23
        // 删除键
24
    }
25
    return C_OK;
26
}

1
// 添加响应
2
void addReply(client *c, robj *obj) {
3
    // 检查是否需要安装写处理器
4
    if (prepareClientToWrite(c) != C_OK) return;
5

6
    // 根据响应大小选择缓冲区
7
    if (sdslen(obj->ptr) < PROTO_REPLY_CHUNK_BYTES) {
8
        // 使用固定缓冲区
9
        if (_addReplyToBuffer(c, obj->ptr, sdslen(obj->ptr)) != C_OK) {
10
            // 失败则使用可变缓冲区
11
            _addReplyStringToList(c, obj->ptr, sdslen(obj->ptr));
12
        }
13
    } else {
14
        // 使用可变缓冲区
15
        _addReplyStringToList(c, obj->ptr, sdslen(obj->ptr));
16
    }
17
}

1
// 发送响应
2
void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
3
    client *c = privdata;
4
    int nwritten = 0, totwritten = 0;
5

6
    while (c->bufpos > 0 || listLength(c->reply)) {
7
        // 发送固定缓冲区
8
        if (c->bufpos > 0) {
9
            nwritten = write(fd, c->buf + c->sentlen, c->bufpos - c->sentlen);
10
            if (nwritten <= 0) break;
11
            c->sentlen += nwritten;
12
            totwritten += nwritten;
13

14
            if (c->sentlen == c->bufpos) {
15
                c->bufpos = 0;
16
                c->sentlen = 0;
17
            }
18
        }
19

20
        // 发送可变缓冲区
21
        if (listLength(c->reply)) {
22
            // ...
23
        }
24
    }
25

26
    // 所有数据发送完毕，移除写处理器
27
    if (c->bufpos == 0 && listLength(c->reply) == 0) {
28
        aeDeleteFileEvent(el, c->fd, AE_WRITABLE);
29
    }
30
}

1
# 危险操作
2
KEYS *           # O(N)，全量扫描
3
HGETALL bigkey   # 大键，内存重分配
4
DEL bigkey       # 删除大键，阻塞

1
# 安全替代
2
SCAN 0 MATCH pattern COUNT 100  # 分批扫描
3
HSCAN key 0 COUNT 100           # 分批获取
4
UNLINK bigkey                   # 异步删除

1
# redis.conf
2
io-threads 4
3
io-threads-do-reads yes

1
// 多线程读取
2
int handleClientsWithPendingReadsUsingThreads(void) {
3
    // 1. 将客户端分配给 IO 线程
4
    listRewind(server.clients_pending_read, &li);
5
    while ((ln = listNext(&li))) {
6
        client *c = listNodeValue(ln);
7
        int item_id = item_id++ % server.io_threads_num;
8
        listAddNodeTail(io_threads_list[item_id], c);
9
    }
10

11
    // 2. 唤醒 IO 线程
12
    for (int j = 1; j < server.io_threads_num; j++) {
13
        int processed = listLength(io_threads_list[j]);
14
        io_threads_pending[j] = processed;
15
    }
16

17
    // 3. 主线程也处理一部分
18
    while (listLength(io_threads_list[0])) {
19
        client *c = listFirst(io_threads_list[0])->value;
20
        readQueryFromClient(NULL, c->fd, c, 0);
21
    }
22

23
    // 4. 等待所有 IO 线程完成
24
    while (1) {
25
        int pending = 0;
26
        for (int j = 1; j < server.io_threads_num; j++)
27
            pending += io_threads_pending[j];
28
        if (!pending) break;
29
    }
30

31
    return processed;
32
}

1
# redis.conf
2
appendonly yes
3
appendfsync everysec
4
auto-aof-rewrite-percentage 100
5
auto-aof-rewrite-min-size 64mb

1
# 配置慢查询
2
CONFIG SET slowlog-log-slower-than 10000  # 10 毫秒
3
CONFIG SET slowlog-max-len 128
4

5
# 查看慢查询
6
SLOWLOG GET 10

1
# 开启延迟监控
2
CONFIG SET latency-monitor-threshold 100
3

4
# 查看延迟事件
5
LATENCY LATEST
6
LATENCY HISTORY command

1
# 服务器信息
2
INFO server
3

4
# 内存信息
5
INFO memory
6

7
# 统计信息
8
INFO stats
9

10
# 客户端信息
11
INFO clients