曹工说Redis源码(3)-- redis server 启动过程完整解析(中)

文章导航

Redis源码系列的初衷,是帮助我们更好地理解Redis,更懂Redis,而怎么才能懂,光看是不够的,建议跟着下面的这一篇,把环境搭建起来,后续可以自己阅读源码,或者跟着我这边一起阅读。由于我用c也是好几年以前了,些许错误在所难免,希望读者能不吝指出。

曹工说Redis源码(1)-- redis debug环境搭建,使用clion,达到和调试java一样的效果

曹工说Redis源码(2)-- redis server 启动过程解析及简单c语言基础知识补充

本讲主题

首先,会再补充一点c语言中,指针的相关知识;接下来,开始接着昨天的那篇,讲redis的启动过程,由大到小来讲,避免迅速陷入到细节中。

关于指针的理解

指针,其实就是指向一个内存地址,在知道这个地址前后存储的内容的前提下,这个指针可以被你任意解释。我举个例子:

typedef struct Test_Struct{
int a;
int b;
}Test_Struct; int main() {
// 1
void *pVoid = malloc(4);
// 2
memset(pVoid,0x01,4); // 3
int *pInt = pVoid;
// 4
char *pChar = pVoid;
// 5
short *pShort = pVoid;
// 6
Test_Struct *pTestStruct = pVoid; // 7
printf("address:%p, point to %d\n", pChar, *pChar);
printf("address:%p, point to %d\n", pShort, *pShort);
printf("address:%p, point to %d\n", pInt, *pInt);
printf("address:%p, point to %d\n", pTestStruct, pTestStruct->a);
}
  • 1处,分配一片内存,4个字节,32位;返回一个指针,指向这片内存区域,准确地说,指向第一个字节,因为分配的内存是连续的,你可以理解为数组。

    The malloc() function allocates size bytes and returns a pointer to the allocated memory.

  • 2处,调用memset,将这个pVoid 指向的内存开始的4个字节,设置为0x01,其实就是把每个字节设置为00000001。

    这个memset的注释如下:

    NAME
    memset - fill memory with a constant byte SYNOPSIS
    #include <string.h> void *memset(void *s, int c, size_t n); DESCRIPTION
    The memset() function fills the first n bytes of the memory area pointed to by s with the constant byte c.

    参考资料: https://www.cnblogs.com/yhlboke-1992/p/9292877.html

    这里我们把每个字节,设为0x01,最终的二进制,其实就是如下这样:

    曹工说Redis源码(3)-- redis server 启动过程完整解析(中)

  • 3处,定义int类型的指针,将pVoid赋值给它,int占4字节

  • 4处,定义char类型的指针,将pVoid赋值给它,char占1字节

  • 5处,定义short类型的指针,将pVoid赋值给它,short占2字节

  • 6处,定义Test_Struct类型的指针,这是个结构体,类似于高级语言的类,这个结构体的结构如下:

    typedef struct Test_Struct{
    int a;
    int b;
    }Test_Struct;

    同样,我们将pVoid赋值给它。

  • 7处,分别打印各类指针的地址,和对其解引用后的值。

输出如下:

曹工说Redis源码(3)-- redis server 启动过程完整解析(中)

257的二进制就是:0000 0001 0000 0001

16843009的二进制就是:0000 0001 0000 0001 0000 0001 0000 0001

结构体那个,也好理解,因为这个结构体,第一个属性a,就是int类型的,占4个字节。

另外,大家要注意,上面输出的指针地址都是一模一样的。

如果大家能理解这个demo,再看看这个链接,相信会更加理解指针:

C 指针的算术运算

redis server大致的启动过程

