渐进式解析 Redis 源码 - REDIS 服务器

2018-10-08 阅读量

千里之行，始于足下

为了让大家对 Redis 服务有个比较直观的认知，在底层结构的分享前我们会先从 服务器(redis-server) 和 客户端(redis-cli) 两个方面的概况进行简单展示，至于流程中的一些细节可能不甚了了，放心，后续我们基本都会针对性的进行详解。

这篇主要对 服务器(redis-server) 端进行讲解…

流

用户角度

即 用户操作执行流程，以 SET KEY VALUE 操作命令为例，那么从用户发送操作命令到响应主要经历如下流程：

发送命令请求：客户端接收到用户输入命令，会将命令明文转换成 Redis 协议格式，然后通过 socket 将协议内容发送给服务端，协议格式如 *3\r\n$3\r\nSET\r\n$3\r\nKEY\r\n$5\r\nVALUE\r\n (其实也可以发现一个小规律，每个参数前都有一个 $n 代表该参数的长度，然后 \r\n 分割)
读取命令请求：
- 读取 socket 中协议格式的命令请求，将其保存至该客户端状态的输入缓存区里
- 对输入缓冲区的命令进行解析，提取出命令请求中包含的命令参数，以及命令参数的个数，分别保存在该客户端状态的 argv / argc 中
- 调用命令执行器，执行指定命令
命令执行器：
- 根据 argv[0] 参数，在服务器启动过程中构建的命令列表（command table）中查找参数指定的命令。命令表为一个字典，键为命令名称（set、get、del 等），值为一个 redisCommand 结构体，每个 redisCommand 结构体都记录着该命令的详细实现信息，其中就包括执行函数的指针（redisCommandProc），比如 set 对应的执行函数就是 setCommand，get 对应 getCommand
- 预执行，现在我们已经有了 argc、argv、执行主体函数，但是我们还需要对一些数据进行预验证，比如命令和参数的对等性、身份验证是否通过、内存情况等等，当然如果在复制状态以及集群状态，预执行的操作将会更多
- 调用命令的执行函数，c->cmd->proc(c)，调用执行逻辑后产生的响应会被保存至该客户端状态的输出缓冲区里面（buf 属性和 reply 属性）
- 执行扫尾工作，比如是否将这次执行记录进慢日志、calls 计数器增加、是否执行持久化操作，如果在复制状态下，还会有复制操作
客户端接收响应：
- 上面我们提到由执行函数将响应保存至客户端状态的输出缓冲区里，并为客户端的 socket 关联命令回复处理器，将缓冲区中的响应回复给客户端，当然响应内容是协议格式，比如 +OK\r\n
- 客户端接收到协议格式的响应，通过协议解析器把内容转换成 “明文”，显示给用户

服务角度

即 服务端启动流程，但是从服务端的角度来看，其启动流程为：

初始化全局状态
载入配置
创建 daemon 守护进程
初始化服务器功能模块
载入持久化数据
开始事件循环

源码解析

由于有些方法里的代码量比较大，我们这里按照典型的代码片段进行解析，同志们可以根据文章提示的代码位置和代码里面的关键词在源码中搜素，可能数据结构一些元素看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

初始化全局状态

