当前位置: 首页 > news >正文

一条Redis命令是如何执行的?

我一开始想这个问题的时候的答案是: 客户端发送命令给服务端, 服务端收到执行之后再处理将命令执行结果返回给客户端。那更细节的过程呢?
在这里插入图片描述

那么在了解一条Redis命令是如何执行之前,我们首先来看看Redis的架构。
在这里插入图片描述

Redis 的部署架构(宏观视角)

  1. 单机模式基础部署形态,单节点运行,无高可用保障,适用于开发测试或非核心场景。
  2. 主从复制(Master-Slave)数据从主节点同步至多个从节点,提供读写分离能力,增强读性能与数据冗余。
  3. 哨兵模式(Sentinel)在主从复制基础上,引入监控节点自动故障转移,实现高可用(HA),当主节点故障时自动选举新主节点。
  4. 集群模式(Cluster)分布式架构,数据分片存储在多个节点,支持自动故障转移与水平扩展,适用于海量数据与高并发场景。

Redis 的核心组件(微观视角)

  1. 事件驱动引擎:基于 I/O 多路复用(如 epoll)实现的高性能网络模型,单线程处理并发请求,避免线程切换开销。
  2. 命令处理层:解析客户端命令,执行对应操作逻辑(如 GET/SET/HSET),支持丰富的数据结构(String、Hash、List 等)。
  3. 内存管理系统:负责内存分配与回收,支持内存淘汰策略(如 LRU、LFU),优化内存使用效率
  4. 持久化模块- RDB(快照):定期生成二进制快照文件,恢复速度快但可能丢失部分数据。- AOF(日志):追加写命令到日志文件,数据安全性高,支持重写机制压缩日志。
  5. 监控与统计系统提供运行状态监控(如内存使用、QPS)、慢查询分析、性能指标采集等功能,辅助运维优化。

这些核心组件的作用:
在这里插入图片描述

主要的名词解释:

  • Redis客户端:与Redis服务器交互的程序或工具(如Jedis、Lettuce、redis-cli等); 能够通过TCP协议连接Redis服务器(默认端口6379),使用RESP(Redis Serialization Protocol)协议通信,发送命令(如SET, GET)并接收响应。
  • 事件驱动层:单线程Reactor模式(6.0后引入多线程IO); 组成为:文件事件处理器:通过I/O多路复用(如epoll/kqueue)监听套接字;时间事件处理器:处理定时任务(如过期键清理)。

在这里插入图片描述

  • 命令层:解析客户端请求(将RESP协议转换为内存数据结构)、校验命令合法性、执行命令逻辑(Get、Set等)。
  • 内存分配/回收:使用jemalloc/tcmalloc代替系统malloc(减少碎片),内存淘汰策略(LRU/LFU/random/TTL等)。
  • RDB与AOF:Redis提供的持久化策略,以保证数据可靠性。
  • 副本(Replaction):Redis通过副本,实现【主-从】运行模式,是故障切换的基石,用于提高系统运行可靠性,支持读写分离、提高性能。哨兵(Sentinel):哨兵用于支持故障时,主从节点自动切换。
  • 哨兵:为Redis高可用提高了保证。
  • 集群(Cluster):Redis基于数据分片,支持横向拓展的一种高性能模式。主节点负责数据存储,从节点作备份。
  • 监控与统计:Redis提供监控信息和性能分析工具,包括内存使用(used_memory)、命令统计(commandstats)等。
    核心模块:
    在这里插入图片描述

在分析命令是如何执行之前,我们需要关注Redis最核心的模块-事件驱动, 事件驱动也是Redis高性能的基石。

1、Reactor模式实现

// 核心数据结构:aeEventLoop
typedef struct aeEventLoop {int maxfd;                // 最大文件描述符aeFileEvent *events;      // 注册的文件事件数组aeTimeEvent *timeEventHead; // 时间事件链表aeFiredEvent *fired;      // 触发事件数组
} aeEventLoop;
  1. 事件处理的三要素
  1. 事件注册 → 2. 多路复用监听 → 3. 事件派发

Redis注册的事件可以分为: 文件事件(fileEvent) 和 时间事件 (timeEvent)
在这里插入图片描述

3、关键代码执行流
a. 新建连接事件在Redis启动时注册,当Redis收到新建连接请求后,会调用 acceptTcpHandler


void initServer(void) {if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK) {serverPanic("Unrecoverable error creating TCP socket accept handler.");}
}

b. 读事件处理函数 readQueryFromClient,在新建连接时注册。写事件处理函数 sendReplyToClient 在发送执行结果时注册。


// 读事件处理函数。新建连接时注册
connSetReadHandler(conn, readQueryFromClient);
// 写事件处理函数。单次事件循环,无法发完数据时注册
connSetWriteHandler(c->conn, sendReplyToClient)

c. 在Redis启动后,进入事件循环 aeMain