int main(int argc, char **argv) {
struct timeval tv; /**
* 1 设置时区等等
*/
setlocale(LC_COLLATE,"");
... // 2 检查服务器是否以 Sentinel 模式启动
server.sentinel_mode = checkForSentinelMode(argc,argv); // 3 初始化服务器配置
initServerConfig(); // 4
if (server.sentinel_mode) {
initSentinelConfig();
initSentinel();
} // 5 检查用户是否指定了配置文件,或者配置选项
if (argc >= 2) {
...
// 载入配置文件, options 是前面分析出的给定选项
loadServerConfig(configfile,options);
sdsfree(options);
} // 6 将服务器设置为守护进程
if (server.daemonize) daemonize(); // 7 创建并初始化服务器数据结构
initServer(); // 8 如果服务器是守护进程,那么创建 PID 文件
if (server.daemonize) createPidFile(); // 9 为服务器进程设置名字
redisSetProcTitle(argv[0]); // 10 打印 ASCII LOGO
redisAsciiArt(); // 11 如果服务器不是运行在 SENTINEL 模式,那么执行以下代码
if (!server.sentinel_mode) {
// 从 AOF 文件或者 RDB 文件中载入数据
loadDataFromDisk();
// 启动集群
if (server.cluster_enabled) {
if (verifyClusterConfigWithData() == REDIS_ERR) {
redisLog(REDIS_WARNING,
"You can't have keys in a DB different than DB 0 when in "
"Cluster mode. Exiting.");
exit(1);
}
}
// 打印 TCP 端口
if (server.ipfd_count > 0)
redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
} else {
sentinelIsRunning();
} // 12 运行事件处理器,一直到服务器关闭为止
aeSetBeforeSleepProc(server.el,beforeSleep);
aeMain(server.el); // 13 服务器关闭,停止事件循环
aeDeleteEventLoop(server.el); return 0;
}
  • 1,2,3处,在前面那篇中已经讲过,主要是初始化各种配置参数,比如socket相关的;redis.conf中涉及的,aof,rdb,replication,sentinel等;redis server自己内部的数据结构等,如runid,配置文件地址,服务器的相关信息(32位还是64位,因为redis直接运行在操作系统上,而不是像高级语言有虚拟机,32位和64位下,不同数据的长度是不同的),日志级别,最大客户端数量,客户端最大idle时间等等

  • 4处,因为sentinel和普通的redis server其实是共用同一份代码,所以这里启动时,要看是启动sentinel,还是启动普通的redis server,如果是启动sentinel,则进行sentinel相关配置

  • 5处,检查启动时的命令行参数中,是否指定了配置文件,如果指定了,要使用配置文件的配置为准

  • 6处,设置为守护进程

  • 7处,根据前面的配置,初始化redis server

  • 8处,创建pid文件,一般默认路径:/var/run/redis.pid,这个可以在redis.conf进行配置,如:

    pidfile "/var/run/redis_6379.pid"

  • 9处,为服务器进程设置名字

  • 10处,打印logo

  • 11处,如果不是sentinel模式启动的话,加载aof或rdb文件

  • 12处,跳入死循环,开始等待接收连接,处理客户端的请求;同时,周期执行后台任务,比如删除过期key等

  • 13处,服务器关闭,一般来说,走不到这里,一般都是陷入在12处的死循环中;只有在某些场景下,将一个全局变量stop修改为true后,程序会从12处跳出死循环,然后走到这里。

初始化redis server的过程

这一节,主要是细化前面的第7步操作,即初始化redis server。这一个函数,位于redis.c中,名为initServer,做的事情很多,接下来会顺序讲解。

设置全局的信号处理函数

    // 设置信号处理函数
signal(SIGHUP, SIG_IGN);
signal(SIGPIPE, SIG_IGN);
setupSignalHandlers();

最重要的是最后一行:

void setupSignalHandlers(void) {
// 1
struct sigaction act; /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction is used.
* Otherwise, sa_handler is used. */
sigemptyset(&act.sa_mask);
act.sa_flags = 0;
// 2
act.sa_handler = sigtermHandler;
// 3
sigaction(SIGTERM, &act, NULL); return;
}