/* server.c: line 3709 */
int main(int argc, char **argv) {
    struct timeval tv;
    int j;
// 一般在 使用 Hiredis 的时候会定义 REDIS_TEST 宏，Hiredis 为 Redis的极简约C客户端库 
#ifdef REDIS_TEST
    // ....省略了一部分代码
    
#endif
    // 初始化库
    /* We need to initialize our libraries, and the server configuration. */
#ifdef INIT_SETPROCTITLE_REPLACEMENT
    // 这个宏主要是根据操作系统定义的，判断条件 ((defined __linux && defined(__GLIBC__)) || defined __APPLE__)
    // 跟 nginx 里面的 spt_init函数类似，为了修改进程的信息而特定弄的。比如在 linux 环境下，且需要修改主进程的名称，由于主进程的名称就是 main 函数 argv[0] 的名称，所以你只需要修改argv指向的内存信息即可。但是这样会带来一个问题，就是因为除了argv数组的信息，每个进程的环境变量 environ 变量也存在那片内存空间里，而且和argv的内存是毗邻的，所以一旦你修改了 argv[0] 的内容却溢出了，那么后面的包括 environ 变量的信息也就 say goodbye 了，所以为了发生这种溢出，我们需要将这些信息移至别处再修改进程的名称，这也就是 spt_init 这个函数的作用
    spt_init(argc, argv);
#endif
    // 初始化字符串显示，有些字符串函数可以根据当前的本地化信息经行特殊的处理
    setlocale(LC_COLLATE,"");
    // 设置内存溢出错误信息 Out Of Memory allocating %zu bytes!
    zmalloc_set_oom_handler(redisOutOfMemoryHandler);
    // 设置随机函数种子
    srand(time(NULL)^getpid());
    // 当前时间
    gettimeofday(&tv,NULL);
    char hashseed[16];
    getRandomHexChars(hashseed,sizeof(hashseed));
    // 根据生成的 hashseed 随机种子设置 dict字典随机种子
    dictSetHashFunctionSeed((uint8_t*)hashseed);
    // 是否哨兵模式
    server.sentinel_mode = checkForSentinelMode(argc,argv);
    // 初始化 config
    initServerConfig();
    // 模块加载，Redis 各种数据处理模块都是通过 REGISTER_API 的方式注册的，包括 第三方开发的模块
    moduleInitModulesSystem();
    /* Store the executable path and arguments in a safe place in order
     * to be able to restart the server later. */
    // 执行路径
    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]);
    /* We need to init sentinel right now as parsing the configuration file
     * in sentinel mode will have the effect of populating the sentinel
     * data structures with master nodes to monitor. */
    // 如果是哨兵模式 则初始化哨兵
    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. */
    // rdb/aof 存储文件检测
    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;
        /* Handle special options --help and --version */
        // 获取版本
        if (strcmp(argv[1], "-v") == 0 ||
            strcmp(argv[1], "--version") == 0) version();
        // help
        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.\n");
                fprintf(stderr,"Example: ./redis-server --test-memory 4096\n\n");
                exit(1);
            }
        }
        /* First argument is the config file name? */
        // 判断第一个参数，如果没有 -，则是 配置文件
        if (argv[j][0] != '-' || argv[j][1] != '-') {
            configfile = argv[j];
            server.configfile = getAbsolutePath(configfile);
            /* Replace the config file in server.exec_argv with
             * its absoulte path. */
            zfree(server.exec_argv[j]);
            server.exec_argv[j] = zstrdup(server.configfile);
            j++;
        }
        /* All the other options are parsed and conceptually appended to the
         * configuration file. For instance --port 6380 will generate the
         * string "port 6380\n" to be parsed after the actual file name
         * is parsed, if any. */
        // 参数检查
        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,"\n");
                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();
        // 加载 config 文件重设 配置
        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");
    }
    // 判断是否守护进程模式，根据 传入参数 和 系统支持 两重判断，如果需要守护进程，则建立，守护进程详细讲解请查看：http://www.web-lovers.com/c-daemonize.html
    server.supervised = redisIsSupervised(server.supervised_mode);
    int background = server.daemonize && !server.supervised;
    if (background) daemonize();
    // 初始化服务端逻辑，比如初始化 部分数据结构、 信号处理、clients 列表、slaves 列表、共享对象、unixsocket、数据库、时间计数、脚本逻辑、慢日志 等
    initServer();
    // 如果有守护进程或者配置 pidfile，则建立 pid 文件
    if (background || server.pidfile) createPidFile();
    // 设置进程名称，之前的 spt_init 基本就是为了这里
    redisSetProcTitle(argv[0]);
    // 打印 logo
    redisAsciiArt();
    // 检查最大连接数
    checkTcpBacklogSettings();
    // 非哨兵模式
    if (!server.sentinel_mode) {
        /* Things not needed when running in Sentinel mode. */
        serverLog(LL_WARNING,"Server initialized");
    #ifdef __linux__
        // 内存设置警告，一般是根据服务器内核的配置数据进行一些警告
        linuxMemoryWarnings();
    #endif
        // 根据配置中的 loadmodule 加载模块，在处理配置文件时，我们无法直接加载模块，因为在加载模块之前必须完全初始化服务器。 该函数在出错时中止服务器，因为从缺少模块开始不被认为是理智的：客户端可能依赖于给定命令的存在，加载AOF也可能需要一些模块存在，如果这个实例是从属，它必须理解 来自master的命令
        moduleLoadFromQueue();
        // 从硬盘中加载数据，根据配置的开启状态判断是加载 AOF或者RDB 文件，默认是使用 RDB 模式加载 dump.rdb 文件
        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);
    // 处理事件，client 的监控运算就在这里处理
    aeMain(server.el);
    // 服务器关闭，停止事件循环
    aeDeleteEventLoop(server.el);
    return 0;
}

载入配置，initServerConfig 初始化 config

