概要
本部分内容主要是研究单机数据库。分别介绍单机数据库的实现原理，数据库的持久化，Redis事件，服务器维护管理客户端以及单机服务器的运作机制。
数据库
-
数据库结构
Redis数据库由redis.h/redisDb定义，结构如下：typedef struct redisDb { dict *dict; //字典，用来保存这个数据库中所有的键值对 dict *expires; //字典，用来存放过期时间 dict *blocking_keys; /* Keys with clients waiting for data (BLPOP) */ dict *ready_keys; /* Blocked keys that received a PUSH */ dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */ struct evictionPoolEntry *eviction_pool; // int id; //数据库ID long long avg_ttl; //平均存活时间，用于统计 } redisDb;一个 RedisServer 可以有多个 RedisDB ，默认情况下为 16 。 RedisServer 用一个 RedisDb 的数组保存对应的数据库。而 RedisClient 也有一个指向 RedisDB 的指针，表示要操作的数据库。当 RedisClient 调用
SELECT命令切换数据库时，实际上就是切换指针指向的 RedisDB 对象。 -
数据库键空间
RedisDb中的*dict保存了数据库所有的键值对，因此这个字典也被称为键空间（key space）。键空间的键为字符串对象，而键空间的值可以是五种对象的任意一种。- 添加新键：添加一个新键值对到数据库的操作实际上就是往键空间中添加一个新键值对。
- 删除键：删除实际上就是删除键空间字典中一个键值对。
- 更新键：更新键实际上就是更新键空间字典中键值对的值。
- 取值：就是根据键空间字典中的键，取对应值。
-
设置键的生存时间或过期时间
- 可以通过EXPIRE（单位时间为秒）和PEXPIRE（单位时间为毫秒）为数据中的某个键设置生存时间大（TTL）。
- 或者通过EXPIREAT（单位时间为秒）和PEXPIREAT（单位时间为毫秒）为数据中的某个键设置过期时间。
无论是哪个命令，最后都会转换成调用PEXPIREAT命令。（设置生存时间的命令EXPIRE和PEXPIRE，会根据生存时间计算出过期时间，再调用PEXPIREAT，而EXPIREAT会转换单位时间，再调用PEXPREAT）。
过期时间保存在expires的字典中。其中字典的键指向 key space 中对应的键，值是一个long long类型的整形，表示毫秒精度的UNIX时间戳。
PERSIST命令和TTL命令：
PERSIST命令可以移除一个键的过期时间TTL（PTTL）可以计算一个键的生存时间
- 过期键删除策略
- 定时删除：为过期键创建定时器，在键过期是删除。优点：最节约内存，缺点：占CPU时间。
- 惰性删除：键被获取时，去判断是否过期，如果过期就删除该键，否则返回值。优点：节约CPU时间，缺点：浪费内存，甚至存在内存泄漏。
- 定期删除：每隔一段时间，对数据库检查一次，删除已经过期的键。
- Redis过期删除的策略
Redis采用惰性删除和定期删除结合的方式处理过期键。一方面，在获取值时，如果键已经过期，那么删除该键，另一方面，RedisServer中有一个serverCron在定期的执行任务，其中的一项工作内容就是清理过期间（它的工作原理是在指定运行时间内，遍历各个数据库，从每个数据库随机抽选部分的键，检查是否过期，如果过期则删除，如果到达了指定运行时间，则结束等待下次继续运行）。 - AOF，RDB和复制对过期键的处理
- RDB在SAVE或BGSAVE时，不会将过期的键保存至RDB文件中
- AOF写入的过程中，会在过期键被删除时，主动增加一条DEL命令至AOF文件。
- AOF重写和保存RDB文件相似，过期的键将不会被重写。
- 复制模式下，从服务器的过期键删除有主服务器控制，主服务器删除过期键后会向从服务器发送DEL命令，而从服务器在获取过期键时，不会主动删除，因此仍能取到过期键的值。
- 数据库通知
数据库通知通过让客户端订阅特定的频道，并在操作后，依据配置的事件往频道中发布消息。
RDB持久化
由于Redis的在运行时，数据都保存在内存中，为了避免进程关闭或是断电等因素造成的数据丢失，REDIS提供了持久化的策略。RDB便是其中一种。RDB可以简单的理解为对Redis内存中的数据进行一次dump。
- RDB文件的创建和载入
- RDB文件的创建：
Redis可以通过SAVE命令和BGSAVE命令生成RDB文件。SAVE命令由服务器主进程运行，其在运行时，将会阻塞服务器进程，也就是说此时服务器无法处理其他命令。而BGSAVE命令则是由主进程创建一个子进程，由子进程执行，此时主进程依旧可以处理其他命令。 - RDB文件的载入：
Redis服务器在启动时会自动执行文件载入的命令，因此没有特定的命令用于RDB文件的载入。并且直到载入完成，Redis服务器才可以接受客户端连接。
- RDB文件的创建：
- 自动间歇性保存
Reids允许用户通过设置save条件，让服务器每隔一段时间自动执行一次BGSAVE命令。save配置了指定时间内，服务器如果发生指定次的操作，那么则进行一次BGSAVE。可以设置多个save条件，当一个条件满足时，BGSAVE便会发生。
redisServer中的dirty和lastsave分别用来保存上次SAVE到现在共发生了多少次操作和上次SAVE发生的时间。同时服务器的周期运行函数serverCron会检查保存条件是否满足。 - RDB文件结构
RDB的文件结构由最粗略的分可以分为以下几个部分：
REDIS|db_version|databases|EOF|check_sum
- REDIS是一个常量，5个字节，用来快速校验是否为RDB文件。
- db_version长度为4字节，表示当前数据版本。
- databases为数据库部分，记录每个数据库及其所有键值对数据。如果数据库全为空，则这部分也为空。
- EOF：结束符常量，1个字节。表示数据库部分到此为止。
- check_sum：校验文件是否损坏或被修改。
其中databases部分可以保存任意多个非空数据库。比如一个保存了0号和3号数据库的datavases部分如下：
database 0| database3
如果对其中一个database划分，又可以分为3个部分：
SELECTDB|db_number|key_value_pairs
- SELECTDB：常量，长度为1字节。可以通过这个值判断下面将读入一个数据库号
- db_number：数据库号码，当读入db_number时，服务器会调用SELECT命令，切换到相应的数据库。
- key_value_pairs：表示数据库键值对，如果键值对带有过期时间，会被一定保存。
其中key_value_pairs的结构如下：
EXPIRETIME_MS|ms|TYPE|key|value
- EXPIRETIME_MS：常量，1个字节。用来告诉服务器接下去要读入过期时间。
- ms：8字节带符号整数，用来记录一个UNIX时间戳，即该键的过期时间。
- TYPE：记录了value的类型，常量，一个字节。
- key:字符串对象，记录了键值对的键。
- value:根据TYPE的不同，编码方式和解析方式也不同，记录了键值对的值。
其中第1点和第2点在未设置过期时间的键上不存在。
AOF持久化
AOF实现原理
AOF持久化是通过保存Redis服务器执行的写命令来记录数据库状态的。
AOF持久化分三个阶段：命令追加（append），文件写入和文件同步。
- 命令追加：将写命令追加到Redis内部的aof_buf中。
- 文件写入：是调用操作系统的写操作将aof_buf中的内容写入文件。
- 文件同步：调用操作系统的写操作并不一定会直接将数据写入到磁盘，操作系统出于效率的考虑，可能会将数据先放入另一个缓存中，到一定时间后，才将缓存中的数据写入磁盘。
上述步骤，用流程图表示如下：
尤其需要注意步骤二和步骤三的区别。
AOF文件载入与数据还原
通过AOF文件还原数据库状态即是AOF文件载入与数据还原的过程。这一过程大致可以分为四个步骤：
- AOF创建一个虚拟客户端，用于模拟执行写命令
- 从AOF中读取一条写命令
- 虚拟客户端执行写命令
- 重复执行2，3步骤，直到AOF文件读取完毕
AOF重写
- 为什么需要重写？
由于AOF是通过追加写命令来记录数据库状态的，当数据库执行的写命令持续增加，那么AOF文件必然会增大，AOF增大不仅会造成服务器磁盘空间的浪费，影响服务器运行，还会造成AOF文件恢复时间变长。 - 什么叫AOF重写？
重写便是用新的较小的AOF文件代替旧AOF文件，合并旧文件中的冗余和无效的命令。 - Redis如何实现重写？
实现重写最简单的策略并非是分析旧AOF文件，并合并。而是通过数据库当前的状态，用一条可以得当相同结果的写命令代替（有些类似RDB在某一时刻dump数据库状态的概念，不过RDB是针对数据库中的数据，而AOF则是针对写命令）。
AOF重写过程也可以分为以下几个步骤：- 创建新AOF文件
- 遍历DB及DB下的key
- 读取该key，并根据键类型的不同，重写时采用不同的写命令
- 如果该key带有过期时间，则重写过期时间
- 重复上述步骤，直到遍历完成，关闭文件
注意：
上述过程只是将写命令写入AOF文件，并非是执行写命令。
同时对于多元素的写命令时，为了避免输入缓冲区溢出，超出限制的命令将被拆为多个写命令。
- 什么是AOF后台重写？
由于AOF重写时，会使线程长时间阻塞，导致Redis服务器在此期间无法处理客户端请求。因此Redis会让AOF重写由子进程执行，而父进程可以继续处理请求。
但是在子进程重写期间，父进程新接受的命令会导致子进程重写完成后数据库状态不一致。
为了解决不一致的问题，父进程在开始AOF重写后，每次执行写命令会同时将写命令发送给aof_buf和AOF重写缓冲区。当子进程重写结束后，发信号通知父进程，父进程在将AOF重写缓冲区中的内容追加到新的AOF文件，并原子性的替代旧文件。这样一来，AOF的后台重写变完成了。而Redis服务器也只会在父进程收到子进程信号到完成文件替代这一瞬间。
事件
Redis服务器是一个事件驱动程序，主要处理两类事件：
- 文件事件 file event：套接字相关的操作会产生对应的文件事件，服务器通过监听并处理这些事件来完成网络通信操作。
- 时间事件 time event:服务器某些操作 (如 serverCron) 需要在特定的时间点去执行，这类操作就是定时时间。
在开始了解Redis事件前，有必要先认识src/ae.h下定义的结构体( ae 表示 A simple event-driven programming library)：
aeFileEvent文件事件结构体
/* File event structure */
typedef struct aeFileEvent {
//事件的标志位
int mask; /* one of AE_(READABLE|WRITABLE) */
//绑定的读处理器
aeFileProc *rfileProc;
//绑定的写处理器
aeFileProc *wfileProc;
//数据
void *clientData;
} aeFileEvent;
aeTimeEvent时间事件结构体
/* Time event structure */
typedef struct aeTimeEvent {
//时间事件的ID
long long id; /* time event identifier. */
//时间事件的达到时间 秒
long when_sec; /* seconds */
//时间事件的到达时间 毫秒
long when_ms; /* milliseconds */
//事件处理器
aeTimeProc *timeProc;
//被清理时处理函数
aeEventFinalizerProc *finalizerProc;
//数据
void *clientData;
//下一个时间事件，构成链表，可以向下遍历
struct aeTimeEvent *next;
} aeTimeEvent;
aeFireEvent待处理的文件事件结构体
/* A fired event */
typedef struct aeFiredEvent {
//文件描述符
int fd;
//文件事件类型
int mask;
} aeFiredEvent;
aeEventLoop事件处理主循环
/* State of an event based program */
typedef struct aeEventLoop {
//最大的文件描述符
int maxfd; /* highest file descriptor currently registered */
//事件循环的最大容量
int setsize; /* max number of file descriptors tracked */
//下一时间事件的Id
long long timeEventNextId;
time_t lastTime; /* Used to detect system clock skew */
//文件事件列表
aeFileEvent *events; /* Registered events */
//待处理文件事件列表
aeFiredEvent *fired; /* Fired events */
//时间事件列表
aeTimeEvent *timeEventHead;
//结束的标志位
int stop;
void *apidata; /* This is used for polling API specific data */
aeBeforeSleepProc *beforesleep;
} aeEventLoop;
文件事件
Redis基于Reactor模式开发了网络事件处理器：使用I/O多路复用达到监听多个套接字的目的，当一个或多个套接字准备就绪时，产生对应的文件事件，由相应的时间处理器处理。
文件处理器由四个部分构成：
- 套接字
- 多路复用程序（底层可能是select,poll,epoll,kqueue等）
- 文件事件分派器
- 事件处理器
文件事件处理流程大致如下，里面涉及了部分代码（有兴趣的可以参考 redis src目录下的ae.c，ae_epoll.c等文件。）：
时间事件
Redis的时间时间主要分为两类：
- [ ] 定时事件：在某个特定的时间点执行
- [ ] 周期事件：每隔一段时间执行
目前Redis版本中只有周期事件，而没有定时事件。
Redis时间事件的实现
前面介绍的aeEventLoop中有一个aeTimeEvent *timeEventHead的事件时间链表，redis会在处理时间事件时，遍历这个链表，并处理已到达的时间事件。具体处理过程见代码src/ae.h的processTimeEvents:
static int processTimeEvents(aeEventLoop *eventLoop) {
int processed = 0;
aeTimeEvent *te;
long long maxId;
time_t now = time(NULL);
/* If the system clock is moved to the future, and then set back to the
* right value, time events may be delayed in a random way. Often this
* means that scheduled operations will not be performed soon enough.
*
* Here we try to detect system clock skews, and force all the time
* events to be processed ASAP when this happens: the idea is that
* processing events earlier is less dangerous than delaying them
* indefinitely, and practice suggests it is. */
//如果系统时钟被重设，则将时间事件的when值更新成0，避免事件时间被延期唤醒
if (now < eventLoop->lastTime) {
te = eventLoop->timeEventHead;
while(te) {
te->when_sec = 0;
te = te->next;
}
}
eventLoop->lastTime = now;
te = eventLoop->timeEventHead;
maxId = eventLoop->timeEventNextId-1;
//遍历时间时间链表
while(te) {
long now_sec, now_ms;
long long id;
if (te->id > maxId) {
te = te->next;
continue;
}
aeGetTime(&now_sec, &now_ms);
//处理已经达到的时间事件
if (now_sec > te->when_sec ||
(now_sec == te->when_sec && now_ms >= te->when_ms))
{
int retval;
id = te->id;
//根据返回值，来确定更新时间事件的when值，还是取消事件
retval = te->timeProc(eventLoop, id, te->clientData);
processed++;
/* After an event is processed our time event list may
* no longer be the same, so we restart from head.
* Still we make sure to don't process events registered
* by event handlers itself in order to don't loop forever.
* To do so we saved the max ID we want to handle.
*
* FUTURE OPTIMIZATIONS:
* Note that this is NOT great algorithmically. Redis uses
* a single time event so it's not a problem but the right
* way to do this is to add the new elements on head, and
* to flag deleted elements in a special way for later
* deletion (putting references to the nodes to delete into
* another linked list). */
if (retval != AE_NOMORE) {
aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
} else {
aeDeleteTimeEvent(eventLoop, id);
}
te = eventLoop->timeEventHead;
} else {
te = te->next;
}
}
return processed;
}
以流程图表示：
时间事件的应用实例：serverCron函数
目前Redis的事件时间只有周期性时间，而基本上运行的事件也只有serverCron一个事件，因此时间事件链表基本退化成了一个指针。serverCron函数主要处理以下四大类任务：
- 统计：统计服务各类信息，内存，CPU等
- 资源清理：清理过期键和失效的客户端
- 持久化：定期进行AOF和RDB的持久化操作
- 多节点的同步工作：主从的同步和集群下的同步
事件的调度与执行
由于Redis中同时存在文件事件和时间事件两种类型，所以服务器需要调度两种事件，决定如何处理。
事件调度的主入口在src/ae.c的aeMain中：
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
//主循环
while (!eventLoop->stop) {
//开始事件处理前的回掉
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}
事件分发器aeProcessEvents（调度）:
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
int processed = 0, numevents;
/* Nothing to do? return ASAP */
//如果没注册时间事件和文件事件，直接返回
if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
/* Note that we want call select() even if there are no
* file events to process as long as we want to process time
* events, in order to sleep until the next time event is ready
* to fire. */
if (eventLoop->maxfd != -1 ||
((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
int j;
aeTimeEvent *shortest = NULL;
struct timeval tv, *tvp;
//获取最近到达事件的到达时间，并计算时间差
if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
shortest = aeSearchNearestTimer(eventLoop);
if (shortest) {
long now_sec, now_ms;
/* Calculate the time missing for the nearest
* timer to fire. */
aeGetTime(&now_sec, &now_ms);
tvp = &tv;
tvp->tv_sec = shortest->when_sec - now_sec;
if (shortest->when_ms < now_ms) {
tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
tvp->tv_sec --;
} else {
tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
}
if (tvp->tv_sec < 0) tvp->tv_sec = 0;
if (tvp->tv_usec < 0) tvp->tv_usec = 0;
} else {
/* If we have to check for events but need to return
* ASAP because of AE_DONT_WAIT we need to set the timeout
* to zero */
if (flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
} else {
/* Otherwise we can block */
tvp = NULL; /* wait forever */
}
}
//让I/O多路复用程序阻塞上述计算的时间差，这样即避免了频繁轮询，也不会让时间事件太晚被处理
numevents = aeApiPoll(eventLoop, tvp);
//处理就绪的文件事件
for (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;
int rfired = 0;
/* note the fe->mask & mask & ... code: maybe an already processed
* event removed an element that fired and we still didn't
* processed, so we check if the event is still valid. */
//读事件，则调用读处理器
if (fe->mask & mask & AE_READABLE) {
rfired = 1;
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
}
//写事件，则调用写处理器
if (fe->mask & mask & AE_WRITABLE) {
if (!rfired || fe->wfileProc != fe->rfileProc)
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
}
processed++;
}
}
/* Check time events */
//检查并处理已经到达的时间事件
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
//一次处理完成
return processed; /* return the number of processed file/time events */
}
上述过程用流程图的表示：
客户端
每个与服务器进行连接的客户端，服务器都为这些客户端建立了相应的redis.h/redisClient的数据接口。在redis.h/redisServer的clients属性是一个链表，保存了所有与服务器连接的redisClient数据结构，具体结构如下:
/* With multiplexing we need to take per-client state.
* Clients are taken in a linked list. */
typedef struct redisClient {
//客户端ID
uint64_t id; /* Client incremental unique ID. */
//客户端对应套接字文件描述符
int fd;
//客户端select的DB
redisDb *db;
//DB的id
int dictid;
//客户端名字
robj *name; /* As set by CLIENT SETNAME */
//输入缓冲区
sds querybuf;
//输入缓冲区峰值
size_t querybuf_peak; /* Recent (100ms or more) peak of querybuf size */
//参数长度
int argc;
//参数数组
robj **argv;
//执行的命令
struct redisCommand *cmd, *lastcmd;
//请求类型
int reqtype;
//剩余未读取的命令内容
int multibulklen; /* number of multi bulk arguments left to read */
//命令内容长度
long bulklen; /* length of bulk argument in multi bulk request */
//变长输出缓冲区
list *reply;
unsigned long reply_bytes; /* Tot bytes of objects in reply list */
int sentlen; /* Amount of bytes already sent in the current
buffer or object being sent. */
//客户端创建时间
time_t ctime; /* Client creation time */
//客户端上次交流时间
time_t lastinteraction; /* time of the last interaction, used for timeout */
//上次达到输出缓冲区软限制的时间
time_t obuf_soft_limit_reached_time;
//客户端标志
int flags; /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */
//是否授权
int authenticated; /* when requirepass is non-NULL */
int replstate; /* replication state if this is a slave */
int repl_put_online_on_ack; /* Install slave write handler on ACK. */
int repldbfd; /* replication DB file descriptor */
off_t repldboff; /* replication DB file offset */
off_t repldbsize; /* replication DB file size */
sds replpreamble; /* replication DB preamble. */
long long reploff; /* replication offset if this is our master */
long long repl_ack_off; /* replication ack offset, if this is a slave */
long long repl_ack_time;/* replication ack time, if this is a slave */
char replrunid[REDIS_RUN_ID_SIZE+1]; /* master run id if this is a master */
int slave_listening_port; /* As configured with: SLAVECONF listening-port */
multiState mstate; /* MULTI/EXEC state */
int btype; /* Type of blocking op if REDIS_BLOCKED. */
blockingState bpop; /* blocking state */
long long woff; /* Last write global replication offset. */
list *watched_keys; /* Keys WATCHED for MULTI/EXEC CAS */
dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */
list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */
sds peerid; /* Cached peer ID. */
/* Response buffer */
//输出缓冲区
int bufpos;
char buf[REDIS_REPLY_CHUNK_BYTES];
} redisClient;
上述结构体中与本节内容相关的字段都添加了注释，其余字段会在涉及相关内容时介绍。
客户端属性
- fd（文件描述符）:表示客户端关联的套接字文件描述符，和网络无关的伪客户端 fd 为 -1 ，普通客户端值均大于 -1 。
- name（名字）：name是一个robj对象，默认情况下为NULL，如果调用setname设置了名字后，其将指向一个字符串对象象，内容为设置的名字。
- flags（标志）：标志可以表示客户端的角色，也可以表示客户端的状态，多个标志可以用
|来表示。 - querybuf（输入缓冲区）：输入缓冲区用来保存客户端向服务器发送的请求，为一个SDS，可以动态的扩大和缩小。当queryBuf增长过大后，将由
serverCron来调整。 - argv（命令参数）和 argc（命令参数数量）：argv是一个数组，每一项都是字符串对象，argv[0]表示要执行的后面，之后的其他项是命令所需的参数。argc则是argv中命令和参数项的和。 argv 和 argc 由 querybuf 中解析而来。
- cmd（命令执行函数）：当服务器从querybuf中分析得出argv和argc属性后，服务器将根据argv[0]的值从命令表中查询命令对应的实现函数，并赋值给cmd。命令查找时，不区分大小写。
- buf（固定输出缓冲区）和 bufpos（缓冲区大小）：固定缓冲区用来保存长度较小的回复，是一个字符数组，bufpos表示数组中使用的字节数量。
- *reply（可变缓冲区）：对于固定缓冲区放不下的回复，将放置在可变缓冲区中，可变缓冲区是一个字符串链表，链表中每一项都是字符串，因此可以保存一个非常长的回复。
- authenticated（身份验证）：表示该客户端是否通过身份验证，当服务器开启身份验证功能时，authenticated 为0（表示未验证）的客户端除验证命令（AUTH）外的其他命令均会被拒绝。
- ctime（客户端创建时间）：表示客户端创建的时间。
- lastinteraction（最后交流时间）：客户端与服务器最后一次交流的时间，用来计算客户端的空闲时间。
- obuf_soft_limt_reached_time（最后到达输出缓冲区软限制时间）：记录了客户端上次到达客户端输出软限制的时间。
客户端创建与关闭
-
客户端创建：当客户端使用connect函数连接时，服务器会为这个新的客户端创建redisClient结构，并添加至*client链表的末端。
-
客户端关闭：
客户端关闭有以下几种情况- [ ] 客户端网络被关闭
- [ ] 客户端发送了不符合协议的数据
- [ ] 客户端成为CLIENT KILL的目标
- [ ] 服务器设置了超时时间，且客户端空闲时间大于超时时间（BLPOP阻塞或是在PUB/SUB的客户端不受限制）
- [ ] 客户端发送数据超过了输入缓冲区
- [ ] 给客户端回复的数据超过了硬限制(hard limt)，或是给客户端回复的数据超出了软限制(soft limt)且在监测时间内，均超出软限制
-
Lua脚本伪客户端和AOF伪客户端，Lua脚本伪客户端会在服务器初始化时创建，负责执行Lua脚本，一直到服务关闭才会被关闭。而AOF伪客户端会在AOF文件载入时被创建，载入完成后被关闭。
服务端
命令执行过程
- 客户端发送请求
- 客户端与服务器间的套接字变得可用，服务器读取套接字上的数据，并放在client的querybuf中（
readQueryFromClient(） - 解析数据，设置argv和argc（
processInputBuffer()） - 命令处理
- 根据argv[0]查找对应的redisCommand，并将client的cmd指针指向找到的对象（
processCommand()） - 执行 command 前对信息进行校验
- 调用 command 对应的函数 （
call()） - 将结果设置到输出缓冲区（
addReply()） - 更新统计信息及同步工作
- 根据argv[0]查找对应的redisCommand，并将client的cmd指针指向找到的对象（
- 命令回复，在套接字可写时，将缓冲区的内容输出
serverCron函数
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
int j;
//避免无用参数发出警告
UNUSED(eventLoop);
UNUSED(id);
UNUSED(clientData);
if (server.watchdog_period) watchdogScheduleSignal(server.watchdog_period);
//更新server结构体中的时间，避免重复调用系统时间函数
updateCachedTime();
server.hz = server.config_hz;
//根据客户端连接数，动态调整频率
if (server.dynamic_hz) {
while (listLength(server.clients) / server.hz >
MAX_CLIENTS_PER_CLOCK_TICK)
{
server.hz *= 2;
if (server.hz > CONFIG_MAX_HZ) {
server.hz = CONFIG_MAX_HZ;
break;
}
}
}
//以100ms周期运行,统计命令执行次数，网络输入输出字节
//run_with_period是一个宏，最快只能以hz的频率执行
run_with_period(100) {
trackInstantaneousMetric(STATS_METRIC_COMMAND,server.stat_numcommands);
trackInstantaneousMetric(STATS_METRIC_NET_INPUT,
server.stat_net_input_bytes);
trackInstantaneousMetric(STATS_METRIC_NET_OUTPUT,
server.stat_net_output_bytes);
}
//更新内存使用峰值
if (zmalloc_used_memory() > server.stat_peak_memory)
server.stat_peak_memory = zmalloc_used_memory();
run_with_period(100) {
zmalloc_get_allocator_info(&server.cron_malloc_stats.allocator_allocated,
&server.cron_malloc_stats.allocator_active,
&server.cron_malloc_stats.allocator_resident);
if (!server.cron_malloc_stats.allocator_resident) {
size_t lua_memory = lua_gc(server.lua,LUA_GCCOUNT,0)*1024LL;
server.cron_malloc_stats.allocator_resident = server.cron_malloc_stats.process_rss - lua_memory;
}
if (!server.cron_malloc_stats.allocator_active)
server.cron_malloc_stats.allocator_active = server.cron_malloc_stats.allocator_resident;
if (!server.cron_malloc_stats.allocator_allocated)
server.cron_malloc_stats.allocator_allocated = server.cron_malloc_stats.zmalloc_used;
}
//处理SIGTERM信号
if (server.shutdown_asap) {
if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) exit(0);
serverLog(LL_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
server.shutdown_asap = 0;
}
//日志记录数据库状态
run_with_period(5000) {
for (j = 0; j < server.dbnum; j++) {
long long size, used, vkeys;
size = dictSlots(server.db[j].dict);
used = dictSize(server.db[j].dict);
vkeys = dictSize(server.db[j].expires);
if (used || vkeys) {
serverLog(LL_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
/* dictPrintStats(server.dict); */
}
}
}
//日志记录客户端链接
if (!server.sentinel_mode) {
run_with_period(5000) {
serverLog(LL_VERBOSE,
"%lu clients connected (%lu replicas), %zu bytes in use",
listLength(server.clients)-listLength(server.slaves),
listLength(server.slaves),
zmalloc_used_memory());
}
}
//客户端周期处理
clientsCron();
//数据库周期处理
databasesCron();
//如果没有子进程在RDB或是AOF且有AOF重写被延迟
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&
server.aof_rewrite_scheduled)
{
rewriteAppendOnlyFileBackground();
}
//如果RDB子进程或者AOF紫子进程存在
if (server.rdb_child_pid != -1 || server.aof_child_pid != -1 ||
ldbPendingChildren())
{
int statloc;
pid_t pid;
//检查是否有子进程发来信号，如果有，则说明子进程任务完成，需要父进程进行后续操作
if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
int exitcode = WEXITSTATUS(statloc);
int bysignal = 0;
if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
if (pid == -1) {
serverLog(LL_WARNING,"wait3() returned an error: %s. "
"rdb_child_pid = %d, aof_child_pid = %d",
strerror(errno),
(int) server.rdb_child_pid,
(int) server.aof_child_pid);
} else if (pid == server.rdb_child_pid) {
backgroundSaveDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else if (pid == server.aof_child_pid) {
backgroundRewriteDoneHandler(exitcode,bysignal);
if (!bysignal && exitcode == 0) receiveChildInfo();
} else {
if (!ldbRemoveChild(pid)) {
serverLog(LL_WARNING,
"Warning, detected child with unmatched pid: %ld",
(long)pid);
}
}
updateDictResizePolicy();
closeChildInfoPipe();
}
} else {
//否则，看是否满足BGSAVE条件或是AOF重写条件
for (j = 0; j < server.saveparamslen; j++) {
struct saveparam *sp = server.saveparams+j;
/* Save if we reached the given amount of changes,
* the given amount of seconds, and if the latest bgsave was
* successful or if, in case of an error, at least
* CONFIG_BGSAVE_RETRY_DELAY seconds already elapsed. */
if (server.dirty >= sp->changes &&
server.unixtime-server.lastsave > sp->seconds &&
(server.unixtime-server.lastbgsave_try >
CONFIG_BGSAVE_RETRY_DELAY ||
server.lastbgsave_status == C_OK))
{
serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...",
sp->changes, (int)sp->seconds);
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
rdbSaveBackground(server.rdb_filename,rsiptr);
break;
}
}
/* Trigger an AOF rewrite if needed. */
if (server.aof_state == AOF_ON &&
server.rdb_child_pid == -1 &&
server.aof_child_pid == -1 &&
server.aof_rewrite_perc &&
server.aof_current_size > server.aof_rewrite_min_size)
{
long long base = server.aof_rewrite_base_size ?
server.aof_rewrite_base_size : 1;
long long growth = (server.aof_current_size*100/base) - 100;
if (growth >= server.aof_rewrite_perc) {
serverLog(LL_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth);
rewriteAppendOnlyFileBackground();
}
}
}
//将AOF刷新至文件
if (server.aof_flush_postponed_start) flushAppendOnlyFile(0);
//如果上次刷新失败，再次刷新
run_with_period(1000) {
if (server.aof_last_write_status == C_ERR)
flushAppendOnlyFile(0);
}
/* Clear the paused clients flag if needed. */
clientsArePaused(); /* Don't check return value, just use the side effect.*/
//复制相关功能
run_with_period(1000) replicationCron();
//集群模式
run_with_period(100) {
if (server.cluster_enabled) clusterCron();
}
//哨兵模式
if (server.sentinel_mode) sentinelTimer();
/* Cleanup expired MIGRATE cached sockets. */
run_with_period(1000) {
migrateCloseTimedoutSockets();
}
/* Stop the I/O threads if we don't have enough pending work. */
stopThreadedIOIfNeeded();
/* Start a scheduled BGSAVE if the corresponding flag is set. This is
* useful when we are forced to postpone a BGSAVE because an AOF
* rewrite is in progress.
*
* Note: this code must be after the replicationCron() call above so
* make sure when refactoring this file to keep this order. This is useful
* because we want to give priority to RDB savings for replication. */
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&
server.rdb_bgsave_scheduled &&
(server.unixtime-server.lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
server.lastbgsave_status == C_OK))
{
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
if (rdbSaveBackground(server.rdb_filename,rsiptr) == C_OK)
server.rdb_bgsave_scheduled = 0;
}
server.cronloops++;
return 1000/server.hz;
}
clientsCron客户端定时处理任务，主要负责清理超时客户端，释放客户端的输入缓冲区及记录客户端信息：
void clientsCron(void) {
//统计客户端数，将客户段拆分成几段，均摊至每次调用
int numclients = listLength(server.clients);
int iterations = numclients/server.hz;
mstime_t now = mstime();
if (iterations < CLIENTS_CRON_MIN_ITERATIONS)
iterations = (numclients < CLIENTS_CRON_MIN_ITERATIONS) ?
numclients : CLIENTS_CRON_MIN_ITERATIONS;
//对该段内的客户端，遍历处理
while(listLength(server.clients) && iterations--) {
client *c;
listNode *head;
listRotate(server.clients);
head = listFirst(server.clients);
c = listNodeValue(head);
//处理超时客户端
if (clientsCronHandleTimeout(c,now)) continue;
//处理客户端的query buf
if (clientsCronResizeQueryBuffer(c)) continue;
//统计客户端信息，内存，输入输出等
if (clientsCronTrackExpansiveClients(c)) continue;
}
}
databasesCron负责数据库清理，由于Redis的一个数据库是一个哈希表，当哈希表的数据经过一段时间的增加或删除，处于内存和查询效率的考虑，需要对哈希表resizing和rehashing，同时databasesCron也负责处理过期键。
void databasesCron(void) {
//清理过期键
if (server.active_expire_enabled) {
if (server.masterhost == NULL) {
activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
} else {
expireSlaveKeys();
}
}
/* Defrag keys gradually. */
if (server.active_defrag_enabled)
activeDefragCycle();
//如果没有子进程在RDB或是AOF重写
if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) {
static unsigned int resize_db = 0;
static unsigned int rehash_db = 0;
int dbs_per_call = CRON_DBS_PER_CALL;
int j;
if (dbs_per_call > server.dbnum) dbs_per_call = server.dbnum;
// Resize
for (j = 0; j < dbs_per_call; j++) {
tryResizeHashTables(resize_db % server.dbnum);
resize_db++;
}
// Rehash
if (server.activerehashing) {
for (j = 0; j < dbs_per_call; j++) {
int work_done = incrementallyRehash(rehash_db);
if (work_done) {
/* If the function did some work, stop here, we'll do
* more at the next cron loop. */
break;
} else {
/* If this db didn't need rehash, we'll try the next one. */
rehash_db++;
rehash_db %= server.dbnum;
}
}
}
}
}
以流程图概括如下：
初始化服务器
Redis在启动后便会执行服务器的初始步骤，我们可以在main函数中了解服务器启动的步骤：
int main(int argc, char **argv) {
struct timeval tv;
int j;
#ifdef REDIS_TEST
if (argc == 3 && !strcasecmp(argv[1], "test")) {
if (!strcasecmp(argv[2], "ziplist")) {
return ziplistTest(argc, argv);
} else if (!strcasecmp(argv[2], "quicklist")) {
quicklistTest(argc, argv);
} else if (!strcasecmp(argv[2], "intset")) {
return intsetTest(argc, argv);
} else if (!strcasecmp(argv[2], "zipmap")) {
return zipmapTest(argc, argv);
} else if (!strcasecmp(argv[2], "sha1test")) {
return sha1Test(argc, argv);
} else if (!strcasecmp(argv[2], "util")) {
return utilTest(argc, argv);
} else if (!strcasecmp(argv[2], "endianconv")) {
return endianconvTest(argc, argv);
} else if (!strcasecmp(argv[2], "crc64")) {
return crc64Test(argc, argv);
} else if (!strcasecmp(argv[2], "zmalloc")) {
return zmalloc_test(argc, argv);
}
return -1; /* test not found */
}
#endif
/* We need to initialize our libraries, and the server configuration. */
#ifdef INIT_SETPROCTITLE_REPLACEMENT
spt_init(argc, argv);
#endif
/******************* 程序运行基础环境设置 ***********************************/
setlocale(LC_COLLATE,"");
tzset(); /* Populates 'timezone' global. */
zmalloc_set_oom_handler(redisOutOfMemoryHandler);
srand(time(NULL)^getpid());
gettimeofday(&tv,NULL);
/*********************初始化服务器结构体 及初始化基本配置**********************/
char hashseed[16];
getRandomHexChars(hashseed,sizeof(hashseed));
dictSetHashFunctionSeed((uint8_t*)hashseed);
server.sentinel_mode = checkForSentinelMode(argc,argv);
//服务器的基本初始化，创建Server结构体，并初始化Server中的一些基本变量，以及创建命令函数字典表
initServerConfig();
ACLInit(); /* The ACL subsystem must be initialized ASAP because the
basic networking code and client creation depends on it. */
moduleInitModulesSystem();
//保存运行参数
server.executable = getAbsolutePath(argv[0]);
server.exec_argv = zmalloc(sizeof(char*)*(argc+1));
server.exec_argv[argc] = NULL;
for (j = 0; j < argc; j++) server.exec_argv[j] = zstrdup(argv[j]);
//哨兵模式的初始化配置
if (server.sentinel_mode) {
initSentinelConfig();
initSentinel();
}
/*********************运行参数解析 载入服务器配置***************************/
/* Check if we need to start in redis-check-rdb/aof mode. We just execute
* the program main. However the program is part of the Redis executable
* so that we can easily execute an RDB check on loading errors. */
if (strstr(argv[0],"redis-check-rdb") != NULL)
redis_check_rdb_main(argc,argv,NULL);
else if (strstr(argv[0],"redis-check-aof") != NULL)
redis_check_aof_main(argc,argv);
//处理执行输入参数
if (argc >= 2) {
j = 1; /* First option to parse in argv[] */
sds options = sdsempty();
char *configfile = NULL;
//基本信息参数
if (strcmp(argv[1], "-v") == 0 ||
strcmp(argv[1], "--version") == 0) version();
if (strcmp(argv[1], "--help") == 0 ||
strcmp(argv[1], "-h") == 0) usage();
if (strcmp(argv[1], "--test-memory") == 0) {
if (argc == 3) {
memtest(atoi(argv[2]),50);
exit(0);
} else {
fprintf(stderr,"Please specify the amount of memory to test in megabytes.
");
fprintf(stderr,"Example: ./redis-server --test-memory 4096
");
exit(1);
}
}
//配置文件路径
if (argv[j][0] != '-' || argv[j][1] != '-') {
configfile = argv[j];
server.configfile = getAbsolutePath(configfile);
/* Replace the config file in server.exec_argv with
* its absolute path. */
zfree(server.exec_argv[j]);
server.exec_argv[j] = zstrdup(server.configfile);
j++;
}
//其他配置选项参数
while(j != argc) {
if (argv[j][0] == '-' && argv[j][1] == '-') {
/* Option name */
if (!strcmp(argv[j], "--check-rdb")) {
/* Argument has no options, need to skip for parsing. */
j++;
continue;
}
if (sdslen(options)) options = sdscat(options,"
");
options = sdscat(options,argv[j]+2);
options = sdscat(options," ");
} else {
/* Option argument */
options = sdscatrepr(options,argv[j],strlen(argv[j]));
options = sdscat(options," ");
}
j++;
}
if (server.sentinel_mode && configfile && *configfile == '-') {
serverLog(LL_WARNING,
"Sentinel config from STDIN not allowed.");
serverLog(LL_WARNING,
"Sentinel needs config file on disk to save state. Exiting...");
exit(1);
}
resetServerSaveParams();
//加载服务器配置
loadServerConfig(configfile,options);
sdsfree(options);
}
serverLog(LL_WARNING, "oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo");
serverLog(LL_WARNING,
"Redis version=%s, bits=%d, commit=%s, modified=%d, pid=%d, just started",
REDIS_VERSION,
(sizeof(long) == 8) ? 64 : 32,
redisGitSHA1(),
strtol(redisGitDirty(),NULL,10) > 0,
(int)getpid());
if (argc == 1) {
serverLog(LL_WARNING, "Warning: no config file specified, using the default config. In order to specify a config file use %s /path/to/%s.conf", argv[0], server.sentinel_mode ? "sentinel" : "redis");
} else {
serverLog(LL_WARNING, "Configuration loaded");
}
/**************创建为守护进程 及初始化Server内部结构****************************************/
server.supervised = redisIsSupervised(server.supervised_mode);
int background = server.daemonize && !server.supervised;
if (background) daemonize();
initServer();
if (background || server.pidfile) createPidFile();
redisSetProcTitle(argv[0]);
redisAsciiArt();
checkTcpBacklogSettings();
if (!server.sentinel_mode) {
/* Things not needed when running in Sentinel mode. */
serverLog(LL_WARNING,"Server initialized");
#ifdef __linux__
linuxMemoryWarnings();
#endif
moduleLoadFromQueue();
ACLLoadUsersAtStartup();
/*************还原数据库状态********************************************************/
//从磁盘上读取RDB或是AOF文件，还原数据库状态
loadDataFromDisk();
if (server.cluster_enabled) {
if (verifyClusterConfigWithData() == C_ERR) {
serverLog(LL_WARNING,
"You can't have keys in a DB different than DB 0 when in "
"Cluster mode. Exiting.");
exit(1);
}
}
if (server.ipfd_count > 0)
serverLog(LL_NOTICE,"Ready to accept connections");
if (server.sofd > 0)
serverLog(LL_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket);
} else {
sentinelIsRunning();
}
/* Warning the user about suspicious maxmemory setting. */
if (server.maxmemory > 0 && server.maxmemory < 1024*1024) {
serverLog(LL_WARNING,"WARNING: You specified a maxmemory value that is less than 1MB (current value is %llu bytes). Are you sure this is what you really want?", server.maxmemory);
}
/********配置事件循环****************************************************************/
aeSetBeforeSleepProc(server.el,beforeSleep);
aeSetAfterSleepProc(server.el,afterSleep);
aeMain(server.el);
aeDeleteEventLoop(server.el);
return 0;
}
从上述代码中，我们可以看到初始化服务器一共可以分为6个步骤：
-
程序基本运行环境设置
基本运行环境设置主要是加载一些库，设置LOCATE等。 -
初步初始化服务器结构
初始化服务器结构主要的过程在initServerConfig中void initServerConfig(void) { int j; //设置系统时间缓存 updateCachedTime(); //随机设置服务器ID getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE); server.runid[CONFIG_RUN_ID_SIZE] = ' '; changeReplicationId(); clearReplicationId2(); //时区设置 server.timezone = getTimeZone(); /* Initialized by tzset(). */ //初始化基本配置 server.configfile = NULL; server.executable = NULL; server.hz = server.config_hz = CONFIG_DEFAULT_HZ; server.dynamic_hz = CONFIG_DEFAULT_DYNAMIC_HZ; server.arch_bits = (sizeof(long) == 8) ? 64 : 32; server.port = CONFIG_DEFAULT_SERVER_PORT; server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG; server.bindaddr_count = 0; server.unixsocket = NULL; server.unixsocketperm = CONFIG_DEFAULT_UNIX_SOCKET_PERM; server.ipfd_count = 0; server.sofd = -1; server.protected_mode = CONFIG_DEFAULT_PROTECTED_MODE; server.gopher_enabled = CONFIG_DEFAULT_GOPHER_ENABLED; server.dbnum = CONFIG_DEFAULT_DBNUM; server.verbosity = CONFIG_DEFAULT_VERBOSITY; server.maxidletime = CONFIG_DEFAULT_CLIENT_TIMEOUT; server.tcpkeepalive = CONFIG_DEFAULT_TCP_KEEPALIVE; server.active_expire_enabled = 1; server.active_defrag_enabled = CONFIG_DEFAULT_ACTIVE_DEFRAG; server.active_defrag_ignore_bytes = CONFIG_DEFAULT_DEFRAG_IGNORE_BYTES; server.active_defrag_threshold_lower = CONFIG_DEFAULT_DEFRAG_THRESHOLD_LOWER; server.active_defrag_threshold_upper = CONFIG_DEFAULT_DEFRAG_THRESHOLD_UPPER; server.active_defrag_cycle_min = CONFIG_DEFAULT_DEFRAG_CYCLE_MIN; server.active_defrag_cycle_max = CONFIG_DEFAULT_DEFRAG_CYCLE_MAX; server.active_defrag_max_scan_fields = CONFIG_DEFAULT_DEFRAG_MAX_SCAN_FIELDS; server.proto_max_bulk_len = CONFIG_DEFAULT_PROTO_MAX_BULK_LEN; server.client_max_querybuf_len = PROTO_MAX_QUERYBUF_LEN; server.saveparams = NULL; server.loading = 0; server.logfile = zstrdup(CONFIG_DEFAULT_LOGFILE); server.syslog_enabled = CONFIG_DEFAULT_SYSLOG_ENABLED; server.syslog_ident = zstrdup(CONFIG_DEFAULT_SYSLOG_IDENT); server.syslog_facility = LOG_LOCAL0; server.daemonize = CONFIG_DEFAULT_DAEMONIZE; server.supervised = 0; server.supervised_mode = SUPERVISED_NONE; server.aof_state = AOF_OFF; server.aof_fsync = CONFIG_DEFAULT_AOF_FSYNC; server.aof_no_fsync_on_rewrite = CONFIG_DEFAULT_AOF_NO_FSYNC_ON_REWRITE; server.aof_rewrite_perc = AOF_REWRITE_PERC; server.aof_rewrite_min_size = AOF_REWRITE_MIN_SIZE; server.aof_rewrite_base_size = 0; server.aof_rewrite_scheduled = 0; server.aof_last_fsync = time(NULL); server.aof_rewrite_time_last = -1; server.aof_rewrite_time_start = -1; server.aof_lastbgrewrite_status = C_OK; server.aof_delayed_fsync = 0; server.aof_fd = -1; server.aof_selected_db = -1; /* Make sure the first time will not match */ server.aof_flush_postponed_start = 0; server.aof_rewrite_incremental_fsync = CONFIG_DEFAULT_AOF_REWRITE_INCREMENTAL_FSYNC; server.rdb_save_incremental_fsync = CONFIG_DEFAULT_RDB_SAVE_INCREMENTAL_FSYNC; server.aof_load_truncated = CONFIG_DEFAULT_AOF_LOAD_TRUNCATED; server.aof_use_rdb_preamble = CONFIG_DEFAULT_AOF_USE_RDB_PREAMBLE; server.pidfile = NULL; server.rdb_filename = zstrdup(CONFIG_DEFAULT_RDB_FILENAME); server.aof_filename = zstrdup(CONFIG_DEFAULT_AOF_FILENAME); server.acl_filename = zstrdup(CONFIG_DEFAULT_ACL_FILENAME); server.rdb_compression = CONFIG_DEFAULT_RDB_COMPRESSION; server.rdb_checksum = CONFIG_DEFAULT_RDB_CHECKSUM; server.stop_writes_on_bgsave_err = CONFIG_DEFAULT_STOP_WRITES_ON_BGSAVE_ERROR; server.activerehashing = CONFIG_DEFAULT_ACTIVE_REHASHING; server.active_defrag_running = 0; server.notify_keyspace_events = 0; server.maxclients = CONFIG_DEFAULT_MAX_CLIENTS; server.blocked_clients = 0; memset(server.blocked_clients_by_type,0, sizeof(server.blocked_clients_by_type)); server.maxmemory = CONFIG_DEFAULT_MAXMEMORY; server.maxmemory_policy = CONFIG_DEFAULT_MAXMEMORY_POLICY; server.maxmemory_samples = CONFIG_DEFAULT_MAXMEMORY_SAMPLES; server.lfu_log_factor = CONFIG_DEFAULT_LFU_LOG_FACTOR; server.lfu_decay_time = CONFIG_DEFAULT_LFU_DECAY_TIME; server.hash_max_ziplist_entries = OBJ_HASH_MAX_ZIPLIST_ENTRIES; server.hash_max_ziplist_value = OBJ_HASH_MAX_ZIPLIST_VALUE; server.list_max_ziplist_size = OBJ_LIST_MAX_ZIPLIST_SIZE; server.list_compress_depth = OBJ_LIST_COMPRESS_DEPTH; server.set_max_intset_entries = OBJ_SET_MAX_INTSET_ENTRIES; server.zset_max_ziplist_entries = OBJ_ZSET_MAX_ZIPLIST_ENTRIES; server.zset_max_ziplist_value = OBJ_ZSET_MAX_ZIPLIST_VALUE; server.hll_sparse_max_bytes = CONFIG_DEFAULT_HLL_SPARSE_MAX_BYTES; server.stream_node_max_bytes = OBJ_STREAM_NODE_MAX_BYTES; server.stream_node_max_entries = OBJ_STREAM_NODE_MAX_ENTRIES; server.shutdown_asap = 0; server.cluster_enabled = 0; server.cluster_node_timeout = CLUSTER_DEFAULT_NODE_TIMEOUT; server.cluster_migration_barrier = CLUSTER_DEFAULT_MIGRATION_BARRIER; server.cluster_slave_validity_factor = CLUSTER_DEFAULT_SLAVE_VALIDITY; server.cluster_require_full_coverage = CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE; server.cluster_slave_no_failover = CLUSTER_DEFAULT_SLAVE_NO_FAILOVER; server.cluster_configfile = zstrdup(CONFIG_DEFAULT_CLUSTER_CONFIG_FILE); server.cluster_announce_ip = CONFIG_DEFAULT_CLUSTER_ANNOUNCE_IP; server.cluster_announce_port = CONFIG_DEFAULT_CLUSTER_ANNOUNCE_PORT; server.cluster_announce_bus_port = CONFIG_DEFAULT_CLUSTER_ANNOUNCE_BUS_PORT; server.cluster_module_flags = CLUSTER_MODULE_FLAG_NONE; server.migrate_cached_sockets = dictCreate(&migrateCacheDictType,NULL); server.next_client_id = 1; /* Client IDs, start from 1 .*/ server.loading_process_events_interval_bytes = (1024*1024*2); server.lazyfree_lazy_eviction = CONFIG_DEFAULT_LAZYFREE_LAZY_EVICTION; server.lazyfree_lazy_expire = CONFIG_DEFAULT_LAZYFREE_LAZY_EXPIRE; server.lazyfree_lazy_server_del = CONFIG_DEFAULT_LAZYFREE_LAZY_SERVER_DEL; server.always_show_logo = CONFIG_DEFAULT_ALWAYS_SHOW_LOGO; server.lua_time_limit = LUA_SCRIPT_TIME_LIMIT; server.io_threads_num = CONFIG_DEFAULT_IO_THREADS_NUM; server.io_threads_do_reads = CONFIG_DEFAULT_IO_THREADS_DO_READS; //更新LRU时钟 server.lruclock = getLRUClock(); resetServerSaveParams(); appendServerSaveParams(60*60,1); /* save after 1 hour and 1 change */ appendServerSaveParams(300,100); /* save after 5 minutes and 100 changes */ appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */ /* Replication related */ server.masterauth = NULL; server.masterhost = NULL; server.masterport = 6379; server.master = NULL; server.cached_master = NULL; server.master_initial_offset = -1; server.repl_state = REPL_STATE_NONE; server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT; server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA; server.repl_slave_ro = CONFIG_DEFAULT_SLAVE_READ_ONLY; server.repl_slave_ignore_maxmemory = CONFIG_DEFAULT_SLAVE_IGNORE_MAXMEMORY; server.repl_slave_lazy_flush = CONFIG_DEFAULT_SLAVE_LAZY_FLUSH; server.repl_down_since = 0; /* Never connected, repl is down since EVER. */ server.repl_disable_tcp_nodelay = CONFIG_DEFAULT_REPL_DISABLE_TCP_NODELAY; server.repl_diskless_sync = CONFIG_DEFAULT_REPL_DISKLESS_SYNC; server.repl_diskless_sync_delay = CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY; server.repl_ping_slave_period = CONFIG_DEFAULT_REPL_PING_SLAVE_PERIOD; server.repl_timeout = CONFIG_DEFAULT_REPL_TIMEOUT; server.repl_min_slaves_to_write = CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE; server.repl_min_slaves_max_lag = CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG; server.slave_priority = CONFIG_DEFAULT_SLAVE_PRIORITY; server.slave_announce_ip = CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP; server.slave_announce_port = CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT; server.master_repl_offset = 0; /* Replication partial resync backlog */ server.repl_backlog = NULL; server.repl_backlog_size = CONFIG_DEFAULT_REPL_BACKLOG_SIZE; server.repl_backlog_histlen = 0; server.repl_backlog_idx = 0; server.repl_backlog_off = 0; server.repl_backlog_time_limit = CONFIG_DEFAULT_REPL_BACKLOG_TIME_LIMIT; server.repl_no_slaves_since = time(NULL); /* Client output buffer limits */ for (j = 0; j < CLIENT_TYPE_OBUF_COUNT; j++) server.client_obuf_limits[j] = clientBufferLimitsDefaults[j]; /* Double constants initialization */ R_Zero = 0.0; R_PosInf = 1.0/R_Zero; R_NegInf = -1.0/R_Zero; R_Nan = R_Zero/R_Zero; /* Command table -- we initiialize it here as it is part of the * initial configuration, since command names may be changed via * redis.conf using the rename-command directive. */ //创建命令函数字典表 server.commands = dictCreate(&commandTableDictType,NULL); server.orig_commands = dictCreate(&commandTableDictType,NULL); populateCommandTable(); server.delCommand = lookupCommandByCString("del"); server.multiCommand = lookupCommandByCString("multi"); server.lpushCommand = lookupCommandByCString("lpush"); server.lpopCommand = lookupCommandByCString("lpop"); server.rpopCommand = lookupCommandByCString("rpop"); server.zpopminCommand = lookupCommandByCString("zpopmin"); server.zpopmaxCommand = lookupCommandByCString("zpopmax"); server.sremCommand = lookupCommandByCString("srem"); server.execCommand = lookupCommandByCString("exec"); server.expireCommand = lookupCommandByCString("expire"); server.pexpireCommand = lookupCommandByCString("pexpire"); server.xclaimCommand = lookupCommandByCString("xclaim"); server.xgroupCommand = lookupCommandByCString("xgroup"); /* Slow log */ server.slowlog_log_slower_than = CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN; server.slowlog_max_len = CONFIG_DEFAULT_SLOWLOG_MAX_LEN; /* Latency monitor */ server.latency_monitor_threshold = CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD; /* Debugging */ server.assert_failed = "<no assertion failed>"; server.assert_file = "<no file>"; server.assert_line = 0; server.bug_report_start = 0; server.watchdog_period = 0; /* By default we want scripts to be always replicated by effects * (single commands executed by the script), and not by sending the * script to the slave / AOF. This is the new way starting from * Redis 5. However it is possible to revert it via redis.conf. */ server.lua_always_replicate_commands = 1; }上述代码虽然很长，但是大部分的内容都是对服务器的配置做一个基本的初始化，除此之外的还做了设置服务器ID，设置时间缓存，创建命令函数字典表。
-
载入配置文件及选项
初始服务器基本配置只是根据默认值设置服务器属性，初始完成后，服务器还会结合配置文件及运行时输入参数在对配置做进一步调整 -
服务器结构体初始化
完成上面两部配置后，服务器的基本配置就已经确定，此时，服务器可以根据配置的情况，初始化服务器内部结构。（某些内部结构只有在确定服务器配置后，才能进行初始化，如DB）。初始化结构体的步骤主要在initServer中。void initServer(void) { int j; //信号处理，及设置信号处理器 signal(SIGHUP, SIG_IGN); signal(SIGPIPE, SIG_IGN); setupSignalHandlers(); //打开日志 if (server.syslog_enabled) { openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT, server.syslog_facility); } server.hz = server.config_hz; server.pid = getpid(); server.current_client = NULL; server.clients = listCreate(); server.clients_index = raxNew(); server.clients_to_close = listCreate(); server.slaves = listCreate(); server.monitors = listCreate(); server.clients_pending_write = listCreate(); server.clients_pending_read = listCreate(); server.slaveseldb = -1; /* Force to emit the first SELECT command. */ server.unblocked_clients = listCreate(); server.ready_keys = listCreate(); server.clients_waiting_acks = listCreate(); server.get_ack_from_slaves = 0; server.clients_paused = 0; server.system_memory_size = zmalloc_get_memory_size(); //创建共享对象，主要是一些常用的SDS createSharedObjects(); adjustOpenFilesLimit(); //创建事件循环 server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR); if (server.el == NULL) { serverLog(LL_WARNING, "Failed creating the event loop. Error message: '%s'", strerror(errno)); exit(1); } server.db = zmalloc(sizeof(redisDb)*server.dbnum); //开始监听端口 /* Open the TCP listening socket for the user commands. */ if (server.port != 0 && listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR) exit(1); /* Open the listening Unix domain socket. */ if (server.unixsocket != NULL) { unlink(server.unixsocket); /* don't care if this fails */ server.sofd = anetUnixServer(server.neterr,server.unixsocket, server.unixsocketperm, server.tcp_backlog); if (server.sofd == ANET_ERR) { serverLog(LL_WARNING, "Opening Unix socket: %s", server.neterr); exit(1); } anetNonBlock(NULL,server.sofd); } /* Abort if there are no listening sockets at all. */ if (server.ipfd_count == 0 && server.sofd < 0) { serverLog(LL_WARNING, "Configured to not listen anywhere, exiting."); exit(1); } /* Create the Redis databases, and initialize other internal state. */ //初始化服务器的DB for (j = 0; j < server.dbnum; j++) { server.db[j].dict = dictCreate(&dbDictType,NULL); server.db[j].expires = dictCreate(&keyptrDictType,NULL); server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL); server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType,NULL); server.db[j].watched_keys = dictCreate(&keylistDictType,NULL); server.db[j].id = j; server.db[j].avg_ttl = 0; server.db[j].defrag_later = listCreate(); } evictionPoolAlloc(); /* Initialize the LRU keys pool. */ //创建PUB/SUB相关结构 server.pubsub_channels = dictCreate(&keylistDictType,NULL); server.pubsub_patterns = listCreate(); listSetFreeMethod(server.pubsub_patterns,freePubsubPattern); listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern); //初始化循环次数，RDB和AOF子进程号为-1等 server.cronloops = 0; server.rdb_child_pid = -1; server.aof_child_pid = -1; server.rdb_child_type = RDB_CHILD_TYPE_NONE; server.rdb_bgsave_scheduled = 0; server.child_info_pipe[0] = -1; server.child_info_pipe[1] = -1; server.child_info_data.magic = 0; aofRewriteBufferReset(); server.aof_buf = sdsempty(); server.lastsave = time(NULL); /* At startup we consider the DB saved. */ server.lastbgsave_try = 0; /* At startup we never tried to BGSAVE. */ server.rdb_save_time_last = -1; server.rdb_save_time_start = -1; server.dirty = 0; resetServerStats(); /* A few stats we don't want to reset: server startup time, and peak mem. */ //初始化统计信息 server.stat_starttime = time(NULL); server.stat_peak_memory = 0; server.stat_rdb_cow_bytes = 0; server.stat_aof_cow_bytes = 0; server.cron_malloc_stats.zmalloc_used = 0; server.cron_malloc_stats.process_rss = 0; server.cron_malloc_stats.allocator_allocated = 0; server.cron_malloc_stats.allocator_active = 0; server.cron_malloc_stats.allocator_resident = 0; server.lastbgsave_status = C_OK; server.aof_last_write_status = C_OK; server.aof_last_write_errno = 0; server.repl_good_slaves_count = 0; /* Create the timer callback, this is our way to process many background * operations incrementally, like clients timeout, eviction of unaccessed * expired keys and so forth. */ //创建时间事件 if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) { serverPanic("Can't create event loop timers."); exit(1); } /* Create an event handler for accepting new connections in TCP and Unix * domain sockets. */ //创建文件时间，关联请求接收处理器 for (j = 0; j < server.ipfd_count; j++) { if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL) == AE_ERR) { serverPanic( "Unrecoverable error creating server.ipfd file event."); } } if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE, acceptUnixHandler,NULL) == AE_ERR) serverPanic("Unrecoverable error creating server.sofd file event."); /* Register a readable event for the pipe used to awake the event loop * when a blocked client in a module needs attention. */ if (aeCreateFileEvent(server.el, server.module_blocked_pipe[0], AE_READABLE, moduleBlockedClientPipeReadable,NULL) == AE_ERR) { serverPanic( "Error registering the readable event for the module " "blocked clients subsystem."); } /* Open the AOF file if needed. */ //打开AOF文件 if (server.aof_state == AOF_ON) { server.aof_fd = open(server.aof_filename, O_WRONLY|O_APPEND|O_CREAT,0644); if (server.aof_fd == -1) { serverLog(LL_WARNING, "Can't open the append-only file: %s", strerror(errno)); exit(1); } } /* 32 bit instances are limited to 4GB of address space, so if there is * no explicit limit in the user provided configuration we set a limit * at 3 GB using maxmemory with 'noeviction' policy'. This avoids * useless crashes of the Redis instance for out of memory. */ //根据操作系统 设置最大内存使用量 if (server.arch_bits == 32 && server.maxmemory == 0) { serverLog(LL_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now."); server.maxmemory = 3072LL*(1024*1024); /* 3 GB */ server.maxmemory_policy = MAXMEMORY_NO_EVICTION; } //集群/同步相关设置 if (server.cluster_enabled) clusterInit(); replicationScriptCacheInit(); //LUA脚本相关设置 scriptingInit(1); //慢查询日志相关设置 slowlogInit(); latencyMonitorInit(); //I/O初始化 bioInit(); initThreadedIO(); server.initial_memory_usage = zmalloc_used_memory(); } -
还原数据库状态
loadDataFromDisk从磁盘中读取RDB文件或AOF文件，并执行文件，从文还原数据库状态 -
执行事件循环
事件循环的主入口已经在上文中介绍过了，主要是调用aeMain，开始执行事件。
由此，服务器便初始化完成，可以处理客户端发送的请求。
以流程图表示服务器初始化过程：