void aeMain(aeEventLoop *eventLoop) {eventLoop->stop = 0;while (!eventLoop->stop) {// 事件循环处理函数// 关注读、写、时间事件aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_BEFORE_SLEEP|AE_CALL_AFTER_SLEEP);}
}

d. 单次事件循环 aeProcessEvents 函数简化后,执行流程如下。

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{int processed = 0, numevents;if (eventLoop->maxfd != -1 ||((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {// 事件触发前执行函数 beforeSleepif (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP)eventLoop->beforesleep(eventLoop);// 获取触发事件numevents = aeApiPoll(eventLoop, tvp);// 事件触发后执行函数 afterSleepif (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)eventLoop->aftersleep(eventLoop);// 循环处理事件for (j = 0; j < numevents; j++) {aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];// 执行读事件回调函数 rfileProcif (fe->mask & mask & AE_READABLE)fe->rfileProc(eventLoop,fd,fe->clientData,mask);// 执行写事件回调函数 wfileProcif (fe->mask & mask & AE_WRITABLE)fe->wfileProc(eventLoop,fd,fe->clientData,mask);}}// 时间事件if (flags & AE_TIME_EVENTS)processed += processTimeEvents(eventLoop);return processed;
}
// 其中 beforeSleep 函数。在每次事件触发前,会执行一些特定功能。

补充:

  1. beforeSleep核心作用:
    • 将命令回复缓冲区数据写入客户端(handleClientsWithPendingWrites)。
    • 集群模式下发送心跳包。
    • Module系统的事件钩子执行。
  2. serverCron时间事件:
    • 每100ms执行一次(可配置)。
    • 执行过期键清理、持久化触发、主从重连、集群故障检测等。
  3. 写事件注册策略:
    • 延迟注册:默认不注册写事件,仅在输出缓冲区满时注册(installClientWriteHandler)。
    • 一次性触发:发送完成后立即取消写事件监听,避免空转。
  4. 多线程IO扩展(Redis 6.0+):
    • 主线程:仍负责命令执行和事件调度。
    • IO线程:在aftersleep阶段处理解析后的命令,通过postponeClientRead分流读操作。

了解了事件驱动后,我们现在来看,一条Redis命令是如何执行的。
我给出示意图,方便理解。
在这里插入图片描述

1、 建立连接阶段:
客户端发起请求,由 Redis 事件驱动模块 ae 接收。ae 是一个基于 IO 多路复用的 while 无限循环。ae 模块在接收连接请求后,会触发「新建连接事件」,由 「acceptTcpHandler」 函数执行。该函数负责接收连接、新建连接,以及初始化 client 数据结构。

// 核心代码路径aeMain() → aeProcessEvents() → acceptTcpHandler() → createClient() → connSetReadHandler(conn, readQueryFromClient)
  1. 读and解析 阶段:

Redis 收到命令后,触发 ae 模块「读事件」,进入「readQueryFromClient」执行流程。该流程判断是否启用 IO 多线程,选择以下两条分支之一。

  • 若启用,则主线程将该连接客户端加入「clients_pending_read」读就绪队列,并将客户端 flag 标记为「CLIENT_PENDING_READ」,表示可读。下一次循环时,会将 clients_pending_read 队列分发给 IO 线程和主线程,执行读取请求、解析命令等操作。最终,由主线程执行命令。
  • 若未启用,则主线程「独自」执行读取命令、解析命令、执行命令、发送结果等全部流程
    在这里插入图片描述

其中,解析命令流程,会解析客户端发来的请求字符串。具体为以下两个步骤。

  • 找到命令对应的执行函数,放到 client->cmd->proc 中。
  • 解析参数,放到 client->argv、client->argc 中。

Redis 所有命令的执行函数,保存在「redisCommandTable」中。SET 命令对应为「setCommand」。

struct redisCommand redisCommandTable[] = {...{"set",setCommand,-3,"write use-memory @string",0,NULL,1,1,1,0,0,0},...
}

前面我们提到,每次事件循环,Redis 会执行预处理函数「beforeSleep」,该函数内会将 clients_pending_read 读就绪队列进行分发。具体调用函数如下:

int handleClientsWithPendingReadsUsingThreads(void) {// 未开启 IO 线程,直接返回if (!server.io_threads_active || !server.io_threads_do_reads) return 0;...// 否则,分发「读」就绪队列到线程私有队列 io_threads_list[target_id] 中while((ln = listNext(&li))) {client *c = listNodeValue(ln);int target_id = item_id % server.io_threads_num;listAddNodeTail(io_threads_list[target_id],c);item_id++;}...// 主线程执行 io_threads_list[0] 任务listRewind(io_threads_list[0],&li);while((ln = listNext(&li))) {client *c = listNodeValue(ln);readQueryFromClient(c->conn);}listEmpty(io_threads_list[0]);// 主线程等待其它 IO 线程执行任务while(1) {unsigned long pending = 0;for (int j = 1; j < server.io_threads_num; j++)pending += getIOPendingCount(j);if (pending == 0) break;}while(listLength(server.clients_pending_read)) {...// 主线程,执行命令(已读取完成,解析好的命令)。if (processPendingCommandsAndResetClient(c) == C_ERR) {continue;}...}return processed;
}

该函数遍历 clients_pending_read 「读」就绪队列,将「读」任务分发给 IO 线程和主线程的任务队列「io_threads_list」。收到任务后,IO线程和主线程进入「readQueryFromClient」执行流程。注意,本次执行 readQueryFromClient 前,client 状态已被设置为 「CLIENT_PENDING_READ」 ,所以执行时,client 不会再次加入任务队列,而是进入真正的执行流程。

3、 执行命令阶段


int processPendingCommandsAndResetClient(client *c)
processPendingCommandsAndResetClient()
├── processCommand()          // 命令校验(权限/内存/集群等)
├── call()                    // 执行命令前钩子(monitor/watch)
├── c->cmd->proc(c)          // 实际执行命令(如setCommand)
├── propagate()              // 主从复制/AOF传播
└── addReply()               // 响应处理

其中,c->cmd->proc 用来执行真正的命令 setCommand。

执行完命令后,主线程进入最后一步「addReply」,调用 prepareClientToWrite,将执行结果,加入 「clients_pending_write」 写就绪队列中,等待返回客户端。


void addReply(client *c, robj *obj) {// 加入 clients_pending_write 写就绪队列if (prepareClientToWrite(c) != C_OK) return;...
}

在进入下一次事件循环时,beforeSleep 函数,将 clients_pending_write 写就绪队列,分发给 IO 线程和主线程。执行函数如下:

int handleClientsWithPendingWritesUsingThreads(void) {// 如果开启 IO 线程或者客户端连接很少// 主线程直接同步发送结果if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {return handleClientsWithPendingWrites();}...// 否则,分发 clients_pending_write 给 IO 线程和主线程执行while((ln = listNext(&li))) {int target_id = item_id % server.io_threads_num;// 添加到线程任务队列listAddNodeTail(io_threads_list[target_id],c);item_id++;}...// 主线程处理分配给自己的任务,这里是同步执行listRewind(io_threads_list[0],&li);while((ln = listNext(&li))) {client *c = listNodeValue(ln);// 直接发送给客户端writeToClient(c,0);}// 等待 IO 线程执行完毕while(1) {unsigned long pending = 0;for (int j = 1; j < server.io_threads_num; j++)pending += getIOPendingCount(j);if (pending == 0) break;}// 如果同步写数据,没有写完,则注册写事件// 在下一次事件循环中触发listRewind(server.clients_pending_write,&li);while((ln = listNext(&li))) {client *c = listNodeValue(ln);// 注册写事件if (clientHasPendingReplies(c) &&connSetWriteHandler(c->conn, sendReplyToClient) == AE_ERR){freeClientAsync(c);}}listEmpty(server.clients_pending_write);
}

4、响应发送阶段
最终,IO 线程和主线程,通过 writeToClient 函数,将命令执行结果发送给客户端。

http://www.lryc.cn/news/583305.html

相关文章:

  • C++随机打乱函数:简化源码与原理深度剖析
  • 从零开始学前端html篇2
  • 微信小程序控制空调之微信小程序篇
  • 双esp8266-01s间TCP通讯
  • 图像硬解码和软解码
  • RAM带宽计算及分析
  • 区块链系统开发技术应用构建可信数字生态链
  • 以太坊智能合约核心技术解析与应用实践
  • Android 组件内核
  • NO.2数据结构线性表|线性表|顺序表|链表|单链表|双向链表|循环链表|静态链表|插入|删除
  • Qt-翻金币案例
  • 鸿蒙商城开发:ZKmall开源商城系统特性适配与性能优化
  • upload-labs靶场通关详解:第18关 条件竞争
  • CMake构建项目报错“No CUDA toolset found.”
  • android——热修复(补丁)
  • [Pytest][Part 4]多种测试运行方式
  • 三、Docker常用命令
  • 安装nvm管理node.js,详细安装使用教程和详细命令
  • 字体 Unicode 区块字符展示 PDF 生成器
  • 身份证识别api-便捷生活与安全社会的双重保障
  • Ubuntu 下 MySql 使用
  • 【Unity】MiniGame编辑器小游戏(十)连连看【Link】
  • VsCode 接入Continue 远程调用(持续扩展 + DeepSeek R1)— 免本地算力
  • Vim 编辑器常用操作详解(新手快速上手指南)
  • 【Unity】MiniGame编辑器小游戏(十一)消消乐【Crush】
  • 【AI】环境——深度学习cuda+pytorch配置
  • 项目进度管控缺乏闭环,如何形成反馈机制
  • 【c++八股文】Day5:const和constexpr,define
  • 【话题长廊测试报告】
  • 【算法深练】DFS题型拆解:沿着路径“深挖到底”、递归深入、回溯回探的算法解题思路