《redis设计与实现》2-数据库实现篇

上一篇文章介绍了 redis 基本的数据结构和对象 《redis设计与实现》1-数据结构与对象篇

本文主要关于：

• redis数据库实现的介绍
• 前面介绍的各种数据，在redis服务器中的内存模型是什么样的的。
• RDB文件将这些内存数据持久化后的格式是什么样的
• RDB和AOF序列化的区别是什么
• redis提供什么机制保障AOF文件不会一直增长
• RDB文件转储成json文件和内存分析 工具 介绍
• 客户端和服务端数据结构介绍

数据库

服务器的数据库

• redis是内存型数据库，所有数据都放在内存中
• 保存这些数据的是redisServer这个结构体，源码中该结构体包括大概300多行的代码。具体参考server.h/redisServer
• 和数据库相关的两个属性是：
  • int类型的dbnum：表示数据库数量，默认16个
  • redisDb指针类型的db：数据库对象数组

    

数据库对象

所在文件为server.h。数据库中所有针对键值对的增删改查，都是对dict做操作

typedef struct redisDb {
    dict *dict;                 /* The keyspace for this DB  */
    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 */
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */
    int id;                     /* Database ID */
    long long avg_ttl;          /* Average TTL, just for stats */
} redisDb;
复制代码

• dict：保存了该数据库中所有的键值对，键都是字符串，值可以是多种类型
• expires：保存了该数据中所有设置了过期时间的key
• blocking_keys：保存了客户端阻塞的
• watched_keys：保存被watch的命令
• id：保存数据库索引
• avg_ttl

客户端切换数据库

• 客户端通过select dbnum 命令切换选中的数据库
• 客户端的信息保存在client这个数据结构中，参考server.h/client
• client的类型为redisDb的db指针指向目前所选择的数据库

读写键空间时的其他操作

读写键空间时，是针对dict做操作，但是除了完成基本的增改查找操作，还会执行一些额外的维护操作，包括：

• 读写键时，会根据是否命中，更新hit和miss次数。

  相关命令：info stats keyspace_hits, info stats keyspace_misses

• 读取键后，会更新键的LRU时间，前面章节介绍过该字段
• 读取时，如果发现键已经过期，会先删除该键，然后才执行其他操作
• 如果watch监视了某个键，修改时会标记该键为脏（dirty）
• 每修改一个键，会对脏键计数器加1，触发持久化和复制操作
• 如果开启通知功能，修改键会下发通知

设置过期时间

• expire key ttl：设置生存时间为ttl秒
• pexpire key ttl：设置生存时间为ttl毫秒
• expireat key timestamp：设置过期时间为timstamp的秒数时间戳
• pexpireat key timestamp：过期时间为毫秒时间戳
• persist key：解除过期时间
• ttl key：获取剩余生存时间

保存过期时间

过期时间保存在expires的字典中，值为long类型的毫秒时间戳

过期键删除策略

各种删除策略的对比

策略类型	描述	优点	缺点	redis是否采用
定时删除	通过定时器实现	保证过期键能尽快释放	对cpu不友好，影响相应时间和吞吐量	否
惰性删除	放任不管，查询时才去检查	对cpu友好	没有被访问的永远不会被释放，相当于内存泄露	是
定期删除	每隔一段时间检查	综合前面的优点	难于确定执行时长和频率	是

redis使用的过期键删除策略

redis采用了惰性删除和定期删除策略

惰性删除的实现

• 由db.c中的expireIfNeeded实现
• 每次执行redis命令前都会调用该函数对输入键做检查

定期删除的实现

• server.c中的serverCron函数执行定时任务
• 函数每次运行时，都从一定数量的数据库中取出一定数量的键进行检查，并删除过期键

数据库通知

• 键空间通知：客户端获取数据库中键执行了什么命令。实现代码为notify.c文件的notifyKeyspaceEvent函数

  subscribe __keyspace@0__:keyname
  复制代码

• 键事件通知：某个命令被什么键执行了

  subscribe __keyevent@0__:del
  复制代码

RDB持久化

• redis是内存数据库，为了避免服务器进程异常导致数据丢失，redis提供了RDB持久化功能
• 持久化后的RDB文件是一个经过压缩的二进制文件

RDB文件的创建与载入

生成rdb文件的两个命令如下，实现函数为rdb.c文件的rdbSave函数：

• SAVE：阻塞redis服务器进程，知道RDB创建完成。阻塞期间不能处理其他请求
• BGSAVE：派生出子进程，子进程负责创建RDB文件，父进程继续处理请求

RDB文件的载入是在服务器启动时自动执行的，实现函数为rdb.c文件的rdbload函数。载入期间服务器一直处于阻塞状态

自动间隔保存

