广东电白建设集团有限公司官方网站,电子产品网页设计,重庆软件开发工资一般多少,淮安网站建设服务运营研发团队 李乐 前言 本文主要讲解服务器处理客户端命令请求的整个流程#xff0c;包括服务器启动监听#xff0c;接收命令请求并解析#xff0c;执行命令请求#xff0c;返回命令回复等#xff0c;这也是本文的主题“命令处理的生命周期”。 Redis服务器作为典型的事件… 运营研发团队 李乐 前言 本文主要讲解服务器处理客户端命令请求的整个流程包括服务器启动监听接收命令请求并解析执行命令请求返回命令回复等这也是本文的主题“命令处理的生命周期”。 Redis服务器作为典型的事件驱动程序事件处理显得尤为重要而Redis将事件分为两大类文件事件与时间事件。文件事件即socket的可读可写事件时间事件用于处理一些需要周期性执行的定时任务本文将对这两种事件作详细介绍。 基本知识 为了更好的理解服务器与客户端的交互还需要学习一些基础知识比如客户端信息的存储Redis对外支持的命令集合客户端与服务器socket读写事件的处理Redis内部定时任务的执行等本小节将对这些知识作简要介绍。 1.1 对象结构体robj简介 Redis是一个Key-Value数据库key只能是字符串value可能是字符串、哈希表、列表、集合和有序集合这5种数据类型用结构体robj表示我们称之为redis对象。结构体robj的type字段表示对象类型5种对象类型在server.h文件定义 #define OBJ_STRING 0
#define OBJ_LIST 1
#define OBJ_SET 2
#define OBJ_ZSET 3
#define OBJ_HASH 4 针对某一种类型的对象redis在不同情况下可能采用不同的数据结构存储结构体robj的的encoding字段表示当前对象底层存储采用的数据结构即对象的编码总共定义了10种encoding常量如下表-1所示表-1 对象编码类型表 encoding常量数据结构可存储对象类型OBJ_ENCODING_RAW简单动态字符串sds字符串OBJ_ENCODING_INT整数字符串OBJ_ENCODING_HT字典dict集合、哈希表、有序集合OBJ_ENCODING_ZIPMAP未使用 OBJ_ENCODING_LINKEDLIST不再使用 OBJ_ENCODING_ZIPLIST压缩列表ziplist哈希表、有序集合BJ_ENCODING_INTSET整数集合intset集合OBJ_ENCODING_SKIPLIST跳跃表skiplist有序集合OBJ_ENCODING_EMBSTR简单动态字符串sds字符串OBJ_ENCODING_QUICKLIST快速链表quicklist列表对象的整个生命周期中编码不是一成不变的比如集合对象。当集合中所有元素都可以用整数表示时底层数据结构采用整数集合执行SADD命令往集合添加元素时redis总会校验待添加元素是否可以解析为整数如果解析失败则会将集合存储结构转换为字典。 if (subject-encoding OBJ_ENCODING_INTSET) {if (isSdsRepresentableAsLongLong(value,llval) C_OK) {subject-ptr intsetAdd(subject-ptr,llval,success);} else {//编码转换setTypeConvert(subject,OBJ_ENCODING_HT);}
} 对象在不同情况下可能采用不同的数据结构存储那对象可能同时采用多种数据结构存储吗根据上面的表格有序集合可能采用压缩列表、跳跃表和字典存储。使用字典存储时根据成员查找分值的时间复杂度为O(1)而对于ZRANGE与ZRANK等命令需要排序才能实现时间复杂度至少为O(NlogN)使用跳跃表存储时ZRANGE与ZRANK等命令的时间复杂度为O(logN)而根据成员查找分值的时间复杂度同样是O(logN)。字典与跳跃表各有优势因此Redis会同时采用字典与跳跃表存储有序集合。这里有读者可能会有疑问同时采用两种数据结构存储不浪费空间吗数据都是通过指针引用的两种存储方式只需要额外存储一些指针即可空间消耗是可以接受的。有序集合存储结构定义如下 typedef struct zset {dict *dict;zskiplist *zsl;
} zset; 观察表-1注意到编码OBJ_ENCODING_RAW和OBJ_ENCODING_EMBSTR都表示的是简单动态字符串那么这两种编码有什么区别吗在回答此问题之前需要先了解结构体robj的定义 #define LRU_BITS 24typedef struct redisObject {unsigned type:4;unsigned encoding:4;unsigned lru:LRU_BITS; //缓存淘汰使用int refcount; //引用计数void *ptr;
} robj; 下面详细分析结构体各字段含义 ptr是void*类型的指针指向实际存储的某一种数据结构但是当robj存储的是数据可以用long类型表示时数据直接存储在ptr字段。可以看出为了创建一个字符串对象必须分配两次内存robj与sds存储空间两次内存分配效率低下且数据分离存储降低了计算机高速缓存的效率。因此提出OBJ_ENCODING_EMBSTR编码的字符串当字符串内容比较短时只分配一次内存robj与sds连续存储以此提升内存分配效率与数据访问效率。OBJ_ENCODING_EMBSTR编码的字符串内存结构如下图-1所示 图-1 EMBSTR编码字符串对象内存结构refcount存储当前对象的引用次数用于实现对象的共享。共享对象时refcount加1删除对象时refcount减1当refcount值为0时释放对象空间。删除对象的代码如下 void decrRefCount(robj *o) {if (o-refcount 1) {switch(o-type) { //根据对象类型释放其指向数据结构空间case OBJ_STRING: freeStringObject(o); break;case OBJ_LIST: freeListObject(o); break;case OBJ_SET: freeSetObject(o); break;…………}zfree(o); //释放对象空间} else {//引用计数减1if (o-refcount ! OBJ_SHARED_REFCOUNT) o-refcount--; }
} lru字段占24比特用于实现缓存淘汰策略可以在配置文件中使用maxmemory-policy指令配置已用内存达到最大内存限制时的缓存淘汰策略。lru根据用户配置缓存淘汰策略存储不同数据常用的策略就是LRU与LFULRU的核心思想是如果数据最近被访问过那么将来被访问的几率也更高此时lru字段存储的是对象访问时间LFU的核心思想是如果数据过去被访问多次那么将来被访问的频率也更高此时lru字段存储的是上次访问时间与访问次数。假如使用GET命令访问数据时会执行下面代码更新对象的lru字段 if (server.maxmemory_policy MAXMEMORY_FLAG_LFU) {updateLFU(val);
} else {val-lru LRU_CLOCK();
} LRU_CLOCK函数用于获取当前时间注意此时间不是实时获取的redis1秒为周期执行系统调用获取精确时间缓存在全局变量server.lruclockLRU_CLOCK函数获取的只是缓存在此变量中的时间。updateLFU函数用于更新对象的上次访问时间与访问次数函数实现如下 void updateLFU(robj *val) {unsigned long counter LFUDecrAndReturn(val);counter LFULogIncr(counter);val-lru (LFUGetTimeInMinutes()8) | counter;
} 可以发现lru的低8比特存储的是对象的访问次数高16比特存储的是对象的上次访问时间以分钟为单位需要特别注意的是函数LFUDecrAndReturn其返回计数值counter对象的访问次数在此值上累加。为什么不直接累加呢假设每次只是简单的对访问次数累加那么越老的数据一般情况下访问次数越大即使该对象可能很长时间已经没有访问。因此访问次数应该有一个随时间衰减的过程函数LFUDecrAndReturn实现了此衰减功能。 1.2 客户端结构体client简介 Redis是典型的客户端服务器结构客户端通过socket与服务端建立网络连接并发送命令请求服务端处理命令请求并回复。Redis使用结构体client存储客户端连接的所有信息包括但不限于客户端的名称、客户端连接的套接字描述符、客户端当前选择的数据库ID、客户端的输入缓冲区与输出缓冲区等。结构体client字段较多此处只介绍命令处理主流程所需的关键字段。 typedef struct client {uint64_t id; int fd; redisDb *db; robj *name;time_t lastinteractionsds querybuf; int argc; robj **argv;struct redisCommand *cmd; list *reply; unsigned long long reply_bytes;size_t sentlen; char buf[PROTO_REPLY_CHUNK_BYTES];int bufpos;} client; 各字段含义如下 1) id客户端唯一ID通过全局对象server的next_client_id字段实现2) fd客户端socket的文件描述符3) db客户端使用select命令选择的数据库对象其结构体定义如下typedef struct redisDb {int id; long long avg_ttl;dict *dict; dict *expires; dict *blocking_keys; dict *ready_keys; dict *watched_keys;
} redisDb; 其中id为数据库序号默认情况下Redis有16个数据库id序号为0~15dict存储数据库所有键值对expires存储键的过期时间avg_ttl存储数据库对象的平均TTL用于统计使用命令BLPOP阻塞获取列表元素时如果链表为空会阻塞客户端同时将此列表键记录在blocking_keys当使用命令PUSH向列表添加元素时会从字典blocking_keys中查找该列表键如果找到说明有客户端正阻塞等待获取此列表键于是将此列表键记录到字典ready_keys以便后续响应正在阻塞的客户端Redis支持事务命令用于MULTI开启事务命令EXEC用于执行事务但是开启事务到执行事务期间如何保证关心的数据不会被修改呢Redis采用乐观锁实现。开启事务的同时可以使用WATCH key命令监控关心的数据键而watched_keys字典存储的就是被WATCH命令监控的所有数据键其中key-value分别为数据键与客户端对象。当Redis服务器接收到写命令时会从字典watched_keys中查找该数据键如果找到说明有客户端正在监控此数据键于是会标记客户端对象为dirty待Redis服务器收到客户端EXEC命令时如果客户端带有dirty标记则会拒绝执行事务。 4) name客户端名称可以使用命令CLIENT SETNAME设置5) lastinteraction客户端上次与服务器交互的时间以此实现客户端的超时处理6) querybuf输入缓冲区recv函数接收的客户端命令请求会暂时缓存在此缓冲区7) argc输入缓冲区的命令请求是按照Redis协议格式编码字符串需要解析出命令请求的所有参数参数个数存储在argc字段参数内容被解析为robj对象存储在argv数组8) cmd待执行的客户端命令解析命令请求后会根据命令名称查找该命令对应的命令对象存储在客户端cmd字段可以看到其类型为struct redisCommand9) reply输出链表链表节点的类型是robj存储待返回给客户端的命令回复数据reply_bytes表示已返回给客户端的字节数10) sentlen当输出数据缓存在reply字段时表示已返回给客户端的对象数目当输出数据缓存在buf字段时表示已返回给客户端的字节数目看到这里读者可能会有疑问为什么同时需要reply和buf的存在呢其实二者只是用于返回不同的数据类型而已详情参见3.3节11) buf输出缓冲区存储待返回给客户端的命令回复数据bufpos表示输出缓冲区中数据的最大字节位置显然sentlen~bufpos区间的数据都是需要返回给客户端的。1.3 服务端结构体redisServer简介 结构体redisServer存储Redis服务器的所有信息包括但不限于数据库、配置参数、命令表、监听端口与地址、客户端列表、若干统计信息、RDB与AOF持久化相关信息、主从复制相关信息、集群相关信息等。结构体redisServer的段非常多这里只对部分字段做简要说明以便读者对于服务端有个粗略了解至于其他字段在讲解各知识点时会做说明。 struct redisServer {char *configfile;int hz;int dbnum;redisDb *db;dict *commands;aeEventLoop *el;int port; char *bindaddr[CONFIG_BINDADDR_MAX];int bindaddr_count;int ipfd[CONFIG_BINDADDR_MAX]; int ipfd_count;list *clients; int maxidletime;
} 各字段含义如下 1) configfile配置文件绝对路径2) hzserverCron函数的执行频率默认为10可通过参数hz配置最小值1最大值500。Redis服务器有很多任务需要定时执行比如说定时清除过期键定时处理超时客户端链接等直接使用系统定时器开销较大函数serverCron就用于执行这些定时任务详情参见1.4.2节。当serverCron函数的执行频率确定时通过函数的执行次数就可以判断是否需要执行某个定时任务宏定义run_with_period就实现了此功能其中server.cronloops字段就表示serverCron函数已经执行的次数#define run_with_period(_ms_) if ((_ms_ 1000/server.hz) || !(server.cronloops%((_ms_)/(1000/server.hz))) 当然由于hz是用户配置的其并不能代表真实的serverCron函数执行频率。 3) dbnum数据库的数目可通过参数databases配置默认164) db数据库数组数组的每个元素都是redisDb类型5) commands命令字典Redis支持的所有命令都存储在这个字典中key为命令名称vaue为struct redisCommand对象6) elRedis是典型的事件驱动程序el即代表着Redis的事件循环7) port服务器监听端口号可通过参数port配置默认端口号63798) bindaddr绑定的所有IP地址可以通过参数bind配置多个例如bind 192.168.1.100 10.0.0.1bindaddr_count为用户配置的IP地址数目CONFIG_BINDADDR_MAX常量为16即绑定16个IP地址Redis默认会绑定到当前机器所有可用的Ip地址9) ipfd针对bindaddr字段的所有IP地址创建的socket文件描述符ipfd_count为创建的socket文件描述符数目10) clients当前连接到Redis服务器的所有客户端11) maxidletime最大空闲时间可通过参数timeout配置结合client对象的lastinteraction字段当客户端超过maxidletime没有与服务器交互时会认为客户端超时并释放该客户端连接1.4 命令结构体redisCommand简介 Redis支持的所有命令初始都存储在全局变量redisCommandTable类型为struct redisCommand[ ]定义及初始化如下 struct redisCommand redisCommandTable[] {{get,getCommand,2,rF,0,NULL,1,1,1,0,0},{set,setCommand,-3,wm,0,NULL,1,1,1,0,0},…………
} 结构体redisCommand相对简单主要定义了命令的名称、命令处理函数以及命令标志等 struct redisCommand {char *name;redisCommandProc *proc;int arity;char *sflags; int flags; long long microseconds, calls;
}; 各字段含义如下 1) name命令名称2) proc命令处理函数3) arity命令参数数目用于校验命令请求格式是否正确当arity小于0时表示命令参数数目大于等于arity当arity大于0时表示命令参数数目必须为arity注意命令请求中命令的名称本身也是一个参数如GET命令的参数数目为2命令请求格式为“GET key”4) sflags命令标志例如标识命令时读命令还是写命令详情参见表-2注意到sflags的类型为字符串此处只是为了良好的可读性5) flags命令的二进制标志服务器启动时解析sflags字段生成6) calls从服务器启动至今命令执行的次数用于统计7) microseconds从服务器启动至今命令总的执行时间microseconds/calls即可计算出该命令的平均处理时间用于统计表-2 命令标志类型 字符标识二进制标识含义相关命令wCMD_WRITE写命令set、del、incr、lpushrCMD_READONLY读命令get、exists、llenmCMD_DENYOOM内存不足时拒绝执行此类命令set、append、lpushaCMD_ADMIN管理命令save、shutdown、slaveofpCMD_PUBSUB发布订阅相关命令subscribe、unsubscribesCMD_NOSCRIPT命令不可以在lua脚本使用auth、save、brpopRCMD_RANDOM随机命令即使命令请求参数完全相同返回结果也可能不容srandmember、scan、timeSCMD_SORT_FOR_SCRIPT当在lua脚本使用此类命令时需要对输出结果做排序sinter、sunion、sdifflCMD_LOADING服务器启动载入过程中只能执行此类命令select、auth、infotCMD_STALE当从服务器与主服务器断开链接且从服务器配置slave-serve-stale-data no时从服务器只能执行此类命令auth、shutdown、infoMCMD_SKIP_MONITOR此类命令不会传播给监视器execkCMD_ASKING restore-askingFCMD_FAST命令执行时间超过阈值时会记录延迟事件此标志用于区分延迟事件类型F表示fast-commandget、setnx、strlen、exists当服务器接收到一条命令请求时需要从命令表中查找命令而redisCommandTable命令表是一个数组意味着查询命令的时间复杂度为O(N)效率低下。因此Redis在服务器初始化时会将redisCommandTable转换为一个字典存储在redisServer对象的commands字段key为命令名称value为命令redisCommand对象。populateCommandTable函数实现了命令表从数组到字典的转化同时解析sflags生成flags void populateCommandTable(void) {int numcommands sizeof(redisCommandTable)/sizeof(structredisCommand);for (j 0; j numcommands; j) {struct redisCommand *c redisCommandTablej;char *f c-sflags;while(*f ! \0) {switch(*f) {case w: c-flags | CMD_WRITE; break;case r: c-flags | CMD_READONLY; break;}f;}retval1 dictAdd(server.commands, sdsnew(c-name), c);}
} 同时对于经常使用的命令Redis甚至会在服务器初始化的时候将命令缓存在redisServer对象这样使用的时候就不需要每次都从commands字典中查找了 struct redisServer {struct redisCommand *delCommand,*multiCommand,*lpushCommand,*lpopCommand,*rpopCommand, *sremCommand, *execCommand,*expireCommand,*pexpireCommand;
} 1.5 事件处理 Redis服务器是典型的事件驱动程序而事件又分为文件事件socket的可读可写事件与时间事件定时任务两大类。无论是文件事件还是时间事件都封装在结构体aeEventLoop typedef struct aeEventLoop {int stop;aeFileEvent *events; aeFiredEvent *fired; aeTimeEvent *timeEventHead;aeBeforeSleepProc *beforesleep;aeBeforeSleepProc *aftersleep;
} aeEventLoop; stop标识事件循环是否结束events为文件事件数组存储已经注册的文件事件fired存储被触发的文件事件Redis有多个定时任务因此理论上应该有多个时间事件节点多个时间事件形成链表timeEventHead即为时间事件链表头结点Redis服务器需要阻塞等待文件事件的发生进程阻塞之前会调用beforesleep函数进程因为某种原因被唤醒之后会调用aftersleep函数。事件驱动程序通常存在while/for循环循环等待事件发生并处理Redis也不例外其事件循环如下 while (!eventLoop-stop) {if (eventLoop-beforesleep ! NULL)eventLoop-beforesleep(eventLoop);aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
} 函数aeProcessEvents为事件处理主函数其第二个参数是一个标志位AE_ALL_EVENTS表示函数需要处理文件事件与时间事件AE_CALL_AFTER_SLEEP表示阻塞等待文件事件之后需要执行aftersleep函数。 1.5.1 文件事件 Redis客户端通过TCP socket与服务端交互文件事件指的就是socket的可读可写事件。 socket读写操作有阻塞与非阻塞之分采用阻塞模式时一个进程只能处理一条网络连接的读写事件为了同时处理多条网络连接通常会采用多线程或者多进程效率低下非阻塞模式下可以使用目前比较成熟的IO多路复用模型select/epoll/kqueue等视不同操作系统而定。这里只对epoll作简要介绍。epoll是linux内核为处理大量并发网络连接而提出的解决方案能显著提升系统CPU利用率。epoll使用非常简单总共只有三个APIepoll_create函数创建一个epoll专用的文件描述符用于后续epoll相关API调用epoll_ctl函数向epoll注册、修改或删除需要监控的事件epoll_wait函数会阻塞进程直到监控的某个网络连接有事件发生。 int epoll_create(int size) 输入参数size通知内核程序期望注册的网络连接数目内核以此判断初始分配空间大小注意在linux2.6.8版本以后内核动态分配空间此参数会被忽略。返回参数为epoll专用的文件描述符不再使用时应该及时关闭此文件描述符。 int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event) 函数执行成功时返回0否则返回-1错误码设置在变量errno输入参数含义如下 1) epfd函数epoll_create返回的epoll文件描述符2) op需要进行的操作EPOLL_CTL_ADD表示注册事件EPOLL_CTL_MOD表示修改网络连接事件EPOLL_CTL_DEL表示删除事件3) fd网络连接的socket文件描述符4) event需要监控的事件或者已触发的事件结构体epoll_event定义如下struct epoll_event {__uint32_t events; epoll_data_t data;
};
typedef union epoll_data {void *ptr;int fd;__uint32_t u32;__uint64_t u64;
} epoll_data_t; 其中events表示需要监控的事件类型或已触发的事件类型比较常用的是EPOLLIN文件描述符可读事件EPOLLOUT文件描述符可写事件data保存与文件描述符关联的数据。 int epoll_wait(int epfd,struct epoll_event * events,int maxevents,int timeout) 函数执行成功时返回0否则返回-1错误码设置在变量errno输入参数含义如下 1) epfd函数epoll_create返回的epoll文件描述符2) epoll_event作为输出参数使用用于回传已触发的事件数组3) maxevents每次能处理的最大事件数目4) timeoutepoll_wait函数阻塞超时时间如果超过timeout时间还没有事件发生函数不再阻塞直接返回当timeout等于0时函数立即返回timeout等于-1时函数会一直阻塞直到有事件发生。Redis并没有直接使用epoll提供的的API而是同时支持四种IO多路复用模型并将每种模型的API进一步统一封装由文件ae_evport.c、ae_epoll.c、ae_kqueue.c和ae_select.c实现。 static int aeApiCreate(aeEventLoop *eventLoop);
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask);
static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int delmask)
static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp); 以epoll为例aeApiCreate函数是对epoll_create的封装aeApiAddEvent函数用于添加事件是对epoll_ctl的封装aeApiDelEvent函数用于删除事件是对epoll_ctl的封装aeApiPoll是对epoll_wait的封装。四个函数输入参数含义如下 1) eventLoop事件循环与文件事件相关最主要有三个字段apidata指向IO多路复用模型对象注意四种IO多路复用模型对象的类型不同因此此字段是void*类型events存储需要监控的事件数组以socket文件描述符作为数组索引存取元素fired存储已出发的事件数组。以epoll模型为例apidata字段指向的IO多路复用模型对象定义如下 typedef struct aeApiState {int epfd;struct epoll_event *events;
} aeApiState; 其中epfd函数epoll_create返回的epoll文件描述符events存储epoll_wait函数返回时已触发的事件数组。 2) fd操作的socket文件描述符3) mask或delmask添加或者删除的事件类型AE_NONE表示没有任何事件AE_READABLE表示可读事件AE_WRITABLE表示可写事件4) tvp阻塞等待文件事件的超时时间这里只对等待事件函数aeApiPoll实现作简要介绍 static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {aeApiState *state eventLoop-apidata;//阻塞等待事件的发生retval epoll_wait(state-epfd,state-events,eventLoop-setsize,tvp ? (tvp-tv_sec*1000 tvp-tv_usec/1000) : -1);if (retval 0) {int j;numevents retval;for (j 0; j numevents; j) {int mask 0;struct epoll_event *e state-eventsj;//转换事件类型为Redis定义的if (e-events EPOLLIN) mask | AE_READABLE;if (e-events EPOLLOUT) mask | AE_WRITABLE;//记录已发生事件到fired数组eventLoop-fired[j].fd e-data.fd;eventLoop-fired[j].mask mask;}}return numevents;
} 函数首先需要通过eventLoop-apidata字段获取到epoll模型对应的aeApiState结构体对象才能调用epoll_wait函数等待事件的发生而epoll_wait函数将已触发的事件存储到aeApiState对象的events字段Redis再次遍历所有已触发事件将其封装在eventLoop-fired数组数组元素类型为结构体aeFiredEvent只有两个字段fd表示发生事件的socket文件描述符mask表示发生的事件类型如AE_READABLE可读事件和AE_WRITABLE可写事件。上面简单介绍了epoll的使用以及Redis对epoll等IO多路复用模型的封装下面我们回到本小节的主题文件事件。结构体aeEventLoop有一个关键字段events类型为aeFileEvent数组存储所有需要监控的文件事件。文件事件结构体定义如下 typedef struct aeFileEvent {int mask; aeFileProc *rfileProc;aeFileProc *wfileProc;void *clientData;
} aeFileEvent; 其中mask存储监控的文件事件类型如AE_READABLE可读事件和AE_WRITABLE可写事件rfileProc为函数指针指向读事件处理函数wfileProc同样为函数指针指向写事件处理函数clientData指向对应的客户端对象。调用aeApiAddEvent函数添加事件之前之前首先需要调用aeCreateFileEvent函数创建对应的文件事件并存储在aeEventLoop结构体的events字段aeCreateFileEvent函数简单实现如下 int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,aeFileProc *proc, void *clientData){aeFileEvent *fe eventLoop-events[fd];if (aeApiAddEvent(eventLoop, fd, mask) -1)return AE_ERR;fe-mask | mask;if (mask AE_READABLE) fe-rfileProc proc;if (mask AE_WRITABLE) fe-wfileProc proc;fe-clientData clientData;return AE_OK;
} Redis服务器启动时需要创建socket并监听等待客户端连接客户端与服务器建立socket连接之后服务器会等待客户端的命令请求服务器处理完成客户端的命令请求之后命令回复会暂时缓存在client结构体的buf缓冲区待客户端文件描述符的可写事件发生时才会真正往客户端发送命令回复。这些都需要创建对应的文件事件 aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL);aeCreateFileEvent(server.el,fd,AE_READABLE,readQueryFromClient, c);aeCreateFileEvent(server.el, c-fd, ae_flags, sendReplyToClient, c); 可以发现接收客户端连接的处理函数为acceptTcpHandler此时还没有创建对应的客户端对象因此函数aeCreateFileEvent第四个参数为NULL接收客户端命令请求的处理函数为readQueryFromClient向发送命令回复的处理函数为sendReplyToClient。 最后思考一个问题 aeApiPoll函数的第二个参数是时间结构体timeval存储调用epoll_wait时传入的超时时间那么这个函数怎么计算出来的呢我们之前提过Redis除了要处理各种文件事件外还需要处理很多定时任务时间事件那么当Redis由于执行epoll_wait而阻塞时恰巧定时任务到期而需要处理怎么办要回答这个问题需要分析下Redis事件循环的执行函数aeProcessEvents函数在调用aeApiPoll之前会遍历Redis的时间事件链表查找最早会发生的时间事件以此作为aeApiPoll需要传入的超时时间。 int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{shortest aeSearchNearestTimer(eventLoop);long long ms shortest-when_sec - now_sec)*1000 shortest-when_ms - now_ms;//阻塞等待文件事件发生numevents aeApiPoll(eventLoop, tvp);for (j 0; j numevents; j) {aeFileEvent *fe eventLoop-events[eventLoop-fired[j].fd];//处理文件事件即根据类型执行rfileProc或wfileProc}//处理时间事件processed processTimeEvents(eventLoop);
} 1.5.2 时间事件 1.5.1节介绍了Redis文件事件已经知道事件循环执行函数aeProcessEvents的主要逻辑1查找最早会发生的时间事件计算超时时间2阻塞等待文件事件的产生3处理文件事件4处理时间事件。时间事件的执行函数为processTimeEvents。Redis服务器内部有很多定时任务需要执行比如说定时清除超时客户端连接定时删除过期键等定时任务被封装为时间事件结构体aeTimeEvent存储多个时间事件形成链表存储在aeEventLoop结构体的timeEventHead字段其指向链表首节点。时间事件aeTimeEvent定义如下 typedef struct aeTimeEvent {long long id; long when_sec; long when_ms; aeTimeProc *timeProc;aeEventFinalizerProc *finalizerProc;void *clientData;struct aeTimeEvent *next;
} aeTimeEvent; 各字段含义如下 1) id时间事件唯一ID通过字段eventLoop-timeEventNextId实现2) when_sec与when_ms时间事件触发的秒数与毫秒数3) timeProc函数指针指向时间事件处理函数4) finalizerProc函数指针删除时间事件节点之前会调用此函数5) clientData指向对应的客户端对象6) next指向下一个时间事件节点。时间事件执行函数processTimeEvents的处理逻辑比较简单只是遍历时间事件链表判断当前时间事件是否已经到期如果到期则执行时间事件处理函数timeProc static int processTimeEvents(aeEventLoop *eventLoop) {te eventLoop-timeEventHead;while(te) {aeGetTime(now_sec, now_ms);if (now_sec te-when_sec ||(now_sec te-when_sec now_ms te-when_ms)) {//处理时间事件retval te-timeProc(eventLoop, id, te-clientData);//重新设置时间事件到期时间if (retval ! AE_NOMORE) {aeAddMillisecondsToNow(retval,te-when_sec,te-when_ms);}}te te-next;}
} 注意时间事件处理函数timeProc返回值retval其表示此时间事件下次应该被触发的时间单位毫秒且是一个相对时间即从当前时间算起retval毫秒后此时间事件会被触发。其实Redis只有一个时间事件节点看到这里读者可能会有疑惑服务器内部不是有很多定时任务吗为什么只有一个时间事件呢回答此问题之前我们需要先分析这个唯一的时间事件节点。Redis创建时间事件节点的函数为aeCreateTimeEvent内部实现非常简单只是创建时间事件节点并添加到时间事件链表。aeCreateTimeEvent函数定义如下 long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,aeTimeProc *proc, void *clientData,aeEventFinalizerProc *finalizerProc); 其中输入参数eventLoop指向事件循环结构体milliseconds表示此时间事件触发时间单位毫秒注意这是一个相对时间即从当前时间算起milliseconds毫秒后此时间事件会被触发proc指向时间事件的处理函数clientData指向对应的结构体对象finalizerProc同样是函数指针删除时间事件节点之前会调用此函数。读者可以在代码目录全局搜索aeCreateTimeEvent会发现确实只创建了一个时间事件节点 aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL); 该时间事件在1毫秒后会被触发处理函数为serverCron参数clientData与finalizerProc都为NULL。而函数serverCron实现了Redis服务器所有定时任务的周期执行。 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {run_with_period(100) {//100毫秒周期执行}run_with_period(5000) {//5000毫秒周期执行}//清除超时客户端链接clientsCron();//处理数据库databasesCron();server.cronloops;return 1000/server.hz;
} 变量server.cronloops用于记录serverCron函数的执行次数变量server.hz表示serverCron函数的执行频率用户可配置最小为1最大为500默认为10。假设server.hz取默认值10函数返回1000/server.hz会更新当前时间事件的触发时间为100毫秒后即serverCron的执行周期为100毫秒。run_with_period宏定义实现了定时任务按照指定时间周期执行其会被替换为一个if条件判断条件为真才会执行定时任务定义如下 #define run_with_period(_ms_) if ((_ms_ 1000/server.hz) || !(server.cronloops%((_ms_)/(1000/server.hz)))) 另外我们可以看到serverCron函数会无条件执行某些定时任务比如清除超时客户端连接以及处理数据库清除数据库过期键等。需要特别注意一点serverCron函数的执行时间不能过长否则会导致服务器不能及时响应客户端的命令请求。以过期键删除为例分析下Redis是如何保证serverCron函数的执行时间。过期键删除由函数activeExpireCycle实现由函数databasesCron调用其函数是实现如下 #define ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC 25void activeExpireCycle(int type) {timelimit
1000000*ACTIVE_EXPIRE_CYCLE_SLOW_TIME_PERC/server.hz/100;timelimit_exit 0;for (j 0; j dbs_per_call timelimit_exit 0; j) {do {//查找过期键并删除if ((iteration 0xf) 0) {elapsed ustime()-start;if (elapsed timelimit) {timelimit_exit 1;break;}}}while (expired ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP/4)}
} 函数activeExpireCycle最多遍历dbs_per_call个数据库并记录每个数据库删除的过期键数目当删除过期键数目大于门限时认为此数据库过期键较多需要再次处理。考虑到极端情况当数据库键数目非常多且基本都过期时do-while循环会一直执行下去。因此我们添加timelimit时间限制每执行16次do-while循环检测函数activeExpireCycle执行时间是否超过timelimit如果超过则强制结束循环。初看timelimit的计算方式可能会比较疑惑其计算结果使得函数activeExpireCycle的总执行时间占CPU时间的25%。仍然假设server.hz取默认值10即每秒钟函数activeExpireCycle执行10次那么每秒钟函数activeExpireCycle的总执行时间为100000025/100每次函数activeExpireCycle的执行时间为100000025/100/10单位微妙。 2 sever启动过程 上一节我们讲述了客户端服务端事件处理等基础知识下面开始学习Redis服务器的启动过程这里主要分为server初始化监听端口以及等待命令三个小节。 2.1 server初始化 服务器初始化主流程可以简要分为7个步骤1初始化配置包括用户可配置的参数以及命令表的初始化2加载并解析配置文件3初始化服务端内部变量其中就包括数据库4创建事件循环eventLoop5创建socket并启动监听6创建文件事件与时间事件7开启事件循环。下面详细介绍步骤1~4至于步骤5~7将会在2.2小节介绍。 图-2 server初始化流程步骤1初始化配置由函数initServerConfig实现其实就是给配置参数赋初始值 void initServerConfig(void) {//serverCron函数执行频率默认10server.hz CONFIG_DEFAULT_HZ; //监听端口默认6379server.port CONFIG_DEFAULT_SERVER_PORT; //最大客户端数目默认10000server.maxclients CONFIG_DEFAULT_MAX_CLIENTS; //客户端超时时间默认0即永不超时server.maxidletime CONFIG_DEFAULT_CLIENT_TIMEOUT;//数据库数目默认16server.dbnum CONFIG_DEFAULT_DBNUM;//初始化命令表1.4小节已经讲过这里不再详述populateCommandTable();…………
} 步骤2加载并解析配置文件入口函数为loadServerConfig函数声明如下 void loadServerConfig(char *filename, char *options) 输入参数filename表示配置文件全路径名称options表示命令行输入的配置参数例如我们通常以以下命令启动Redis服务器 /home/user/redis/redis-server /home/user/redis/redis.conf -p 4000 使用GDB启动redis-server打印函数 loadServerConfig输入参数如下 (gdb) p filename
$1 0x778880 /home/user/redis/redis.conf
(gdb) p options
$2 0x7ffff1a21d33 \-p\ \4000\ Redis的配置文件语法相对简单每一行是一条配置格式如“配置 参数1 [参数2] [……]”加载配置文件只需要一行一行将文件内容读取到内存中即可GDB打印加载到内存中的配置如下 (gdb) p config
bind 127.0.0.1\n\nprotected-mode yes\n\nport 6379\ntcp-backlog 511\n\ntcp-keepalive 300\n\n……… 加载完成后会调用loadServerConfigFromString函数解析配置输入参数config即配置字符串实现如下 void loadServerConfigFromString(char *config) {//分割配置字符串多行totlines记录行数lines sdssplitlen(config,strlen(config),\n,1,totlines);for (i 0; i totlines; i) {//跳过注释行与空行if (lines[i][0] # || lines[i][0] \0) continue;argv sdssplitargs(lines[i],argc); //解析配置参数//赋值if (!strcasecmp(argv[0],timeout) argc 2) {server.maxidletime atoi(argv[1]);}else if (!strcasecmp(argv[0],port) argc 2) {server.port atoi(argv[1]);}//其他配置}
} 函数首先将输入配置字符串以“n”为分隔符划分为多行totlines记录总行数lines数组存储分割后的配置数组元素类型为字符串SDSfor循环遍历所有配置行解析配置参数并根据参数内容设置结构体server各字段。注意Redis配置文件中行开始“#”字符标识本行内容为注释解析时需要跳过。步骤3初始化服务器内部变量比如客户端链表数据库全局变量共享对象等入口函数为initServer函数逻辑相对简单这里只做简要说明 void initServer(void) {server.clients listCreate(); //初始化客户端链表//创建数据库字典server.db zmalloc(sizeof(redisDb)*server.dbnum);for (j 0; j server.dbnum; j) {server.db[j].dict dictCreate(dbDictType,NULL);…………}
} 注意数据库字典的dictType指向的是结构体dbDictType其中定义了数据库字典键的哈希函数键比较函数以及键与值的析构函数定义如下 dictType dbDictType {dictSdsHash, NULL, NULL, dictSdsKeyCompare,dictSdsDestructor,dictObjectDestructor
}; 数据库的键都是SDS类型键哈希函数为dictSdsHash键比较函数为dictSdsKeyCompare键析构函数为dictSdsDestructor数据库的值是robj对象值析构函数为dictObjectDestructor键和值的内容赋值函数都为NULL。1.1节提到对象robj的refcount字段存储当前对象的引用次数意味着对象是可以共享的。要注意的是只有当对象robj存储的是0~10000以内的整数对象robj才会被共享且这些共享整数对象的引用计数初始化为INT_MAX保证不会被释放。执行命令时Redis会返回一些字符串回复这些字符串对象同样在服务器初始化时创建且永远不会尝试释放这类对象。所有共享对象都存储在全局结构体变量shared。 void createSharedObjects(void) {//创建命令回复字符串对象shared.ok createObject(OBJ_STRING,sdsnew(OK\r\n));shared.err createObject(OBJ_STRING,sdsnew(-ERR\r\n));//创建0~10000整数对象for (j 0; j OBJ_SHARED_INTEGERS; j) {shared.integers[j] makeObjectShared(createObject(OBJ_STRING,(void*)(long)j));shared.integers[j]-encoding OBJ_ENCODING_INT;}
} 步骤4创建事件循环eventLoop即分配结构体所需内存并初始化结构体各字段epoll就是在此时创建的 aeEventLoop *aeCreateEventLoop(int setsize) {if ((eventLoop zmalloc(sizeof(*eventLoop))) NULL) goto err;eventLoop-events zmalloc(sizeof(aeFileEvent)*setsize);eventLoop-fired zmalloc(sizeof(aeFiredEvent)*setsize);if (aeApiCreate(eventLoop) -1) goto err;
} 输入参数setsize理论上等于用户配置的虽大客户端数目即可但是为了确保安全这里设置setsize等于最大客户端数目加128。函数aeApiCreate内部调用epoll_create创建epoll并初始化结构体eventLoop的字段apidata。 2.2 启动监听 上节介绍了服务器初始化的前面4个步骤初始化配置加载并解析配置文件初始化服务端内部遍历包括数据库全局共享变量等创建时间循环eventLoop。完成这些操作之后Redis将创建socket并启动监听同时创建对应的文件事件与时间事件并开始事件循环。下面将详细介绍步骤5~7。步骤5创建socket并启动监听用户可通过指令port配置socket绑定端口号指令bind配置socket绑定IP地址注意指令bind可配置多个IP地址中间用空格隔开创建socket时只需要循环所有IP地址即可。 int listenToPort(int port, int *fds, int *count) {for (j 0; j server.bindaddr_count || j 0; j) {//创建socket并启动监听文件描述符存储在fds数组作为返回参数fds[*count] anetTcpServer(server.neterr,port,server.bindaddr[j],server.tcp_backlog);//设置socket非阻塞anetNonBlock(NULL,fds[*count]);(*count);}
} 输入参数port表示用户配置的端口号server结构体的bindaddr_count字段存储用户配置的IP地址数目bindaddr字段存储用户配置的所有IP地址。函数anetTcpServer实现了socket的创建绑定以及监听流程这里不做过多详述。参数fds与count可用作输出参数fds数组存储创建的所有socket文件描述符count存储socket数目。注意到所有创建的socket都会设置为非阻塞模式原因在于Redis使用了IO多路复用模式其要求socket读写必须是非阻塞的函数anetNonBlock通过系统调用fcntl设置socket非阻塞模式。步骤6创建文件事件与时间事件步骤5中已经完成了socket的创建与监听1.5.1节提到socket的读写事件被抽象为文件事件因为对于监听的socket还需要创建对应的文件事件。 for (j 0; j server.ipfd_count; j) {if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL) AE_ERR){}
} server结构体的ipfd_count字段存储创建的监听socket数目ipfd数组存储创建的所有监听socket文件描述符需要循环所有的监听socket为其创建对应的文件事件。可以看到监听事件的处理函数为acceptTcpHandler实现了socket连接请求的accept以及客户端对象的创建。1.5.2小节提到定时任务被抽象为时间事件且Redis只创建了一个时间事件在服务端初始化时创建。此时间事件的处理函数为serverCron初次创建时1毫秒后备触发。 if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) AE_ERR) {exit(1);
} 步骤7开启事件循环前面6个步骤已经完成了服务端的初始化工作并在指定IP地址、端口监听客户端连接同时创建了文件事件与时间事件此时只需要开启事件循环等待事件发生即可。 void aeMain(aeEventLoop *eventLoop) {eventLoop-stop 0;//开始事件循环while (!eventLoop-stop) {if (eventLoop-beforesleep ! NULL)eventLoop-beforesleep(eventLoop);//事件处理主函数aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);}
} 事件处理主函数aeProcessEvents已经详细介绍过这里需要重点关注函数beforesleep其在每次事件循环开始即Redis阻塞等待文件事件之前执行。函数beforesleep会执行一些不是很费时的操作集群相关操作过期键删除操作这里可称为快速过期键删除向客户端返回命令回复等。这里简要介绍下快速过期键删除操作。 void beforeSleep(struct aeEventLoop *eventLoop) {if (server.active_expire_enabled server.masterhost NULL)activeExpireCycle(ACTIVE_EXPIRE_CYCLE_FAST);
} Redis过期键删除有两种策略1访问数据库键时校验该键是否过期如果过期则删除2周期性删除过期键beforeSleep函数与serverCron函数都会执行。server结构体的active_expire_enabled字段表示是否开启周期性删除过期键策略用户可通过set-active-expire指令配置masterhost字段存储当前Redis服务器的master服务器的域名如果为NULL说明当前服务器不是某个Redis服务器的slaver。注意到这里依然是调用函数activeExpireCycle执行过期键删除只是参数传递的是ACTIVE_EXPIRE_CYCLE_FAST表示快速过期键删除。回顾下1.5.2节讲述函数activeExpireCycle的实现函数计算出timelimit即函数最大执行时间循环删除过期键时会校验函数执行时间是否超过此限制超过则结束循环。显然快速过期键删除时只需要缩短timelimit即可计算策略如下 void activeExpireCycle(int type) {static int timelimit_exit 0; static long long last_fast_cycle 0if (type ACTIVE_EXPIRE_CYCLE_FAST) {//上次activeExpireCycle函数是否已经执行完毕if (!timelimit_exit) return;//当前时间距离上次执行快速过期键删除是否已经超过2000微妙if (start last_fast_cycle 1000*2) return;last_fast_cycle start;}//快速过期键删除时函数执行时间不超过1000微妙if (type ACTIVE_EXPIRE_CYCLE_FAST)timelimit 1000;
} 执行快速过期键删除有很多限制当函数activeExpireCycle正在执行时直接返回当上次执行快速过期键删除的时间距离当前时间小于2000微妙时直接返回。思考下为什么可以通过变量timelimit_exit判断函数activeExpireCycle是否正在执行呢注意到变量timelimit_exit声明为static即函数执行完毕不会释放变量空间。那么可以在函数activeExpireCycle入口赋值timelimit_exit为0返回之前赋值timelimit_exit为1由此便可通过变量timelimit_exit判断函数activeExpireCycle是否正在执行。变量last_fast_cycle声明为static也是同样的原因。同时可以看到当执行快速过期键删除时设置函数activeExpireCycle的最大执行时间为1000微妙。函数aeProcessEvents为事件处理主函数首先查找最近发生的时间事件调用epoll_wait阻塞等待文件事件的发生并设置超时事件待epoll_wait返回时处理触发的文件事件最后处理时间事件。步骤6中已经创建了文件事件为监听socket的读事件事件处理函数为acceptTcpHandler即当客户端发起socket连接请求时服务端会执行函数acceptTcpHandler处理。acceptTcpHandler函数主要做了两件事1accept客户端的连接请求2创建客户端对象3创建文件事件。步骤2与步骤3由函数createClient实现输入参数fd为accept客户端连接请求后生成的socket文件描述符。 client *createClient(int fd) {client *c zmalloc(sizeof(client));//设置socket为非阻塞模式anetNonBlock(NULL,fd);//设置TCP_NODELAYanetEnableTcpNoDelay(NULL,fd);//如果服务端配置了tcpkeepalive则设置SO_KEEPALIVEif (server.tcpkeepalive)anetKeepAlive(NULL,fd,server.tcpkeepalive);if (aeCreateFileEvent(server.el,fd,AE_READABLE,readQueryFromClient, c) AE_ERR){ }
} 为了使用IO多路复用模式此处同样需要设置socket为非阻塞模式。TCP是基于字节流的可靠传输层协议为了提升网络利用率一般默认都会开启Nagle。当应用层调用write函数发送数据时TCP并不一定会立刻将数据发送出去根据Nagle算法还必须满足一定条件才行。Nagle是这样规定的如果数据包长度大于一定门限时则立即发送如果数据包中含有FIN表示断开TCP链接字段则立即发送如果当前设置了TCP_NODELAY选项则立即发送如果所有条件都不满足默认需要等待200毫秒超时后才会发送。Redis服务器向客户端返回命令回复时希望TCP能立即将该回复发送给客户端因此需要设置TCP_NODELAY。思考下如果不设置会怎么样呢从客户端分析命令请求的响应时间会大大加长。TCP是可靠的传输层协议每次都需要经历三次握手与四次挥手为了提升效率可以设置SO_KEEPALIVE即TCP长连接这样TCP传输层会定时发送心跳包确认该连接的可靠性。应用层也不再需要频繁的创建于释放TCP连接了。server结构体的tcpkeepalive字段表示是否启用TCP长连接用户可通过参数tcp-keepalive配置。接收到客户端连接请求之后服务器需要创建文件事件等待客户端的命令请求可以看到文件事件的处理函数为readQueryFromClient当服务器接收到客户端的命令请求时会执行此此函数。 3 命令处理过程 上一节分析了服务器的启动过程包括配置文件的解析创建socket并启动监听创建文件事件与时间事件并开启事件循环。服务器启动完成后只需要等待客户端连接并发送命令请求即可。本小节主要介绍命令的处理过程可以分为三个阶段解析命令请求命令调用和返回结果给客户端。 3.1 命令解析 TCP是一种基于字节流的传输层通信协议因此接收到的TCP数据不一定是一个完整的数据包其有可能是多个数据包的组合也有可能是某一个数据包的部分这种现象被称为半包与粘包。如图-3所示。 图-3 TCP半包与粘包客户端应用层分别发送三个数据包data3、data2和data1但是TCP传输层在真正发送数据时将data3数据包分割为data3_1与data3_2并且将data1与data2数据合并此时服务器接收到的数据包就不是一个完整的数据包。为了区分一个完整的数据包通常有如下三种方法1数据包长度固定2通过特定的分隔符区分比如HTTP协议就是通过换行符区分的3通过在数据包头部设置长度长度字段区分数据包长度比如FastCGI协议。Redis采用自定义协议格式实现不同命令请求的区分例如当用户在redis-cli客户端键入下面命令 SET redis-key value1 vlaue2 value3 客户端会将该命令请求转换为以下协议格式然后发送给服务器 *5\r\n$3\r\n$9redis-key\r\n$6value1\r\n$6vlaue2\r\n$6value3\r\n 其中换行符rn用于区分命令请求的若干参数“*5”表示该命令请求有5个参数“$3”、“$9”和“$6”等表示该参数字符串长度多个请求参数之间用“rn”分隔开需要注意的是Redis还支持在telnet会话输入命令的方式只是此时没有了请求协议中的“*”来声明参数的数量因此必须使用空格来分割各个参数服务器在接收到数据之后会将空格作为参数分隔符解析命令请求。这种方式的命令请求称为内联命令。Redis服务器接收到的命令请求首先存储在客户端对象的querybuf输入缓冲区然后解析命令请求各个参数并存储在客户端对象的argv参数对象数组和argc参数数目字段。参考2.2小节可以知道解析客户端命令请求的入口函数为readQueryFromClient会读取socket数据存储到客户端对象的输入缓冲区并调用函数processInputBuffer解析命令请求。processInputBuffer函数主要逻辑如图-4所示。 图-4 命令解析流程图下面简要分析通过redis-cli客户端发送的命令请求的解析过程。假设客户端命令请求为“SET redis-key value1”在函数processMultibulkBuffer添加断点GDB打印客户端输入缓冲区内容如下 (gdb) p c-querybuf
$3 (sds) 0x7ffff1b45505 *3\r\n$3\r\nSET\r\n$9\r\nredis-key\r\n$6\r\nvalue1\r\n 解析该命令请求可以分为2个步骤1解析命令请求参数数目2循环解析每个请求参数。下面详细分析每个步骤的源码实现步骤1解析命令请求参数数目querybuf指向命令请求首地址命令请求参数数目的协议格式为“3rn”即首字符必须是“”并且可以使用字符“r”定位到行尾位置解析后的参数数目暂存在客户端对象的multibulklen字段表示等待解析的参数数目变量pos记录已解析命令请求的长度。 //定位到行尾
newline strchr(c-querybuf,\r);//解析命令请求参数数目并存储在客户端对象的multibulklen字段
serverAssertWithInfo(c,NULL,c-querybuf[0] *);
string2ll(c-querybuf1,newline-(c-querybuf1),ll);
c-multibulklen ll;//记录已解析位置偏移量
pos (newline-c-querybuf)2;
//分配请求参数存储空间
c-argv zmalloc(sizeof(robj*)*c-multibulklen); GDB打印主要变量内容如下 (gdb) p c-multibulklen
$9 3
(gdb) p pos
$10 4 步骤2循环解析每个请求参数命令请求各参数的协议格式为“$3\r\nSET\r\n”即首字符必须是“$”。解析当前参数之前需要解析出参数的字符串长度可以使用字符“r”定位到行尾位置注意到解析参数长度时字符串开始位置为querybufpos1字符串参数长度暂存在客户端对象的bulklen字段同时更新已解析字符串长度pos。 //定位到行尾
newline strchr(c-querybufpos,\r);
//解析当前参数字符串长度字符串首字符偏移量为pos
if (c-querybuf[pos] ! $) {return C_ERR;
}
ok string2ll(c-querybufpos1,newline-(c-querybufpos1),ll);
pos newline-(c-querybufpos)2;
c-bulklen ll; GDB打印主要变量内容如下 (gdb) p c-querybufpos
$13 0x7ffff1b4550d SET\r\n$9\r\nredis-key\r\n$6\r\nvalue1\r\n
(gdb) p c-bulklen
$15 3
(gdb) p pos
$16 8 解析出参数字符串长度之后可直接读取该长度的参数内容并创建字符串对象同时需要更新待解析参数multibulklen。 //解析参数
c-argv[c-argc] createStringObject(c-querybufpos,c-bulklen);
pos c-bulklen2;//待解析参数数目减一
c-multibulklen--; 当multibulklen值更新尾0时说明参数解析完成结束循环。读者可以思考下待解析参数数目当前参数长度为什么都需要暂存在客户端结构体使用函数局部变量行不行肯定是不行的原因就在于上面提到的TCP半包与粘包现象服务器可能只接收到部分命令请求例如“3rn$3\r\nSET\r\n$9rnredis”。当函数processMultibulkBuffer执行完毕时同样只会解析部分命令请求“3rn$3\r\nSET\r\n$9rn”此时就需要记录该命令请求待解析的参数数目以及待解析参数的长度而剩余待解析的参数“redis”会继续缓存在客户端的输入缓冲区。 3.2 命令调用 参考图-4解析完成命令请求之后会调用函数processCommand处理该命令请求而处理命令请求之前还有很多校验逻辑比如说客户端是否已经完成认证命令请求参数是否合法等。下面简要列出若干校验规则。校验1如果是quit命令直接返回并关闭客户端 if (!strcasecmp(c-argv[0]-ptr,quit)) {addReply(c,shared.ok);c-flags | CLIENT_CLOSE_AFTER_REPLY;return C_ERR;
} 校验2执行函数lookupCommand查找命令后如果命令不存在返回错误 c-cmd c-lastcmd lookupCommand(c-argv[0]-ptr);
if (!c-cmd) {addReplyErrorFormat(c,unknown command %s,(char*)c-argv[0]-ptr);return C_OK;
} 校验3如果命令参数数目不合法返回错误。命令结构体的arity用于校验参数数目是否合法当arity小于0时表示命令参数数目大于等于arity当arity大于0时表示命令参数数目必须为arity注意命令请求中命令的名称本身也是一个参数。 if ((c-cmd-arity 0 c-cmd-arity ! c-argc) ||(c-argc -c-cmd-arity)) {addReplyErrorFormat(c,wrong number of arguments for %s command,c-cmd-name);return C_OK;
} 校验4如果使用指令“requirepass password”设置了密码且客户端没未认证通过只能执行auth命令auth命令格式为“AUTH password”。 if (server.requirepass !c-authenticated c-cmd-proc ! authCommand){addReply(c,shared.noautherr);return C_OK;
} 校验5如果使用指令“maxmemory bytes”设置了最大内存限制且当前内存使用量超过了该配置门限服务器会拒绝执行带有“m”CMD_DENYOOM标识的命令如SET命令、APPEND命令和LPUSH命令等。命令标识参见1.4小节。 if (server.maxmemory) {int retval freeMemoryIfNeeded();if ((c-cmd-flags CMD_DENYOOM) retval C_ERR) {addReply(c, shared.oomerr);return C_OK;}
} 校验6除了上面的5种校验还有很多校验规则比如集群相关校验持久化相关校验主从复制相关校验发布订阅相关校验以及事务操作等。这些校验规则会在相关章节会作详细介绍。当所有校验规则都通过后才会调用命令处理函数执行命令代码如下 start ustime();
c-cmd-proc(c);
duration ustime()-start;//更新统计信息当前命令执行时间与调用次数
c-lastcmd-microseconds duration;
c-lastcmd-calls;//记录慢查询日志
slowlogPushEntryIfNeeded(c,c-argv,c-argc,duration); 执行命令完成后如果有必要还需要更新统计信息记录慢查询日志AOF持久化该命令请求传播命令请求给所有的从服务器等。持久化与主从复制会在相关章节会作详细介绍这里主要介绍慢查询日志的实现方式。 void slowlogPushEntryIfNeeded(client *c, robj **argv, int argc, long long duration) {//执行时间超过门限记录该命令if (duration server.slowlog_log_slower_than)listAddNodeHead(server.slowlog,slowlogCreateEntry(c,argv,argc,duration));//慢查询日志最多记录条数为slowlog_max_len超过需删除while (listLength(server.slowlog) server.slowlog_max_len)listDelNode(server.slowlog,listLast(server.slowlog));
} 可以使用指令“slowlog-log-slower-than 10000”配置执行时间超过多少毫秒才会记录慢查询日志指令“slowlog-max-len 128”配置慢查询日志最大数目超过会删除最早的日志记录。可以看到慢查询日志记录在服务端结构体的slowlog字段即存储速度非常快并不会影响命令执行效率。用户可通过“SLOWLOG subcommand [argument]”命令查看服务器记录的慢查询日志。 3.3 返回结果 Redis服务器返回结果类型不同协议格式不同而客户端可以根据返回结果的第一个字符判断返回类型。Redis的返回结果可以分为5类 1状态回复第一个字符是“”例如SET命令执行完毕会向客户端返回“OKrn”。addReply(c, ok_reply ? ok_reply : shared.ok); 变量ok_reply通常为NULL则返回的是共享变量shared.ok在服务器启动时就完成了共享变量的初始化。 shared.ok createObject(OBJ_STRING,sdsnew(OK\r\n)); 2错误回复第一个字符是“-”例如当客户端请求命令不存在时会向客户端返回“-ERR unknown command testcmd”。addReplyErrorFormat(c,unknown command %s,(char*)c-argv[0]-ptr); 而函数addReplyErrorFormat内部实现会拼装错误回复字符串。 addReplyString(c,-ERR ,5);
addReplyString(c,s,len);
addReplyString(c,\r\n,2); 3整数回复第一个字符是“:”例如INCR命令执行完毕向客户端返回“:100rn”。addReply(c,shared.colon);
addReply(c,new);
addReply(c,shared.crlf); 其中共享变量shared.colon与shared.crlf同样都是在服务器启动时就完成了初始化。 shared.colon createObject(OBJ_STRING,sdsnew(:));
shared.crlf createObject(OBJ_STRING,sdsnew(\r\n)); 4批量回复第一个字符是“$”例如GET命令查找键向客户端返回结果“$5rnhellorn”其中$5表示返回字符串长度。//计算返回对象obj长度并拼接为字符串“$5\r\n”
addReplyBulkLen(c,obj);
addReply(c,obj);
addReply(c,shared.crlf); 5多条批量回复第一个字符是“”例如LRANGE命令可能会返回多个多个值格式为“3rn$6\r\nvalue1\r\n$6rnvalue2rn$6\r\nvalue3\r\n”与命令请求协议格式相同“*3”表示返回值数目“$6”表示当前返回值字符串长度多个返回值用“rn”分隔开。//拼接返回值数目“*3\r\n”
addReplyMultiBulkLen(c,rangelen);
//循环输出所有返回值
while(rangelen--) {//拼接当前返回值长度“$6\r\n”addReplyLongLongWithPrefix(c,len,$);addReplyString(c,p,len);addReply(c,shared.crlf);
} 可以看到5种类型的返回结果都是调用类似于addReply函数返回的那么是这些方法将返回结果发送给客户端的吗其实不是。回顾1.2小节讲述的客户端结构体client其中有两个关键字段reply和buf分别表示输出链表与输出缓冲区而函数addReply会直接或者间接的调用以下两个函数将返回结果暂时缓存在reply或者buf字段。 //添加字符串都输出缓冲区
int _addReplyToBuffer(client *c, const char *s, size_t len) //添加各种类型的对象到输出链表
void _addReplyObjectToList(client *c, robj *o)
void _addReplySdsToList(client *c, sds s)
void _addReplyStringToList(client *c, const char *s, size_t len) 需要特别注意的是reply和buf字段不可能同时缓存待返回给客户端的数据。从客户端结构体的sentlen字段就能看出当输出数据缓存在reply字段时sentlen表示已返回给客户端的对象数目当输出数据缓存在buf字段时sentlen表示已返回给客户端的字节数目。那么当reply和buf字段同时缓存有输出数据呢只有sentlen字段显然是不够的。从_addReplyToBuffer函数实现同样可以看出该结论。 int _addReplyToBuffer(client *c, const char *s, size_t len) {if (listLength(c-reply) 0) return C_ERR;
} 调用函数_addReplyToBuffer缓存数据到输出缓冲区时如果检测到reply字段有待返回给客户端的数据函数返回错误。而通常缓存数据时都会先尝试缓存到buf输出缓冲区如果失败会再次尝试缓存到reply输出链表。 if (_addReplyToBuffer(c,obj-ptr,sdslen(obj-ptr)) ! C_OK)_addReplyObjectToList(c,obj); 而函数addReply在将待返回给客户端的数据暂时缓存在输出缓冲区或者输出链表的同时会将当前客户端添加到服务端结构体的clients_pending_write链表以便后续能快速查找出哪些客户端有数据需要发送。 listAddNodeHead(server.clients_pending_write,c); 看到这里读者可能会有疑问函数addReply只是将待返回给客户端的数据暂时缓存在输出缓冲区或者输出链表那么什么时候将这些数据发送给客户端呢读者是否还记得在介绍开启事件循环时提到函数beforesleep在每次事件循环阻塞等待文件事件之前执行主要执行一些不是很费时的操作比如过期键删除操作向客户端返回命令回复等。函数beforesleep会遍历clients_pending_write链表中每一个客户端节点并发送输出缓冲区或者输出链表中的数据。 //遍历clients_pending_write链表
listRewind(server.clients_pending_write,li);
while((ln listNext(li))) {client *c listNodeValue(ln);listDelNode(server.clients_pending_write,ln);//向客户端发送数据if (writeToClient(c-fd,c,0) C_ERR) continue;
} 看到这里我想大部分读者可能都会认为返回结果已经发送给客户端命令请求也已经处理完成了。其实不然读者可以思考这么一个问题当返回结果数据量非常大时是无法一次性将所有数据都发送给客户端的即函数writeToClient执行之后客户端输出缓冲区或者输出链表中可能还有部分数据未发送给客户端。这时候怎么办呢很简单只需要添加文件事件监听当前客户端socket文件描述符的可写事件即可。 if (aeCreateFileEvent(server.el, c-fd, AE_WRITABLE,sendReplyToClient, c) AE_ERR){
} 可以看到该文件事件的事件处理函数为sendReplyToClient即当客户端可写时函数sendReplyToClient会发送剩余部分的数据给客户端。至此命令请求才算是真正处理完成了。 4 本文小结 为了更好的理解服务器与客户端的交互本文首先介绍了一些基础结构体如对象结构体robj客户端结构体client服务端结构体redisServer以及命令结构体redisCommand。Redis服务器是典型的事件驱动程序将事件处理分为两大类文件事件与时间事件。文件事件即socket的可读可写事件时间事件即需要周期性执行的一些定时任务。Redis采用比较成熟的IO多路复用模型select/epoll等处理文件事件并对这些IO多路复用模型做了简单封装。Redis服务器只维护了一个时间事件节点该时间事件处理函数为serverCron执行了所有需要周期性执行的一些定时任务。事件是理解Redis的基石希望读者能认真学习。最后本文介绍了服务器处理客户端命令请求的整个流程包括服务器启动监听接收命令请求并解析执行命令请求返回命令回复等。