3处,设置了:接收到SIGTERM信号时,使用act来处理信号,act在1处定义,是一个局部变量,它有一个字段,在2处被赋值,这是一个函数指针。函数指针类似于java中的一个static方法的引用,为什么是static,因为执行这类方法不需要new一个对象;在c语言中,所有的方法都是最*的,调用时,不需要new一个对象;所以,从这点来说,c语言的函数指针类似java中的static方法的引用。

我们可以看看2处,

    act.sa_handler = sigtermHandler;

这个sigtermHandler,应该就是一个全局函数了,看看其怎么被定义的:

// SIGTERM 信号的处理器
static void sigtermHandler(int sig) {
REDIS_NOTUSED(sig); redisLogFromHandler(REDIS_WARNING,"Received SIGTERM, scheduling shutdown..."); // 打开关闭标识
server.shutdown_asap = 1;
}

这个函数就是打开server这个全局变量的shutdown_asap。这个字段在以下地方被使用:

serverCron in redis.c

	/* We received a SIGTERM, shutting down here in a safe way, as it is
* not ok doing so inside the signal handler. */
// 服务器进程收到 SIGTERM 信号,关闭服务器
if (server.shutdown_asap) { // 尝试关闭服务器
if (prepareForShutdown(0) == REDIS_OK) exit(0); // 如果关闭失败,那么打印 LOG ,并移除关闭标识
redisLog(REDIS_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
server.shutdown_asap = 0;
}

以上这段代码的第一行,标识了这段代码所处的位置,为redis.c中的serverCron函数,这个函数,就是redis server的周期执行函数,类似于java中的ScheduledThreadPoolExecutor,当这个周期任务,检测到server.shutdown_asap打开后,就会去关闭服务器。

那,上面这个接收到信号,要执行的动作说完了,那么,什么是信号,信号其实是linux下进程间通讯的一种手段,比如kill -9 ,就会给对应的pid,发送一个SIGKILL 命令;在redis前台运行时,你按下ctrl + c,其实也是发送了一个信号,信号为SIGINT,值为2。大家可以看下图:

曹工说Redis源码(3)-- redis server 启动过程完整解析(中)

那么,前面我们注册的信号是哪个呢,是:SIGTERM,15。也就是我们按下kill -15时,会触发这个信号。

关于kill 9 和kill 15的差别,可以看这篇博客:

Linux kill -9 和 kill -15 的区别

开启syslog

// 设置 syslog
if (server.syslog_enabled) {
openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT,
server.syslog_facility);
}

这个就是发送日志到linux系统的syslog,可以看看openlog函数的说明:

send messages to the system logger

这个感觉用得不多,可以查阅:

redis 的syslog日志没有打印出来的探索过程

初始化当前redisServer的部分属性

	// 初始化并创建数据结构
server.current_client = NULL;
// 1
server.clients = listCreate();
server.clients_to_close = listCreate();
server.slaves = listCreate();
server.monitors = 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;

这个其实没啥说的,大家看到,比如1处,这个server.clients,server是一个全局变量,维护当前redis server的各种状态,clients呢,是用来保存当前连接到redis server的客户端,类型为链表:

    // 一个链表,保存了所有客户端状态结构
list *clients; /* List of active clients */

所以,这里其实就是调用listCreate(),创建了一个空链表,然后赋值给clients。

其他属性,类似。

创建常量字符串池,供复用

大家知道,redis在返回响应的时候,通常就是一句:"+OK"之类的。这个字符串,如果每次响应的时候,再去new一个,也太浪费了,所以,干脆,redis自己把这些常用的字符串,缓存了起来。

void createSharedObjects(void) {
int j; // 常用回复
shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
...
// 常用错误回复
shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
"-WRONGTYPE Operation against a key holding the wrong kind of value\r\n"));
...
}

这个和java中,把字符串字面量缓存起来,是一样的,都是为了提高性能;java里,不是还把128以内的整数也缓存了吗,对吧。