redis允许用户通过设置服务器配置的server选项，让服务器每隔一段时间（100ms）自动执行BGSAVE命令（serverCron函数）

//server.c中main函数内部创建定时器，serverCron为定时任务回调函数
if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
    serverPanic("Can't create event loop timers.");
    exit(1);
}
复制代码

配置参数

// 任意一个配置满足即执行
save 900 1 // 900s内，对服务器进行至少1次修改
save 300 10 // 300s内，对服务器至少修改10次
复制代码

数据结构

// 服务器全局变量，前面介绍过
struct redisServer {
    ...
     /* RDB persistence */
    // 上一次执行save或bgsave后，对数据库进行了多少次修改
    long long dirty;                /* Changes to DB from the last save */
    long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */
    pid_t rdb_child_pid;            /* PID of RDB saving child */
    struct saveparam *saveparams;   /* Save points array for RDB */
    int saveparamslen;              /* Number of saving points */
    char *rdb_filename;             /* Name of RDB file */
    int rdb_compression;            /* Use compression in RDB? */
    int rdb_checksum;               /* Use RDB checksum? */
    // 上一次成功执行save或bgsave的时间
    time_t lastsave;                /* Unix time of last successful save */
    time_t lastbgsave_try;          /* Unix time of last attempted bgsave */
    time_t rdb_save_time_last;      /* Time used by last RDB save run. */
    time_t rdb_save_time_start;     /* Current RDB save start time. */
    int rdb_bgsave_scheduled;       /* BGSAVE when possible if true. */
    int rdb_child_type;             /* Type of save by active child. */
    int lastbgsave_status;          /* C_OK or C_ERR */
    int stop_writes_on_bgsave_err;  /* Don't allow writes if can't BGSAVE */
    int rdb_pipe_write_result_to_parent; /* RDB pipes used to return the state */
    int rdb_pipe_read_result_from_child; /* of each slave in diskless SYNC. */
    ...
};
// 具体每一个参数对应的变量
struct saveparam {
    time_t seconds;
    int changes;
};
复制代码

RDB文件结构

概览

• 头五个字符为‘redis’常量，标识这个rdb文件是redis文件
• dv_version：4字节，标识了rdb文件的版本号
• databases：数据库文件内容
• EOF：常量，1字节，标识文件正文结束
• check_sum：8字节无符号整形，保存校验和，判定文件是否有损坏

dababases部分

每个database的内容：

• SELECTDB：常量，1字节。标识了后面的字节为数据库号码
• db_number：数据库号码
• key_value_pairs：数据库的键值对，如果有过期时间，也放在一起。

key_value_pairs部分

不带过期时间的键值对

type为value的类型，1字节，代表对象类型或底层编码，根据type决定如何读取value

带过期时间的键值对

• EXPIRETIME：常量，1字节，表示接下来要读入的是一个以毫秒为单位的过期时间
• ms：8字节长的无符号整形，过期时间

value的编码

每个value保存一个值对象，与type对应。type不同，value的结构，长度也有所不同

字符串对象

• type为REDIS_RDB_TYPE_STRING, value为字符串对象，而字符串对象本身又包含对象的编码和内容
• 如果编码为整数类型，编码后面直接保存整数值

  

• 如果编码为字符串类型，分为压缩和不压缩
  • 如果字符串长度<=20字节，不压缩

    

  • 如果字符串长度>20字节，压缩保存

    

    • REDIS_RDB_ENC_LZF:常量，标识字符串被lzf算法压缩过
    • compressed_len：被压缩后的长度
    • origin_len：字符串原始长度
    • compressed_string：压缩后的内容

列表对象

• type为REDIS_RDB_TYPE_LIST, value为列表对象

  

• list_length：记录列表的长度
• item：以字符串对象来处理

集合对象

• typw为REDIS_RDB_TYPE_SET，value为集合对象

  

• set_size： 集合大小
• elem：以字符串对象来处理

哈希对象

• type为REDIS_RDB_TYPE_HASH, value为哈希对象

  

• hash_size：哈希对象大小
• key-value都以字符串对象处理

有序集合对象

• type为REDIS_RDB_TYPE_ZSET，value为有序集合对象

  

intset编码集合

• type为REDIS_RDB_TYPE_SET_INTSET, value为整数集合对象
• 先将结合转换为字符串对象，然后保存。读入时，将字符串对象转为整数集合对象

ziplist编码的对象（包括列表，哈希，有序集合）

• type为REDIS_RDB_TYPE_LIST_ZIPLIST, REDIS_RDB_TYPE_HASH_ZIPLIST, REDIS_RDB_TYPE_ZSET_ZIPLIST
• 先将压缩列表转换为字符串对象，保存到rdb文件
• 读取时根据type类型，读入字符串，转换为压缩列表对象

