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

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;

//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;
};
复制代码

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

    - REDIS\_RDB\_ENC\_LZF:常量，标识字符串被lzf算法压缩过
    - compressed\_len：被压缩后的长度
    - origin\_len：字符串原始长度
    - compressed\_string：压缩后的内容

#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. */
复制代码

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);
    }
}

复制代码

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;
复制代码

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 */
    ...
}
复制代码