/* server.c: line 1342 */
void initServerConfig(void) {
    int j;
    // 互斥锁初始化
    pthread_mutex_init(&server.next_client_id_mutex,NULL);
    pthread_mutex_init(&server.lruclock_mutex,NULL);
    pthread_mutex_init(&server.unixtime_mutex,NULL);
    // 设置服务端实例 runid，主从、集群 等多机的情况下用来区分实例
    getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE);
    // runid 后面添加结尾字符
    server.runid[CONFIG_RUN_ID_SIZE] = '\0';
    // 初始化 replication id，主要是用来为后续 master-slave 模式分配唯一 id
    changeReplicationId();
    // 清除辅助 replication id，这种情况有可能发生在 在完全重新同步之后开始新的复制历史记录时
    clearReplicationId2();
    // 配置文件
    server.configfile = NULL;
    // 执行路径
    server.executable = NULL;
    // 设置默认服务器频率，这个频率在 定时任务以及循环因子 方面使用较多
    server.hz = CONFIG_DEFAULT_HZ;
    // 操作系统位数
    server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
    // 默认端口
    server.port = CONFIG_DEFAULT_SERVER_PORT;
    // 默认侦听事件数量，默认 511
    server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG;
    // 绑定个数
    server.bindaddr_count = 0;
    // unix socket套接字
    server.unixsocket = NULL;
    // socket 套接字文件权限
    server.unixsocketperm = CONFIG_DEFAULT_UNIX_SOCKET_PERM;
    // 应答 ip 端口 client 数量
    server.ipfd_count = 0;
    // 应答 socket client
    server.sofd = -1;
    // 保护模式，用在 networking 模块
    server.protected_mode = CONFIG_DEFAULT_PROTECTED_MODE;
    // 数据库个数
    server.dbnum = CONFIG_DEFAULT_DBNUM;
    // 日志等级
    server.verbosity = CONFIG_DEFAULT_VERBOSITY;
    // client timeout 时间
    server.maxidletime = CONFIG_DEFAULT_CLIENT_TIMEOUT;
    // keepalive 设置
    server.tcpkeepalive = CONFIG_DEFAULT_TCP_KEEPALIVE;
    // 是否禁用 debug
    server.active_expire_enabled = 1;
    // 启动碎片整理（一说为 定时逐出 过期 key，不过这里还需要详细探究）
    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;
    // 碎片整理占用 CPU 最小百分比
    server.active_defrag_cycle_min = CONFIG_DEFAULT_DEFRAG_CYCLE_MIN;
    // 碎片整理占用 CPU 最大百分比
    server.active_defrag_cycle_max = CONFIG_DEFAULT_DEFRAG_CYCLE_MAX;
    // 协议最大长度
    server.proto_max_bulk_len = CONFIG_DEFAULT_PROTO_MAX_BULK_LEN;
    // 客户端的查询缓存区最大长度
    server.client_max_querybuf_len = PROTO_MAX_QUERYBUF_LEN;
    // rdb 保存参数
    server.saveparams = NULL;
    // rdb 从磁盘加载数据
    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;
    // 操作系统守护进程，一般跟 daemonize 配合使用
    server.supervised = 0;
    server.supervised_mode = SUPERVISED_NONE;
    // aof 是否激活
    server.aof_state = AOF_OFF;
    // aof 异步模式
    server.aof_fsync = CONFIG_DEFAULT_AOF_FSYNC;
    // aof 是否延迟
    server.aof_no_fsync_on_rewrite = CONFIG_DEFAULT_AOF_NO_FSYNC_ON_REWRITE;
    // aof 增长比例
    server.aof_rewrite_perc = AOF_REWRITE_PERC;
    // aof 增长到一定程度的时候进行 日志重写
    server.aof_rewrite_min_size = AOF_REWRITE_MIN_SIZE;
    // aof 最近一次启动或者重写 的大小
    server.aof_rewrite_base_size = 0;
    // aof 异步重写终止
    server.aof_rewrite_scheduled = 0;
    // aof 最后一次同步时间
    server.aof_last_fsync = time(NULL);
    // aof 最后一次运行时间
    server.aof_rewrite_time_last = -1;
    // aof 这次的开始时间
    server.aof_rewrite_time_start = -1;
    // aof 状态
    server.aof_lastbgrewrite_status = C_OK;
    // aof 延迟
    server.aof_delayed_fsync = 0;
    // aof 句柄描述符
    server.aof_fd = -1;
    // aof 对应数据库
    server.aof_selected_db = -1; /* Make sure the first time will not match */
    // aof flush 延迟时间
    server.aof_flush_postponed_start = 0;
    // aof 增量重写
    server.aof_rewrite_incremental_fsync = CONFIG_DEFAULT_AOF_REWRITE_INCREMENTAL_FSYNC;
    // aof 意外停止
    server.aof_load_truncated = CONFIG_DEFAULT_AOF_LOAD_TRUNCATED;
    // aof 使用 rdb 前导
    server.aof_use_rdb_preamble = CONFIG_DEFAULT_AOF_USE_RDB_PREAMBLE;
    // pid 文件
    server.pidfile = NULL;
    // rdb 文件名称
    server.rdb_filename = zstrdup(CONFIG_DEFAULT_RDB_FILENAME);
    // aof 文件名称
    server.aof_filename = zstrdup(CONFIG_DEFAULT_AOF_FILENAME);
    // 密码验证
    server.requirepass = NULL;
    // 在 rdb 中使用压缩
    server.rdb_compression = CONFIG_DEFAULT_RDB_COMPRESSION;
    // rdb 校验
    server.rdb_checksum = CONFIG_DEFAULT_RDB_CHECKSUM;
    // 错误停止写入
    server.stop_writes_on_bgsave_err = CONFIG_DEFAULT_STOP_WRITES_ON_BGSAVE_ERROR;
    // 在 serverCron 中增量 rehash
    server.activerehashing = CONFIG_DEFAULT_ACTIVE_REHASHING;
    // 碎片整理运行状态
    server.active_defrag_running = 0;
    // sub/pub 事件相关配置
    server.notify_keyspace_events = 0;
    // 最大 client 数目
    server.maxclients = CONFIG_DEFAULT_MAX_CLIENTS;
    // 阻塞客户端个数
    server.bpop_blocked_clients = 0;
    // 最大内存
    server.maxmemory = CONFIG_DEFAULT_MAXMEMORY;
    // 最大内存策略，如果达到内存限制了，Redis如何选择删除key，对所有策略来说，如果Redis找不到合适的可以删除的key都会在写操作时返回一个错误
    // volatile-lru -> 根据LRU算法删除设置过期时间的key
    // allkeys-lru -> 根据LRU算法删除任何key
    // volatile-random -> 随机移除设置过过期时间的key
    // allkeys-random -> 随机移除任何key
    // volatile-ttl -> 移除即将过期的key(minor TTL)
    // noeviction -> 不移除任何key，只返回一个写错误
    server.maxmemory_policy = CONFIG_DEFAULT_MAXMEMORY_POLICY;
    // LRU和最小TTL算法的实现都不是很精确，但是很接近（为了省内存），所以你可以用样本量做检测。 例如：默认Redis会检查3个key然后取最旧的那个，你可以通过下面的配置指令来设置样本的个数。默认值为5，数字越大结果越精确但是会消耗更多CPU
    server.maxmemory_samples = CONFIG_DEFAULT_MAXMEMORY_SAMPLES;
    // lfu 因子参数
    server.lfu_log_factor = CONFIG_DEFAULT_LFU_LOG_FACTOR;
    // lfu 延迟时间
    server.lfu_decay_time = CONFIG_DEFAULT_LFU_DECAY_TIME;
    // ziplist 节点个数
    server.hash_max_ziplist_entries = OBJ_HASH_MAX_ZIPLIST_ENTRIES;
    // ziplist 数值最大长度
    server.hash_max_ziplist_value = OBJ_HASH_MAX_ZIPLIST_VALUE;
    // ziplist list最大长度
    server.list_max_ziplist_size = OBJ_LIST_MAX_ZIPLIST_SIZE;
    // list 压缩等级
    server.list_compress_depth = OBJ_LIST_COMPRESS_DEPTH;
    // set 最大节点个数
    server.set_max_intset_entries = OBJ_SET_MAX_INTSET_ENTRIES;
    // zset 最大 ziplist 节点
    server.zset_max_ziplist_entries = OBJ_ZSET_MAX_ZIPLIST_ENTRIES;
    // zset 最大节点值大小
    server.zset_max_ziplist_value = OBJ_ZSET_MAX_ZIPLIST_VALUE;
    // HyperLogLog 空间大小
    server.hll_sparse_max_bytes = CONFIG_DEFAULT_HLL_SPARSE_MAX_BYTES;
    // 是否快速关闭，当执行 shutdown 命令时或者产生 SIGINT或者SIGTERM信号，不是直接exit退出程序，而是设置 server.shutdown_asap=1，然后在时间事件中，调用 prepareForShutdown 函数，先准一些关闭前的操作，例如刷新 aof和rdb文件，关闭监听描述符等等,最后再关闭
    server.shutdown_asap = 0;
    // 集群开启
    server.cluster_enabled = 0;
    // 集群节点超时时间，超时以后就不可用了
    server.cluster_node_timeout = CLUSTER_DEFAULT_NODE_TIMEOUT;
    // 只有在之前 master 有其它指定数量的工作状态下的 slave 节点时，slave节点才能提升为master。默认为1（即该集群至少有3个节点，1 master＋2 slaves，master宕机，仍有另外1个slave的情况下其中1个slave可以提升），测试环境可设置为0，生成环境中至少设置为1
    server.cluster_migration_barrier = CLUSTER_DEFAULT_MIGRATION_BARRIER;
    // 如果数据太旧，集群中的不可用 master 的 slave 节点会避免成为备用master。如果slave和master失联时间超过 (cluster_node_timeout * cluster_slave_validity_factor) + repl_ping_slave_period 则不会被提升为master。如 cluster_node_timeout 为 15秒，cluster_slave_validity_factor 为10，默认default repl_ping_slave_period 为10 秒，失联时间超过 160秒 slave 就不会成为master。较大的 cluster_slave_validity_factor 值可能允许包含过旧数据的slave 成为 master，同时较小的值可能会阻止集群选举出新 master。为了达到最大限度的高可用性，可以设置为0，即 slave 不管和 master 失联多久都可以提升为 master
    server.cluster_slave_validity_factor = CLUSTER_DEFAULT_SLAVE_VALIDITY;
    // 默认情况下如果 Redis 集群检测到至少有1个 hash slot 不可用，集群将停止查询数据。如果所有 slot 恢复则集群自动恢复。如果需要集群部分可用情况下仍可提供查询服务，设置为no
    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);
    // Redis 4.0 将兼容 NAT 和 Docker。在某些部署环境下，比如采用Docker或其他容器部署，这时的IP地址是处理过的，或者端口是转发后的，这种环境下，Redis集群节点地址发现就会失败。为了使Redis集群能在这种复杂环境下工作，需要将地址信息配置为静态的，即每个节点都配置为他的公共地址（包括IP地址，端口号、集群消息总线端口）。节点的公共信息在消息总线的数据包中能够找到，以便其他节点能够正确映射节点的地址。如果此处不配置，则将使用正常的Redis集群进行自动检测。