分析RDB文件

使用 linux 自带的od命令

使用linux自带的od命令可以查看rdb文件信息，比如od -c dump.rdb,以Ascii打印，下图显示 docker 创建的redis中，空的rdb文件输出的内容

工具

• rdb转json的工具
• 图形界面rdbtool
• rdr内存分析工具

AOF持久化

AOF写入与同步

除了RDB持久化外，redis还提供了AOF持久化功能。区别如下：

• RDB通过保存数据库中键值对记录数据库状态
• AOF通过保存服务器执行的写命令来记录数据库状态

AOF持久化分为三步：

• 命令追加：命令append到redisServer全局变量的aof_buf成员中
• 文件写入：
• 文件同步

事件结束时调用flushAppendOnlyFile函数，考虑是否将aof_buf内容写到AOF文件里（参数决定）

• always：所有内容写入并同步到AOF文件（写入的是缓冲区，同步时从缓冲区刷到磁盘）
• everysec：默认值。写入AOF文件，如果上次同步时间距现在草稿1s，同步AOF。
• no：只写入AOF文件，由系统决定何时同步

AOF载入与还原

服务器只需要读入并执行一遍AOF命令即可还原数据库状态，读取的步骤如下：

• 创建一个不带网络连接的伪客户端：因为命令只能在客户端执行
• 从AOF读取一条写命令
• 使用客户端执行该命令
• 重复上面的步骤，直到完成

AOF重写

• 随着时间流逝，AOF文件内容会越来越大，影响redis性能。redis提供重写功能解决该问题。
• 重写是通过读取redis当前数据状态完成的，而不是解析AOF文件
• 为了不影响redis正常响应，重写功能通过创建子进程（注意不是线程）完成
• 为了解决父子进程数据不一致问题（父进程接收新的请求），redis设置了 AOF重写缓冲区 。新的命令在AOF缓冲区和AOF重写缓冲区中双写。

事件

redis是一个事件驱动程序，事件包括两大类：

• 文件事件：socket通讯完成一系列操作
• 时间事件：某些需要在给定时间执行的操作

文件事件

• redis基于Reactor模式开发事件处理器，使用IO多路复用监听套接字。关于IO多路复用可参考之前的文章五种io模型对比
• ，虽然事件处理器以单线程运行，通过io多路复用，能同时监听多个套接字实现高性能

事件处理器的构成

• 文件事件：套接字操作的抽象
• io多路复用程序：同时监听多个套接字，并向事件分派器传送事件。多个套接字按队列排序
• 文件事件分派器：接收套接字，根据事件类型调用相应的事件处理器
• 事件处理器：不同的函数实现不同的事件

IO多路复用的实现

可选的io多路复用包括select,epoll,evport,kqueue实现。每种实现都放在单独的文件中。编译时根据不同的宏切换不同的实现

事件类型

#define AE_NONE 0       /* No events registered. */
#define AE_READABLE 1   /* Fire when descriptor is readable. */
#define AE_WRITABLE 2   /* Fire when descriptor is writable. */
复制代码

处理器

redis为文件事件编写了多个处理器，分别用于实现不同的网络需求，在networking.c文件中，包括：

• 连接应答处理器：监听套接字，接收客户端命令请求。对应函数为acceptTcpHandler。内部调用socket编程的accpt函数
• 命令请求处理器：负责读入套接字中的命令请求内容。对应函数为readQueryFromClient。内部调用socket编程的read函数
• 命令回复处理器：负责将回复通过套接字返回给客户。对应函数为sendReplyToClient。内部调用socket班车的write函数

时间事件

分类

时间事件分类以下两大类，取决于时间处理器的返回值：

• 定时事件：返回AE_NOMORE（-1）
• 周期性事件：非AE_NOMORE值。单机版只有serverCron一个周期性事件

属性

时间事件包括三个属性：

• id：服务器创建的全局唯一标识
• when：事件到达时间
• timeProc：处理器，一个函数

实现

• 所有时间事件放在一个无序链表中

  

• 执行时需要遍历链表
• ae.c/aeCreateTimeEvent：创建时间处理器
• aeSearchNearestTimer：返回距离当前时间最近的时间事件
• ae.c/processTimeEvents：遍历时间处理器并执行

事件调度

• 事件调度和执行由ae.c/aeProcessEvents函数负责
• 该函数被放在ae.c/aeMain函数中的一段循环里面，不断执行直到服务器关闭
• aeMain被server.c的main函数调用

int main() {
    ...
    aeMain(server.el);
    ...
}
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);
    }
}

复制代码

客户端