调整进程可以打开的最大文件数

服务器一般在真实线上环境,如果是需要应对高并发的话,可能会有几十上百万的客户端,和服务器上的某个进程,建立tcp连接,而这时候,一般就需要调整进程可以打开的最大文件数(socket也是文件)。

在阅读redis源码之前,我知道的,修改进程可以打开的最大文件数的方式是通过ulimit,具体的,大家可以看下面这两个链接:

linux最大文件句柄数量总结

Elasticsearch之优化

但是,在这个源码中,发现了另外一种方式:

  • 获取当前的指定资源的限制值的api
#define RLIMIT_NOFILE	5		/* max number of open files */

struct rlimit {
rlim_t rlim_cur;
rlim_t rlim_max;
};
struct rlimit limit; getrlimit(RLIMIT_NOFILE,&limit)

上面这个代码,获取当前系统中,NOFILE(进程最大文件数)这个值的资源限制大小。

通过man getrlimit(需要先安装,安装方式:yum install man-pages.noarch),可以看到:

曹工说Redis源码(3)-- redis server 启动过程完整解析(中)

  • setrlimit则可以设置资源的相关限制

    limit.rlim_cur = f;
    limit.rlim_max = f;
    setrlimit(RLIMIT_NOFILE,&limit)

创建事件循环相关数据结构

事件循环器的结构如下:

/*
* 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;

初始化上面这个数据结构的代码在:aeCreateEventLoop in redis.c

上面这个结构中,主要就是:

  1. apidata中,主要用于存储多路复用库的相关数据,每次调用多路复用库,去进行select时,如果发现有就绪的io事件发生,就会存放到 fired 属性中。

    比如,select就是linux下,老版本的linux内核中,多路复用的一种实现,redis中,其代码如下:

    static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
    ...
    // 1
    retval = select(eventLoop->maxfd+1,
    &state->_rfds,&state->_wfds,NULL,tvp);
    if (retval > 0) {
    for (j = 0; j <= eventLoop->maxfd; j++) {
    ...
    // 2
    eventLoop->fired[numevents].fd = j;
    eventLoop->fired[numevents].mask = mask;
    numevents++;
    }
    }
    return numevents;
    }

    省略了部分代码,其中,1处,进行select,这一步类似于java中nio的select操作;2处,将select返回的,已就绪的文件描述符,填充到fired 属性。

  2. 另外,我们提到过,redis有一些后台任务,比如清理过期key,这个不是一蹴而就的;每次周期运行后台任务时,就会去清理一部分,而这里的后台任务,其实就是上面这个数据结构中的时间事件。

        // 时间事件
    aeTimeEvent *timeEventHead;

分配16个数据库的内存空间

server.db = zmalloc(sizeof(redisDb) * server.dbnum);

打开listen端口,监听请求

    /* Open the TCP listening socket for the user commands. */
// 打开 TCP 监听端口,用于等待客户端的命令请求
listenToPort(server.port, server.ipfd, &server.ipfd_count)

这里就是打开平时的6379端口的地方。

初始化16个数据库对应的数据结构

    /* Create the Redis databases, and initialize other internal state. */
// 创建并初始化数据库结构
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(&setDictType, NULL);
server.db[j].watched_keys = dictCreate(&keylistDictType, NULL);
server.db[j].eviction_pool = evictionPoolAlloc();
server.db[j].id = j;
server.db[j].avg_ttl = 0;
}

db的数据结构如下:

typedef struct redisDb {

    // 数据库键空间,保存着数据库中的所有键值对
dict *dict; /* The keyspace for this DB */ // 键的过期时间,字典的键为键,字典的值为过期事件 UNIX 时间戳
dict *expires; /* Timeout of keys with a timeout set */ // 正处于阻塞状态的键
dict *blocking_keys; /* Keys with clients waiting for data (BLPOP) */ // 可以解除阻塞的键
dict *ready_keys; /* Blocked keys that received a PUSH */ // 正在被 WATCH 命令监视的键
dict *watched_keys; /* WATCHED keys for MULTI/EXEC CAS */ struct evictionPoolEntry *eviction_pool; /* Eviction pool of keys */ // 数据库号码
int id; /* Database ID */ // 数据库的键的平均 TTL ,统计信息
long long avg_ttl; /* Average TTL, just for stats */ } redisDb;

这里可以看到,设置了过期时间的key,除了会在 dict 属性存储,还会新增一条记录到 expires 字典。

expires字典的key:执行键的指针;value:过期时间。

创建pub/sub相关数据结构并初始化

    // 创建 PUBSUB 相关结构
server.pubsub_channels = dictCreate(&keylistDictType, NULL);
server.pubsub_patterns = listCreate();

初始化部分统计属性

	// serverCron() 函数的运行次数计数器
server.cronloops = 0;
// 负责执行 BGSAVE 的子进程的 ID
server.rdb_child_pid = -1;
// 负责进行 AOF 重写的子进程 ID
server.aof_child_pid = -1;
aofRewriteBufferReset();
// AOF 缓冲区
server.aof_buf = sdsempty();
// 最后一次完成 SAVE 的时间
server.lastsave = time(NULL); /* At startup we consider the DB saved. */
// 最后一次尝试执行 BGSAVE 的时间
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.resident_set_size = 0;
// 最后一次执行 SAVE 的状态
server.lastbgsave_status = REDIS_OK;
server.aof_last_write_status = REDIS_OK;
server.aof_last_write_errno = 0;
server.repl_good_slaves_count = 0;
updateCachedTime();

设置时间事件对应的函数指针

    /* Create the serverCron() time event, that's our main way to process
* background operations. */
// 为 serverCron() 创建时间事件
if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
redisPanic("Can't create the serverCron time event.");
exit(1);
}

这里的serverCron就是一个函数,后续每次周期触发时间事件时,就会运行这个serverCron。

可以看这里的英文注释,作者也提到,这是主要的处理后台任务的方式。

这块以后也会重点分析。

设置connect事件对应的连接处理器

aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler, NULL)

这里的acceptTcpHandler就是处理新连接的函数:

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd, max = MAX_ACCEPTS_PER_CALL;
char cip[REDIS_IP_STR_LEN];
REDIS_NOTUSED(el);
REDIS_NOTUSED(mask);
REDIS_NOTUSED(privdata); while (max--) {
// accept 客户端连接
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
if (cfd == ANET_ERR) {
if (errno != EWOULDBLOCK)
redisLog(REDIS_WARNING,
"Accepting client connection: %s", server.neterr);
return;
}
// 为客户端创建客户端状态(redisClient)
acceptCommonHandler(cfd, 0);
}
}

创建aof文件

如果aof打开了,就需要创建aof文件。

    if (server.aof_state == REDIS_AOF_ON) {
server.aof_fd = open(server.aof_filename,
O_WRONLY | O_APPEND | O_CREAT, 0644);
}

剩下的几个,暂时不涉及的任务

    // 如果服务器以 cluster 模式打开,那么初始化 cluster
if (server.cluster_enabled) clusterInit(); // 初始化复制功能有关的脚本缓存
replicationScriptCacheInit(); // 初始化脚本系统
scriptingInit(); // 初始化慢查询功能
slowlogInit(); // 初始化 BIO 系统
bioInit();

上面的几个,我们暂时还讲解不到,先看看就行。

到此,初始化redis server,就基本结束了。

总结

本讲内容较多,主要是redis启动过程中,要做的事,也太多了。希望我已经大致讲清楚了,其中,连接处理器那些都只是大致讲了,后面会继续。谢谢大家。

上一篇:使用visual studio 2012 编译opencv2.4.9


下一篇:Redis源码阅读(五)集群-故障迁移(上)