注意：当重新映射的时候，总线端口可能不位于 端口号 + 10000的固定偏移量上，因此可以指定任何的端口和总线端口，如果不设置总线端口，通常使用10000的固定偏移量
    // 节点的公网IP地址
    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;
    // 为节点迁移缓存的TCP连接数
    server.migrate_cached_sockets = dictCreate(&migrateCacheDictType,NULL);
    // client id
    server.next_client_id = 1; /* Client IDs, start from 1 .*/
    // 事件进程单次进程块大小
    server.loading_process_events_interval_bytes = (1024*1024*2);
    // lazyfree 内存淘汰机制的 内存满逐出选项
    server.lazyfree_lazy_eviction = CONFIG_DEFAULT_LAZYFREE_LAZY_EVICTION;
    // lazyfree 内存淘汰机制的 过期key删除选项
    server.lazyfree_lazy_expire = CONFIG_DEFAULT_LAZYFREE_LAZY_EXPIRE;
    // lazyfree 内存淘汰机制的 内部删除选项
    server.lazyfree_lazy_server_del = CONFIG_DEFAULT_LAZYFREE_LAZY_SERVER_DEL;
    // 显示 Redis logo
    server.always_show_logo = CONFIG_DEFAULT_ALWAYS_SHOW_LOGO;
    // lua 脚本时间限制
    server.lua_time_limit = LUA_SCRIPT_TIME_LIMIT;
    // lru 锁
    unsigned int lruclock = getLRUClock();
    atomicSet(server.lruclock,lruclock);
    // 重置保存参数
    resetServerSaveParams();
    // 1小时保存一次
    appendServerSaveParams(60*60,1);  /* save after 1 hour and 1 change */
    // 5分钟保存100次
    appendServerSaveParams(300,100);  /* save after 5 minutes and 100 changes */
    // 1分钟保存10000次
    appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */
    /* Replication related */
    // master 密码
    server.masterauth = NULL;
    // master host
    server.masterhost = NULL;
    // master port
    server.masterport = 6379;
    // master 主机
    server.master = NULL;
    // 缓存的 master 主机信息
    server.cached_master = NULL;
    // 复制缓存总量
    server.master_initial_offset = -1;
    // 复制状态
    server.repl_state = REPL_STATE_NONE;
    // 异步 io 超时时间
    server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT;
    // 复制陈旧数据
    server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA;
    // slave 只读
    server.repl_slave_ro = CONFIG_DEFAULT_SLAVE_READ_ONLY;
    // 从服务器执行重新同步操作或者删除整个数据库以便加载RDB文件时的删除策略
    server.repl_slave_lazy_flush = CONFIG_DEFAULT_SLAVE_LAZY_FLUSH;
    // 复制开始
    server.repl_down_since = 0; /* Never connected, repl is down since EVER. */
    // 同步之后是否禁用从站（slave）上的TCP_NODELAY，可设置yes或者no。默认是no，即使用tcp nodelay。如果master设置了yes来禁止tcp nodelay设置，在把数据复制给slave的时候，会减少包的数量和更小的网络带宽。但是这也可能带来数据的延迟。默认我们推荐更小的延迟，但是在数据量传输很大的场景下，即在高负载情况下或者在主从都掉线的情况下建议选择yes
    server.repl_disable_tcp_nodelay = CONFIG_DEFAULT_REPL_DISABLE_TCP_NODELAY;
    // 是否使用socket方式复制数据。目前 Redis 复制提供两种方式，disk（硬盘备份）和socket（无硬盘备份）。如果新的slave连上来或者重连的slave无法部分同步，就会执行全量同步，master会生成rdb文件。有2种方式：disk方式是master创建一个新的进程把rdb文件保存到磁盘，再把磁盘上的rdb文件传递给slave。socket是master创建一个新的进程，直接把rdb文件以socket的方式发给slave。disk方式的时候，当一个rdb保存的过程中，多个slave都能共享这个rdb文件。socket的方式就的一个个slave顺序复制。在磁盘速度缓慢，网速快也就是硬盘低速而网络高速（高带宽）的情况下推荐用socket（无硬盘备份）方式
    server.repl_diskless_sync = CONFIG_DEFAULT_REPL_DISKLESS_SYNC;
    // 当启用无硬盘备份，服务器等待一段时间后才会通过套接字向从站（slave）传送RDB文件，这个等待时间是可配置的。这一点很重要，因为一旦传送开始，就不可能再为一个新到达的从站（slave）服务。从站（slave）则要排队等待下一次RDB传送。因此服务器等待一段时间以期更多的从站（slave）到达。延迟时间以秒为单位，默认为5秒。要关掉这一功能，只需将它设置为0秒，传送会立即启动
    server.repl_diskless_sync_delay = CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY;
    // slave根据指定的时间间隔向服务器发送ping请求。时间间隔可以通过repl_ping_slave_period 来设置，默认10秒
    server.repl_ping_slave_period = CONFIG_DEFAULT_REPL_PING_SLAVE_PERIOD;
    // 主从复制连接超时时间。master和slave都有超时时间的设置。master检测到slave上次发送的时间超过repl_timeout，即认为slave离线，清除该slave信息。slave检测到上次和master交互的时间超过repl_timeout，则认为master离线。需要注意的是repl_timeout需要设置一个比 repl_ping_slave_period 更大的值，否则每次主站和从站之间通信低速时都会被检测为超时
    server.repl_timeout = CONFIG_DEFAULT_REPL_TIMEOUT;
    // Redis 提供了可以让master停止写入的方式，如果配置了min-slaves-to-write N，健康的slave（在线的从服务器）的个数小于N，mater就禁止写入。master最少得有N个健康的slave存活才能执行写命令。这个配置虽然不能保证N个slave都一定能接收到master的写操作，但是能避免没有足够健康的slave（在线的从服务器）的时候，master不能写入来避免数据丢失。设置为0是关闭该功能，默认是0
    server.repl_min_slaves_to_write = CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE;
    // 配合 repl_min_slaves_to_write 使用，设置延时多少秒的 slave 才被认为是健康的 slave, 默认是10秒，延迟的秒数必须<=所定义的值，延迟秒数是从最后一次收到的来自从站的ping开始计算。ping通常是每秒一次
    server.repl_min_slaves_max_lag = CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG;
    // 当master不可用，Redis Sentinel（哨兵）会根据slave的优先级选举一个提升为master。最低的优先级的slave，当选master。而配置成0，永远不会被选举。默认优先级是100
    server.slave_priority = CONFIG_DEFAULT_SLAVE_PRIORITY;
    // slave_announce_ip 和 slave_announce_port 配置是 Redis 3.2.2加上的。在端口转发或地址转换（NAT）网络环境中，如果 Redis 和 Sentinel 做了端口映射，例如在 Docker 默认网络模式下，使用-p参数做端口映射，就需要配置一下从服务器（slave）redis.conf中的 slave_announce_ip 和 slave_announce_port，对应外网的IP和外网端口。Sentinel 的配置文件sentinel.conf 也需要配置 sentinel announce_ip 和 sentinel announce_port ，对应外网的IP和外网端口。当然，如果Docker配置成host网络模式，就不需要配置了
    server.slave_announce_ip = CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP;
    // 配合 slave_announce_ip 使用
    server.slave_announce_port = CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT;
    // 一个全局性的计数器。该属性只有存在积压队列的情况下才会增加计数。当存在积压队列时，每次收到客户端发来的，长度为 len 的请求命令时，就会将 master_repl_offset 增加 len
    server.master_repl_offset = 0;
    /* Replication partial resync backlog */
    // 复制积压缓冲，在2.8版本，Redis 使用了新的复制方式，引入了复制积压缓冲（replication backlog），而不只是直接把 所有命令立即 复制，当主服务器进行命令传播的时候，maser不仅将所有的数据更新命令发送到所有slave的replication buffer，还会写入replication backlog。当断开的slave重新连接上master的时候，slave将会发送psync命令（包含复制的偏移量offset），请求partial resync。如果请求的offset不存在，那么执行全量的sync操作，相当于重新建立主从复制
    server.repl_backlog = NULL;
    // 复制积压缓冲大小
    server.repl_backlog_size = CONFIG_DEFAULT_REPL_BACKLOG_SIZE;
    // 积压队列 repl_backlog 中，当前累积的数据量的大小。该值不会超过积压队列的总容量 repl_backlog_size
    server.repl_backlog_histlen = 0;
    // 在积压队列 repl_backlog 中，每次写入新数据时的起始索引，是一个相对于 repl_backlog 的索引。当 repl_backlog_idx 等于 repl_backlog 的长度repl_backlog_size 时，置其值为0，表示从头开始
    server.repl_backlog_idx = 0;
    // 在积压队列中，最早保存的命令的首字节，在全局范围内（而非积压队列内）的偏移量。在累积命令流时，下列等式恒成立：master_repl_offset - repl_backlog_off + 1 = repl_backlog_histlen
    server.repl_backlog_off = 0;
    // 值表示当主节点没有从节点连接时，积压队列最长的存活时间，该值默认为1个小时。因此，如果主节点当前已没有从节点连接，并且配置了 repl_backlog_time_limit 属性值，并且积压队列还存在的情况下，则判断属性repl_no_slaves_since 最近一次更新时间是否已经超过配置的repl_backlog_time_limit 属性值，若已超过，则调用freeReplicationBacklog 释放积压队列
    server.repl_backlog_time_limit = CONFIG_DEFAULT_REPL_BACKLOG_TIME_LIMIT;
    // 在 freeClient 函数中，每当释放了一个从节点客户端后，都会判断列表server.slaves 当前长度，如果其长度为0，说明该主节点已经没有连接的从节点了，因此就会设置属性 repl_no_slaves_since 为当时的时间戳
    server.repl_no_slaves_since = time(NULL);
    /* Client output buffer limits */
    // 客户端输出 buffer 长度
    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. */
    // 将 command 命令 挂到 server 属性中
    server.commands = dictCreate(&commandTableDictType,NULL);
    // 原始 command，因为 command 后续有可能被改名
    server.orig_commands = dictCreate(&commandTableDictType,NULL);
    // 构建 command 列表
    populateCommandTable();
    server.delCommand = lookupCommandByCString("del");
    server.multiCommand = lookupCommandByCString("multi");
    server.lpushCommand = lookupCommandByCString("lpush");
    server.lpopCommand = lookupCommandByCString("lpop");
    server.rpopCommand = lookupCommandByCString("rpop");
    server.sremCommand = lookupCommandByCString("srem");
    server.execCommand = lookupCommandByCString("exec");
    server.expireCommand = lookupCommandByCString("expire");
    server.pexpireCommand = lookupCommandByCString("pexpire");
    /* Slow log */
    // 执行时间比 slowlog_log_slower_than 大的请求记录到慢日志里面，单位是微秒，1000000就是1秒。注意，负数时间会禁用慢查询日志，而0则会强制记录所有命令
    server.slowlog_log_slower_than = CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN;
    // 慢查询日志所占的内存长度。当一个新的命令被写进日志的时候，最老的那个记录会被删掉。这个长度没有限制。只要有足够的内存就行。你可以通过 SLOWLOG RESET 来释放内存
    server.slowlog_max_len = CONFIG_DEFAULT_SLOWLOG_MAX_LEN;
    /* Latency monitor */
    // 延迟监控阈值（毫秒数），0的话，就是关闭监视。默认延迟监控功能是关闭的，因为如果您没有延迟问题，这个配置就不需要，并且它搜集数据对性能有影响，虽然影响非常小，但是负载大的话，效果就明显了。如果你需要打开，也可以通过CONFIG SET latency-monitor-threshold <milliseconds>命令动态设置
    server.latency_monitor_threshold = CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD;
    /* Debugging */
    // Debug 配置，包括 失败提示、提示文件、bug 提示开始、watchdog 周期
    server.assert_failed = "<no assertion failed>";
    server.assert_file = "<no file>";
    server.assert_line = 0;
    server.bug_report_start = 0;
    server.watchdog_period = 0;
}