redis服务器为每个连接的客户端建立了一个redisClient的结构，保存客户端状态信息。所有客户端的信息放在一个链表里。可通过client list命令查看

struct redisServer {
    ...
    list *clients;
    ...
}
复制代码

客户端数据结构如下：

typedef struct client {
    uint64_t id;            /* Client incremental unique ID. */
    //客户端套接字描述符，伪客户端该值为-1（包括AOF还原和执行 Lua 脚本的命令）
    int fd;                 /* Client socket. */
    redisDb *db;            /* Pointer to currently SELECTed DB. */
    // 客户端名字，默认为空，可通过client setname设置
    robj *name;             /* As set by CLIENT SETNAME. */
    // 输入缓冲区，保存客户端发送的命令请求，不能超过1G
    sds querybuf;           /* Buffer we use to accumulate client queries. */
    size_t qb_pos;          /* The position we have read in querybuf. */
    sds pending_querybuf;   /* If this client is flagged as master, this buffer
                               represents the yet not applied portion of the
                               replication stream that we are receiving from
                               the master. */
    size_t querybuf_peak;   /* Recent (100ms or more) peak of querybuf size. */
    // 解析querybuf，得到参数个数
    int argc;               /* Num of arguments of current command. */
    // 解析querybuf，得到参数值
    robj **argv;            /* Arguments of current command. */
    // 根据前面的argv[0]， 找到这个命令对应的处理函数
    struct redisCommand *cmd, *lastcmd;  /* Last command executed. */
    int reqtype;            /* Request protocol type: PROTO_REQ_* */
    int multibulklen;       /* Number of multi bulk arguments left to read. */
    long bulklen;           /* Length of bulk argument in multi bulk request. */
    // 服务器返回给客户端的可被空间，固定buff用完时才会使用
    list *reply;            /* List of reply objects to send to the client. */
    unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
    size_t 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;
    // 客户端角色和状态：REDIS_MASTER, REDIS_SLAVE, REDIS_LUA_CLIENT等
    int flags;              /* Client flags: CLIENT_* macros. */
    // 客户端是否通过身份验证的标识
    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 read_reploff; /* Read replication offset if this is a master. */
    long long reploff;      /* Applied replication offset if this is a 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. */
    long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
                                       copying this slave output buffer
                                       should use. */
    char replid[CONFIG_RUN_ID_SIZE+1]; /* Master replication ID (if master). */
    int slave_listening_port; /* As configured with: SLAVECONF listening-port */
    char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */
    int slave_capa;         /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
    multiState mstate;      /* MULTI/EXEC state */
    int btype;              /* Type of blocking op if CLIENT_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. */
    listNode *client_list_node; /* list node in client list */

    /* Response buffer */
    // 记录buf数组目前使用的字节数
    int bufpos;
    // (16*1024)=16k,服务器返回给客户端的内容缓冲区。固定大小，存储一下固定返回值（如‘ok’）
    char buf[PROTO_REPLY_CHUNK_BYTES];
} client;
复制代码

服务器

服务器记录了redis服务器所有的信息，包括前面介绍的一些，罗列主要的如下：

struct redisServer {
    ...
    // 所有数据信息
    redisDb *db;
    // 所有客户端信息
    list *clients;
    
     /* time cache */
    // 系统当前unix时间戳,秒
    time_t unixtime;    /* Unix time sampled every cron cycle. */
    time_t timezone;    /* Cached timezone. As set by tzset(). */
    int daylight_active;    /* Currently in daylight saving time. */
    // 系统当前unix时间戳,毫秒
    long long mstime;   /* Like 'unixtime' but with milliseconds resolution. */
    
    // 默认没10s更新一次的时钟缓存，用于计算键idle时长
    unsigned int lruclock;      /* Clock for LRU eviction */
    
    // 抽样相关的参数
    struct {
        // 上次抽样时间
        long long last_sample_time; /* Timestamp of last sample in ms */
        // 上次抽样时，服务器已经执行的命令数
        long long last_sample_count;/* Count in last sample */
        // 抽样结果
        long long samples[STATS_METRIC_SAMPLES];
        int idx;
    } inst_metric[STATS_METRIC_COUNT];
    
    // 内存峰值
    size_t stat_peak_memory;        /* Max used memory record */
    // 关闭服务器的标识
    int shutdown_asap;          /* SHUTDOWN needed ASAP */
    // bgsave命令子进程的id
    pid_t rdb_child_pid;            /* PID of RDB saving child */
    // bgrewriteaof子进程id
    pid_t aof_child_pid;            /* PID if rewriting process */
    // serverCron执行次数
    int cronloops;              /* Number of times the cron function run */
    ...
}
复制代码

参考

• 《redis设计与实现》
• rdbtool wiki
• rdr分析工具