初始化服务器功能模块

/* server.c: line 1812 */
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);
    }
    // 进程 id
    server.pid = getpid();
    server.current_client = NULL;
    // 初始化 clients 链表
    server.clients = listCreate();
    // 初始化 clients 链表
    server.clients_to_close = listCreate();
    // 初始化 从服务器链表
    server.slaves = listCreate();
    // 初始化 监控器 链表
    server.monitors = listCreate();
    // 初始化 clients 追加链表
    server.clients_pending_write = listCreate();
    // 初始化 从服务器选择的 db
    server.slaveseldb = -1; /* Force to emit the first SELECT command. */
    // 初始化 非阻塞客户端链表
    server.unblocked_clients = listCreate();
    // 初始化 解除阻塞的键链表
    server.ready_keys = listCreate();
    // 初始化 clients 等待应答链表
    server.clients_waiting_acks = listCreate();
    // 初始化 向所有从服务发送 ACK 响应的标识，一般发生在 客户端阻塞的时候
    server.get_ack_from_slaves = 0;
    // 初始化 clients 暂停状态
    server.clients_paused = 0;
    // 系统内存
    server.system_memory_size = zmalloc_get_memory_size();
    
    // 初始化 共享对象
    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);
    }
    // 申请 db 的内存空间
    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. */
    // 打开 unix 本地 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. */
    // 判断如果没有 ip 监听
    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. */
    // 创建初始化数据库结构
    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;
    }
    // 创建并初始化 LRU pool
    evictionPoolAlloc(); /* Initialize the LRU keys pool. */
    // 创建并初始化 PUB/SUB(发布/订阅) 相关结构
    server.pubsub_channels = dictCreate(&keylistDictType,NULL);
    server.pubsub_patterns = listCreate();
    // 绑定 free 方法
    listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
    // 绑定 match 方法
    listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
    // 初始化定时任务标识
    server.cronloops = 0;
    // rdb 和 aof 逻辑的进程标识
    server.rdb_child_pid = -1;
    server.aof_child_pid = -1;
    server.rdb_child_type = RDB_CHILD_TYPE_NONE;
    server.rdb_bgsave_scheduled = 0;
    // child 管道
    server.child_info_pipe[0] = -1;
    server.child_info_pipe[1] = -1;
    server.child_info_data.magic = 0;
    // aof 重写 buffer 初始化
    aofRewriteBufferReset();
    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. */
    // rdb 保存时间
    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.resident_set_size = 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;
    updateCachedTime();
    /* 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. */
    // 新增 serverCron 事件
    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. */
    // 为 TCP 连接新增 accept 响应事件，accept 应答处理器
    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.");
            }
    }
    // unix socket（unix 套接字）应答处理器
    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. */
    // 创建 client阻塞管道的应答事件
    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 功能已经启用，则打开或者创建一个 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. */
    // 如果是 32位 系统的话，限制最大 3G 内存
    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();
    // 脚本功能初始化
    scriptingInit(1);
    // 慢日志初始化
    slowlogInit();
    // 监控器初始化
    latencyMonitorInit();
    // bio 初始化
    bioInit();
    // 已占用内存
    server.initial_memory_usage = zmalloc_used_memory();
}

事件处理器主循环（重要）

/* ae.c: line 496 */
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);
    }
}

事件处理主体逻辑

返回处理事件数量
处理所有已到达的时间事件，以及所有已就绪的文件事件
如果不传入特殊 flags 的话，那么函数睡眠直到文件事件就绪，或者下个时间事件到达（如果有的话）
- 如果 flags 为 0 ，那么函数不作动作，直接返回
- 如果 flags 包含 AE_ALL_EVENTS ，所有类型的事件都会被处
- 如果 flags 包含 AE_FILE_EVENTS ，那么处理文件事件
- 如果 flags 包含 AE_TIME_EVENTS ，那么处理时间事
- 如果 flags 包含 AE_DONT_WAIT ，那么函数在处理完所有不许阻塞的事件之后，即刻返回

/* ae.c: line 358 */
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;
            // 计算距今最近的时间事件还要多久才能达到，并将该时间距保存在 tv 结构中
            aeGetTime(&now_sec, &now_ms);
            tvp = &tv;
            /* How many milliseconds we need to wait for the next
             * time event to fire? */
            // 需要阻塞时间，时间差小于 0 ，说明事件已经可以执行了，将秒和毫秒设为 0 （不阻塞）
            long long ms =
                (shortest->when_sec - now_sec)*1000 +
                shortest->when_ms - now_ms;
            if (ms > 0) {
                tvp->tv_sec = ms/1000;
                tvp->tv_usec = (ms % 1000)*1000;
            } else {
                tvp->tv_sec = 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 */
            // 执行到这一步，说明没有时间事件，那么根据 AE_DONT_WAIT 是否设置来决定是否阻塞，以及阻塞的时间长
            if (flags & AE_DONT_WAIT) {
                // 设置文件事件不阻塞
                tv.tv_sec = tv.tv_usec = 0;
                tvp = &tv;
            } else {
                /* Otherwise we can block */
                // 文件事件可以阻塞直到有事件到达为止
                tvp = NULL; /* wait forever */
            }
        }
        /* Call the multiplexing API, will return only on timeout or when
         * some event fires. */
        // 处理文件事件，阻塞时间由 tvp 决定
        numevents = aeApiPoll(eventLoop, tvp);
        /* After sleep callback. */
        if (eventLoop->aftersleep != NULL && flags & AE_CALL_AFTER_SLEEP)
            eventLoop->aftersleep(eventLoop);
        for (j = 0; j < numevents; j++) {
            // 从已就绪数组中获取事件
            aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
            int mask = eventLoop->fired[j].mask;
            int fd = eventLoop->fired[j].fd;
            int fired = 0; /* Number of events fired for current fd. */
            /* Normally we execute the readable event first, and the writable
             * event laster. This is useful as sometimes we may be able
             * to serve the reply of a query immediately after processing the
             * query.
             *
             * However if AE_BARRIER is set in the mask, our application is
             * asking us to do the reverse: never fire the writable event
             * after the readable. In such a case, we invert the calls.
             * This is useful when, for instance, we want to do things
             * in the beforeSleep() hook, like fsynching a file to disk,
             * before replying to a client. */
            int invert = fe->mask & AE_BARRIER;
        /* 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.
             *
             * Fire the readable event if the call sequence is not
             * inverted. */
            // 读事件
            if (!invert && fe->mask & mask & AE_READABLE) {
                // rfired 确保读/写事件只能执行其中一个
                fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                fired++;
            }
            /* Fire the writable event. */
            // 写事件
            if (fe->mask & mask & AE_WRITABLE) {
                if (!fired || fe->wfileProc != fe->rfileProc) {
                    fe->wfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
            }
            /* If we have to invert the call, fire the readable event now
             * after the writable one. */
            if (invert && fe->mask & mask & AE_READABLE) {
                if (!fired || fe->wfileProc != fe->rfileProc) {
                    fe->rfileProc(eventLoop,fd,fe->clientData,mask);
                    fired++;
                }
            }
            processed++;
        }
    }
    /* Check time events */
    // 执行时间事件
    if (flags & AE_TIME_EVENTS)
        processed += processTimeEvents(eventLoop);
    return processed; /* return the number of processed file/time events */
}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-server.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！