源码阅读方式建议：

VScode打开Redis源码，结合黄健宏老师的《Redis设计与实现》，先理解概念，再回归代码。 在阅读到数据库的具体实现时，建议在Linux系统下编译并运行Redis，对其的理解将更为直观。

源码阅读顺序：

1. 阅读数据结构的源码
2. 阅读内存编码数据结构实现
3. 阅读数据类型的实现
4. 阅读单机数据库相关代码
5. 阅读客户端和服务器相关代码
6. 阅读多机（Cluster）的实现代码

黄健宏老师的《Redis设计与实现》（第二版）

部分资料及图片摘抄自网上，并附有参考链接，侵删。

一、基本数据结构

动态字符串SDS

源码文件：sds.h 和 sds.c。

结构定义

// 类型别名，用于指向 sdshdr 的 buf 属性 
typedef char *sds; //保存字符串对象的结构 
struct sdshdr {     
    // buf 中已占用空间的长度     
    int len;     
    // buf 中剩余可用空间的长度     
    int free;     
    // 数据空间     
    char buf[]; 
};

其一、快速获取字符串长度。通过len属性可以在O(1)的时间复杂度下获取一个字符串的长度，而C字符串需要O（n）。

其二、杜绝缓冲区溢出。C字符串由于没有记录自身长度，很可能在一些操作（如strcat等）时，造成缓冲区溢出，见下图：

其三、**减少修改字符串时带来的内存重分配次数。对于一个包含了 N 个字符的C字符串来说，这个C字符串的底层实现总是一个N+1个字符的数组**（额外一个字符用于存空字符）。

通过未使用空间，SDS 实现了空间预分配和惰性空间释放两种优化策略。

其四、空间预分配策略。当API对SDS进行修改时，不仅会分配必要空间还会分配额外空间，具体策略如下：

其五、惰性空间释放。当SDS的API需要缩短SDS保存的字符串时， 程序并不立即使用内存重分配来回收缩短后多出来的字节，而是使用free属性将这些字节的数量记录起来，并等待将来使用。

其六、支持二进制。C字符串是以’\0’结尾的，这使得C字符串只能保存文本数据，而不能保存像图片、音频、视频、压缩文件这样的二进制数据。但是Redis需要、也具备这样的特点。

总结一下：

接口API

sdslen

/*  * 返回 sds 实际保存的字符串的长度  *  * T = O(1)  */ 
static inline size_t sdslen(const sds s) {     
    struct sdshdr *sh = (void*)(s-(sizeof(struct sdshdr)));     
    return sh->len; 
}

sdsclear惰性删除策略

//以O（1)的时间完成字符串的“清空” 
//只需要将终止符放在0即可 
void sdsclear(sds s) { 
    // 取出 sdshdr     
    struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));      
    // 重新计算属性     
    sh->free += sh->len;     
    sh->len = 0;      
    // 将结束符放到最前面（相当于惰性地删除 buf 中的内容）     
    sh->buf[0] = '\0'; 
}

sdsMakeRoomFor

sds sdsMakeRoomFor(sds s, size_t addlen) {     
    struct sdshdr *sh, *newsh;          
    // 获取 s 目前的空余空间长度     
    size_t free = sdsavail(s);     
    size_t len, newlen;          
    // s 目前的空余空间已经足够，无须再进行扩展，直接返回     
    if (free >= addlen) return s;          
    // 获取 s 目前已占用空间的长度     
    len = sdslen(s);     
    sh = (void*) (s-(sizeof(struct sdshdr)));          
    // s 最少需要的长度     
    newlen = (len+addlen);          
    // 根据新长度，为 s 分配新空间所需的大小     
    if (newlen < SDS_MAX_PREALLOC)     
        // 如果新长度小于 SDS_MAX_PREALLOC      
        // 那么为它分配两倍于所需长度的空间     
        newlen *= 2;     
    else     
        // 否则，分配长度为目前长度加上 SDS_MAX_PREALLOC     
        newlen += SDS_MAX_PREALLOC;     
    // T = O(N)     
    newsh = zrealloc(sh, sizeof(struct sdshdr)+newlen+1);          
    // 内存不足，分配失败，返回     
    if (newsh == NULL) return NULL;        
    // 更新 sds 的空余长度     
    newsh->free = newlen - len;      
    // 返回 sds     
    return newsh->buf; 
}

链表List

源码文件：adlist.h 和 adlist.c。

// 双端链表节点 
typedef struct listNode {     
    // 前置节点     
    struct listNode *prev;     
    // 后置节点     
    struct listNode *next;     
    // 节点的值     
    void *value; 
} listNode;

接口分析

listAddNodeHead：插入节点到头部

list *listAddNodeHead(list *list, void *value) {     
    listNode *node;     
    // 为节点分配内存     
    if ((node = zmalloc(sizeof(*node))) == NULL)     
        return NULL;     
    // 保存值指针     
    node->value = value;     
    // 添加节点到空链表     
    if (list->len == 0) {         
        list->head = list->tail = node;         
        node->prev = node->next = NULL;       
    }      
    // 添加节点到非空链表     
    else {         
        node->prev = NULL;         
        node->next = list->head;         
        list->head->prev = node;         
        list->head = node;     
    }        
    // 更新链表节点数     
    list->len++;     
    return list; 
}

listGetIterator：产生一个迭代器

/* 为给定链表创建一个迭代器，  
* 之后每次对这个迭代器调用 listNext 都返回被迭代到的链表节点  
* direction 参数决定了迭代器的迭代方向：  
*  AL_START_HEAD ：从表头向表尾迭代  
*  AL_START_TAIL ：从表尾想表头迭代  */ 
listIter *listGetIterator(list *list, int direction) {     
    // 为迭代器分配内存     
    listIter *iter;     
    if ((iter = zmalloc(sizeof(*iter))) == NULL) 
        return NULL;     
    // 根据迭代方向，设置迭代器的起始节点     
    if (direction == AL_START_HEAD)     
        iter->next = list->head;     
    else     
        iter->next = list->tail;        
    // 记录迭代方向     
    iter->direction = direction;      
    return iter; 
}

其中，

//双端链表迭代器 
typedef struct listIter {     
    // 当前迭代到的节点     
    listNode *next;     
    // 迭代的方向（前向还是后向）     
    int direction;
} listIter;

listSearchKey:在链表中查找指定值的节点，其实就是利用迭代器遍历

listNode *listSearchKey(list *list, void *key) {     
    listIter *iter;     
    listNode *node;          
    // 迭代整个链表     
    iter = listGetIterator(list, AL_START_HEAD);     
    while((node = listNext(iter)) != NULL) {         
        // 对比         
        if (list->match) {             
            if (list->match(node->value, key)) {                 
                listReleaseIterator(iter);                 
                // 找到                 
                return node;             
            }         
        } else {             
            if (key == node->value) {             
                listReleaseIterator(iter);             
                // 找到             
                return node;
            }            
        }     
    }      
    listReleaseIterator(iter);  
    // 未找到   
    return NULL; }

• 话说list迭代器有什么用？

字典Dict

源码文件dict.h 和 dict.c。

结构定义

// 字典定义 
typedef struct dict {     
// 类型特定函数     
// 这一指针指向的结构体中存储了hash表常用的函数指针     
    dictType *type;     
    // 私有数据：保存了需要传给那些类型特定函数的可选参数（见后文）     
    void *privdata;     
    // 哈希表     
    dictht ht[2];     
    // 该值表示rehash进行到的下标索引位置     
    // 当 rehash 不在进行时，值为 -1     
    // 开始时值为0     
    // 正在进行中时，值处于0到size之间     
    int rehashidx;          
    int iterators; 
    // 目前正在运行的安全迭代器的数量 
} dict;

//每个字典都使用两个哈希表，从而实现渐进式 rehash 。 
typedef struct dictht {     
    // 哈希表数组(存储的是指向节点指针数组的指针)     
    // 都使用指针的方式可以节省空间     
    dictEntry **table;     
    // 哈希表大小     
    unsigned long size;     
    // 哈希表大小掩码，用于计算索引值     
    // 总是等于 size - 1     
    unsigned long sizemask;     
    // 该哈希表已有节点的数量     
    unsigned long used;
} dictht;

hash表的节点（Entry）定义如下，Redis是采用开链法来处理hash冲突的：

//哈希表节点 
typedef struct dictEntry {     
    // 键    
    void *key;   
    // 值     
    union {       
        void *val;    
        uint64_t u64;   
        int64_t s64;   
    } v;     
    // 指向下个哈希表节点，形成链表  
    struct dictEntry *next;
} dictEntry;

PS:在redis的hash表结构中，如果出现hash冲突，新的key-value是在旧的key-value前面的，这么做也很好理解：如果插入到链表最后，那么还需要一个遍历链表的操作，O(N)的复杂度。

接口分析

dictAddRaw：新增键值对

dictEntry *dictAddRaw(dict *d, void *key)
{
    int index;
    dictEntry *entry;
    dictht *ht;
    // 如果条件允许的话，进行单步 rehash
    // T = O(1)
    if (dictIsRehashing(d)) _dictRehashStep(d); 
    // 计算键在哈希表中的索引值
    // 如果值为 -1 ，那么表示键已经存在
    if ((index = _dictKeyIndex(d, key)) == -1)
        return NULL;
    // 如果字典正在 rehash ，那么将新键添加到 1 号哈希表
    // 否则，将新键添加到 0 号哈希表
    ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0];
    // 为新节点分配空间
    entry = zmalloc(sizeof(*entry));
    // 将新节点插入到链表表头
    entry->next = ht->table[index];
    ht->table[index] = entry;
    // 更新哈希表已使用节点数量
    ht->used++;
    // 设置新节点的键
    // T = O(1)
    dictSetKey(d, entry, key);
    return entry;
}

为什么负载因子一个是1一个是5？

根据 BGSAVE 命令或 BGREWRITEAOF 命令是否正在执行，服务器执行扩展操作所需的负载因子并不相同。 这是因为在执行BGSAVE命令或BGREWRITEAOF 命令的过程中，Redis 需要创建当前服务器进程的子进程， 而大多数操作系统都采用写时复制（copy-on-write）技术来优化子进程的使用效率，所以在子进程存在期间，服务器会提高执行扩展操作所需的负载因子，从而尽可能地避免在子进程存在期间进行哈希表扩展操作，这可以避免不必要的内存写入操作，最大限度地节约内存。

// 指示字典是否启用 rehash 的标识 
static int dict_can_resize = 1; 
// 强制 rehash 的比率(强制不可被上面的标志所阻止) 
static unsigned int dict_force_resize_ratio = 5;

Rehash的具体操作见接下来的这个API。

int dictRehash(dict *d, int n) {
    // 只可以在 rehash 进行中时执行
    if (!dictIsRehashing(d)) return 0;
    // 进行 N 步迁移
    // T = O(N)
    while(n--) {
        dictEntry *de, *nextde;
        // 如果 0 号哈希表为空，那么表示 rehash 执行完毕
        // T = O(1)
        if (d->ht[0].used == 0) {
        // 释放 0 号哈希表
        zfree(d->ht[0].table);
        // 将原来的 1 号哈希表设置为新的 0 号哈希表
        d->ht[0] = d->ht[1];
        // 重置旧的 1 号哈希表
        _dictReset(&d->ht[1]);
        // 关闭 rehash 标识
        d->rehashidx = -1;
        // 返回 0 ，向调用者表示 rehash 已经完成
        return 0;
        }
    // 确保 rehashidx 没有越界
    assert(d->ht[0].size > (unsigned)d->rehashidx);
    // 略过数组中为空的索引，找到下一个非空索引
    // hash表的entry中第一个就是void* key，所以可以直接访问其是否为空
    // 来判断该出是否存在键值对
    while(d->ht[0].table[d->rehashidx] == NULL) d->rehashidx++;
    // 指向该索引的链表表头节点
    de = d->ht[0].table[d->rehashidx];
    // 将链表中的所有节点迁移到新哈希表
    // T = O(1)
    while(de) {
            unsigned int h;    
            // 保存下个节点的指针
            nextde = de->next;           
            /* Get the index in the new hash table */
            // 计算新哈希表的哈希值，以及节点插入的索引位置
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            // 插入节点到新哈希表
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            // 更新计数器
            d->ht[0].used--;
            d->ht[1].used++;            
            // 继续处理下个节点
            de = nextde;
            }
        // 将刚迁移完的哈希表索引的指针设为空
        d->ht[0].table[d->rehashidx] = NULL;
        // 更新 rehash 索引
        d->rehashidx++;
        }
    return 1;
}

• Redis的hash表采用渐进式分步hash的方法
• rehash的过程中字典的两个hash表同时存在，并且在迭代、更新、删除键的时候都需要考虑这两个hash表，值得注意的是：字典的删除（delete）、查找（find）、更新（update）等操作会在两个哈希表上进行，但插入只会在第二个表中，rehash完毕之后会重置第一个表。
• 在rehash时，如果扩展操作，那么ht[1]的大小为第一个大于等于 ht[0].used*2的2^n；如果收缩操作（负载因子小于0.1），那么ht[1]的大小为第一个大于等于ht[0].used的2^n。
• rehsh时，相关API会经常进行单步rehash，数据库操作主要是调用int dictRehashMilliseconds(dict *d, int ms)，即在指定时间内执行rehash，时间到了就返回已迭代到的index,这种方式能够尽可能利用好CPU的执行时间。

迭代器

安全迭代器

dictIterator *dictGetSafeIterator(dict *d) {
    dictIterator *i = dictGetIterator(d);
    // 设置安全迭代器标识
    i->safe = 1;
    return i;
}

非安全迭代器

dictIterator *dictGetIterator(dict *d)
{
    dictIterator *iter = zmalloc(sizeof(*iter));
    iter->d = d;
    iter->table = 0;
    iter->index = -1;
    iter->safe = 0;
    iter->entry = NULL;
    iter->nextEntry = NULL;
    return iter;
}

跳跃表skiplist

源码文件：t_zset.c 中所有以 zsl 开头的函数。

typedef struct zskiplist {
    // 表头节点和表尾节点
    struct zskiplistNode *header, *tail;
    // 表中节点的数量
    unsigned long length;
    // 表中层数最大的节点的层数
    int level;
} zskiplist;

typedef struct zskiplistNode {
    // 成员对象(存储的对象类型，见下)
    robj *obj;
    // 分值（按照这个大小，升序排列）
    double score;
    // 后退指针（用作链表的倒序遍历）
    struct zskiplistNode *backward;
    // 层
    struct zskiplistLevel {
        // 前进指针
        struct zskiplistNode *forward;
        // 跨度
        unsigned int span;
    } level[];
} zskiplistNode;

注意：在同一个跳跃表中，各个节点保存的成员对象必须是唯一的，但是多个节点保存的分值却可以是相同的：分值相同的节点将按照成员对象的大小（见后）来进行排序，成员对象较小的节点会排在前面(靠近表头的方向)，而成员对象较大的节点则会排在后面(靠近表尾的方向)。

typedef struct redisObject {
    // 类型(日后会说)
    unsigned type:4;
    // 编码(日后会说)
    unsigned encoding:4;
    // 对象最后一次被访问的时间(日后会说)
    unsigned lru:REDIS_LRU_BITS; /* lru time (relative to server.lruclock) */
    // 引用计数(日后会说)
    int refcount;
    // 指向实际值的指针
    void *ptr;
} robj;

注：成员对象的有关知识，参考Redis源码阅读（三）。

查找、删除、添加

• 添加节点

• 删除操作呢？

接口分析

#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */
#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */

创建新跳跃表如下：

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 分配空间
    zsl = zmalloc(sizeof(*zsl));
    // 设置高度和起始层数
    zsl->level = 1;
    zsl->length = 0;
    // 初始化表头节点
    // T = O(1)
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;
    // 设置表尾
    zsl->tail = NULL;
    return zsl;
}

创建新节点如下（请仔细思考源码的实现）：

/*
 * 创建一个成员为 obj ，分值为 score 的新节点，
 * 并将这个新节点插入到跳跃表 zsl 中。
 * 函数的返回值为新节点。
 * T_wrost = O(N^2), T_avg = O(N log N)
 */
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    redisAssert(!isnan(score));
    // 在各个层查找节点的插入位置
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        // 如果 i 不是 zsl->level-1 层
        // 那么 i 层的起始 rank 值为 i+1 层的 rank 值
        // 各个层的 rank 值一层层累积
        // 最终 rank[0] 的值加一就是新节点的前置节点的排位
        // rank[0] 会在后面成为计算 span 值和 rank 值的基础
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 沿着前进指针遍历跳跃表
        // T_wrost = O(N^2), T_avg = O(N log N)
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                // 比对分值
                (x->level[i].forward->score == score &&
                // 比对成员， T = O(N)
                compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
            // 记录沿途跨越了多少个节点
            rank[i] += x->level[i].span;
            // 移动至下一指针
            x = x->level[i].forward;
        }
        // 记录将要和新节点相连接的节点
        update[i] = x;
    }
     /* 
     // zslInsert() 的调用者会确保同分值且同成员的元素不会出现，
     * 所以这里不需要进一步进行检查，可以直接创建新元素。
     */
    // 获取一个随机值作为新节点的层数level
    // T = O(N)
    level = zslRandomLevel();
    // 如果新节点的层数比表中其他节点的层数都要大
    // 那么初始化表头节点中未使用的层，并将它们记录到 update 数组中
    // 将来也指向新节点
    if (level > zsl->level) {
        // 初始化未使用层
        // T = O(1)
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        // 更新表中节点最大层数
        zsl->level = level;
    }
    // 创建新节点
    x = zslCreateNode(level,score,obj);
    // 将前面记录的指针指向新节点，并做相应的设置
    // T = O(1)
    for (i = 0; i < level; i++) {
        // 设置新节点的 forward 指针
        x->level[i].forward = update[i]->level[i].forward;
        // 将沿途记录的各个节点的 forward 指针指向新节点
        update[i]->level[i].forward = x;
        /* update span covered by update[i] as x is inserted here */
        // 计算新节点跨越的节点数量
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 更新新节点插入之后，沿途节点的 span 值
        // 其中的 +1 计算的是新节点
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    /* increment span for untouched levels */
    // 未接触的节点的 span 值也需要增一，这些节点直接从表头指向新节点
    // T = O(1)
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 设置新节点的后退指针
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    // 跳跃表的节点计数增一
    zsl->length++;
    return x;
}

int zslRandomLevel(void) {
    int level = 1;
    //ZSKIPLIST为0.25
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;
    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

补充：有序集合

/*
 * 有序集合
 */
typedef struct zset {
    // 字典，键为成员，值为分值
    // 用于支持 O(1) 复杂度的按成员取分值操作
    dict *dict;
    // 跳跃表，按分值排序成员
    // 用于支持平均复杂度为 O(log N) 的按分值定位成员操作
    // 以及范围操作
    zskiplist *zsl;
} zset;

HyperLogLog

源码文件：hyperloglog.c。

什么是基数？ 比如数据集 {1, 3, 5, 7, 5, 7, 8}， 那么这个数据集的基数集为 {1, 3, 5 ,7, 8}, 基数(不重复元素)为5个。 基数估计就是在误差可接受的范围内，快速计算基数。

算法细节

密集存储

offset_bytes = (idx 6) / 8 offset_bits = (idx 6) % 8

稀疏存储

计数缓存

结构定义

typedef struct intset {
    // 编码方式
    uint32_t encoding;
    // 集合包含的元素数量
    uint32_t length;
    // 保存元素的数组
    int8_t contents[];
} intset;

//intset 的编码方式
#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))

升级操作

举个例子：

升级之后不可降级。

接口分析

/*  
 * 根据给定的编码方式 enc ，返回集合的底层数组在 pos 索引上的元素。
 * T = O(1)
 */
static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) {
    int64_t v64;
    int32_t v32;
    int16_t v16;
    // ((ENCODING*)is->contents) 首先将数组转换回被编码的类型
    // 然后 ((ENCODING*)is->contents)+pos 计算出元素在数组中的正确位置
    // 之后 memcpy(&vEnc, ..., sizeof(vEnc)) 再从数组中拷贝出正确数量的字节
    // 如果有需要的话， memrevEncifbe(&vEnc) 会对拷贝出的字节进行大小端转换
    // 最后将值返回
    if (enc == INTSET_ENC_INT64) {
        memcpy(&v64,((int64_t*)is->contents)+pos,sizeof(v64));
        memrev64ifbe(&v64);
        return v64;
    } else if (enc == INTSET_ENC_INT32) {
        memcpy(&v32,((int32_t*)is->contents)+pos,sizeof(v32));
        memrev32ifbe(&v32);
        return v32;
    } else {
        memcpy(&v16,((int16_t*)is->contents)+pos,sizeof(v16));
        memrev16ifbe(&v16);
        return v16;
    }
}
/* 
 * 根据集合的编码方式，返回底层数组在 pos 索引上的值
 */
static int64_t _intsetGet(intset *is, int pos) {
    return _intsetGetEncoded(is,pos,intrev32ifbe(is->encoding));
}

intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
    // 计算编码 value 所需的长度
    uint8_t valenc = _intsetValueEncoding(value);
    uint32_t pos;
    // 默认设置插入为成功
    if (success) *success = 1;
    // 如果 value 的编码比整数集合现在的编码要大
    // 那么表示 value 必然可以添加到整数集合中
    // 并且整数集合需要对自身进行升级，才能满足 value 所需的编码
    if (valenc > intrev32ifbe(is->encoding)) {
        return intsetUpgradeAndAdd(is,value);
    } else {
        // 运行到这里，表示整数集合现有的编码方式适用于 value
        // 在整数集合中查找 value ，看他是否存在：
        // - 如果存在，那么将 *success 设置为 0 ，并返回未经改动的整数集合
        // - 如果不存在，那么可以插入 value 的位置将被保存到 pos 指针中
        //   等待后续程序使用
        if (intsetSearch(is,value,&pos)) {
            if (success) *success = 0;
            return is;
        }
        // 运行到这里，表示 value 不存在于集合中
        // 程序需要将 value 添加到整数集合中
        // 为 value 在集合中分配空间
        is = intsetResize(is,intrev32ifbe(is->length)+1);
        // 如果新元素不是被添加到底层数组的末尾
        // 那么需要对现有元素的数据进行移动，空出 pos 上的位置，用于设置新值
        // 举个例子
        // 如果数组为：
        // | x | y | z | ? |
        //     |<----->|
        // 而新元素 n 的 pos 为 1 ，那么数组将移动 y 和 z 两个元素
        // | x | y | y | z |
        //         |<----->|
        // 这样就可以将新元素设置到 pos 上了：
        // | x | n | y | z |
        // T = O(N)
        if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
    }
    // 将新值设置到底层数组的指定位置中
    _intsetSet(is,pos,value);
    // 增一集合元素数量的计数器
    is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
    // 返回添加新元素后的整数集合
    return is;
}

那么怎么插入一个数据呢？

/* 
 * 根据集合的编码方式，将底层数组在 pos 位置上的值设为 value 。
 */
static void _intsetSet(intset *is, int pos, int64_t value) {
    // 取出集合的编码方式
    uint32_t encoding = intrev32ifbe(is->encoding);
    // 根据编码 ((Enc_t*)is->contents) 将数组转换回正确的类型
    // 然后 ((Enc_t*)is->contents)[pos] 定位到数组索引上
    // 接着 ((Enc_t*)is->contents)[pos] = value 将值赋给数组
    // 最后， ((Enc_t*)is->contents)+pos 定位到刚刚设置的新值上 
    // 如果有需要的话， memrevEncifbe 将对值进行大小端转换
    if (encoding == INTSET_ENC_INT64) {
        ((int64_t*)is->contents)[pos] = value;
        memrev64ifbe(((int64_t*)is->contents)+pos);
    } else if (encoding == INTSET_ENC_INT32) {
        ((int32_t*)is->contents)[pos] = value;
        memrev32ifbe(((int32_t*)is->contents)+pos);
    } else {
        ((int16_t*)is->contents)[pos] = value;
        memrev16ifbe(((int16_t*)is->contents)+pos);
    }
}

/*将16位小端序转为大端序*/
void memrev16(void *p) {
    unsigned char *x = p, t;
    t = x[0];
    x[0] = x[1];
    x[1] = t;
}
/* 将32位小端序转为大端序 */
void memrev32(void *p) {
    unsigned char *x = p, t;
    t = x[0];
    x[0] = x[3];
    x[3] = t;
    t = x[1];
    x[1] = x[2];
    x[2] = t;
}

static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {
    void *src, *dst;
    // 要移动的元素个数
    uint32_t bytes = intrev32ifbe(is->length)-from;
    // 集合的编码方式
    uint32_t encoding = intrev32ifbe(is->encoding);
    // 根据不同的编码
    // src = (Enc_t*)is->contents+from 记录移动开始的位置
    // dst = (Enc_t*)is_.contents+to 记录移动结束的位置
    // bytes *= sizeof(Enc_t) 计算一共要移动多少字节
    if (encoding == INTSET_ENC_INT64) {
        src = (int64_t*)is->contents+from;
        dst = (int64_t*)is->contents+to;
        bytes *= sizeof(int64_t);
    } else if (encoding == INTSET_ENC_INT32) {
        src = (int32_t*)is->contents+from;
        dst = (int32_t*)is->contents+to;
        bytes *= sizeof(int32_t);
    } else {
        src = (int16_t*)is->contents+from;
        dst = (int16_t*)is->contents+to;
        bytes *= sizeof(int16_t);
    }
    // 进行移动
    // T = O(N)
    memmove(dst,src,bytes);
}

压缩列表

源码参见：ziplist.h和ziplist.c。

当有序集合或哈希的元素数目比较少，且元素都是短字符串或整形时，Redis便使用压缩列表作为其底层数据存储方式。列表使用快速链表（quicklist）数据结构存储，而快速链表就是双向链表与压缩列表的组合(见后)。

/*  
 * 创建并返回一个新的 ziplist 
 */
unsigned char *ziplistNew(void) {
    // ZIPLIST_HEADER_SIZE 是 ziplist 表头的大小
    // +1 字节是表末端 ZIP_END 的大小
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;
    // 为表头和表末端分配空间
    unsigned char *zl = zmalloc(bytes);
    // 初始化表属性
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
    ZIPLIST_LENGTH(zl) = 0;
    // 设置表末端
    zl[bytes-1] = ZIP_END;
    return zl;
}

// 定位到bytes 属性，该属性记录了整个 ziplist 所占用的内存字节数
#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl)))
// 定位到offset 属性，该属性记录了到达表尾节点的偏移量
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))
// 定位到length 属性，该属性记录了 ziplist 包含的节点数量
#define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))
// 返回 ziplist 表头的大小
#define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))
// 返回指向 ziplist 第一个节点（的起始位置）的指针
#define ZIPLIST_ENTRY_HEAD(zl)  ((zl)+ZIPLIST_HEADER_SIZE)
// 返回指向 ziplist 最后一个节点（的起始位置）的指针
#define ZIPLIST_ENTRY_TAIL(zl)  ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl)))
// 返回指向 ziplist 末端 ZIP_END （的起始位置）的指针
#define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)

结构定义

// 保存 ziplist 节点信息的结构
typedef struct zlentry {
    // prevrawlen ：前置节点的长度
    // prevrawlensize ：编码 prevrawlen 所需的字节大小
    unsigned int prevrawlensize, prevrawlen;
    // len ：当前节点值的长度
    // lensize ：编码 len 所需的字节大小
    unsigned int lensize, len;
    // 当前节点 header 的大小
    // 等于 prevrawlensize + lensize
    unsigned int headersize;
    // 当前节点值所使用的编码类型
    unsigned char encoding;
    // 指向当前节点的指针
    unsigned char *p;
} zlentry;

节点细节

接口分析

unsigned char *ziplistNew(void);
/*创建空的压缩列表，只需要分配初始存储空间（11=4+4+2+1个字节），并对zlbytes、zltail、zllen和zlend字段初始化即可。*/
unsigned char *ziplistNew(void) {
    //ZIPLIST_HEADER_SIZE = zlbytes + zltail + zllen;
    //最后这个加1表示bytes本身
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;        
    unsigned char *zl = zmalloc(bytes);
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
    ZIPLIST_LENGTH(zl) = 0;
    //结尾标识0XFF
    zl[bytes-1] = ZIP_END;             
    return zl;
}

unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p,  unsigned char *s, unsigned int slen);

unsigned int zipRawEntryLength(unsigned char *p) {
    unsigned int prevlensize, encoding, lensize, len;
    ZIP_DECODE_PREVLENSIZE(p, prevlensize);
    ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
    return prevlensize + lensize + len;
}

encoding字段标识的是当前元素存储的数据类型以及数据长度，编码时首先会尝试将数据内容解析为整数，如果解析成功则按照压缩列表整数类型编码存储，解析失败的话按照压缩列表字节数组类型编码存储。

if (zipTryEncoding(s,slen,&value,&encoding)) {
    reqlen = zipIntSize(encoding);
} else {
    reqlen = slen;
}
reqlen += zipStorePrevEntryLength(NULL,prevlen);
reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl));
int forcelarge = 0;
nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
if (nextdiff == -4 && reqlen < 4) {
    nextdiff = 0;
    forcelarge = 1;
}
//存储偏移量
offset = p-zl;
//调用realloc重新分配空间
zl = ziplistResize(zl,curlen+reqlen+nextdiff);
//重新偏移到插入位置P
p = zl+offset;

那么nextdiff与forcelarge在这里有什么用呢？

memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff); 
//更新entryX+1元素的previous_entry_length字段字段
if (forcelarge)
    //entryX+1元素的previous_entry_length字段依然占5个字节;
    //但是entryNEW元素长度小于4字节
    zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
else
    zipStorePrevEntryLength(p+reqlen,reqlen);
//更新zltail字段
ZIPLIST_TAIL_OFFSET(zl) =
    intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
zipEntry(p+reqlen, &tail);
if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
    ZIPLIST_TAIL_OFFSET(zl) =
        intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
}
//更新zllen字段
ZIPLIST_INCR_LENGTH(zl,1);

三、数据类型的实现

Redis 并没有直接使用这些数据结构来实现键值对数据库，而是基于这些数据结构创建了一个对象系统，这个系统包含字符串对象、列表对象、哈希对象、集合对象和有序集合对象这五种类型的对象。

对象类型及编码，源码文件：object.c

typedef struct redisObject {
    // 对象类型
    unsigned type:4;
    // 底层编码
    unsigned encoding:4;
    // 对象最后一次被访问的时间(日后会说)
    unsigned lru:REDIS_LRU_BITS; /* lru time (relative to server.lruclock) */
    // 引用计数
    int refcount;
    // 指向实际存储单元的指针
    void *ptr;
} robj;

• 类型 该属性记录对象的类型，可选常量如下：

注：对于Redis键值对来说，键总是字符串对象，而值才是上述五种类型其中一种。

• 编码 对象的ptr指针指向对象的底层实现数据结构，而这些数据结构由对象的encoding属性决定。encoding属性记录了对象所使用的编码， 也即是说使用了什么数据结构作为对象的底层实现，可选的对象编码如下。

举个例子，在列表对象包含的元素比较少时，Redis使用压缩列表作为列表对象的底层实现： 1）因为压缩列表比双端链表更节约内存，并且在元素数量较少时，压缩列表更合适； 2）元素数量增多或者单个元素过大时，对象就会将底层实现从压缩列表转向功能更强、也更适合保存大量元素的双端链表上面；

• 空转时长（lru）：该属性记录了对象最后一次被访问的时间。该值在Redis的内存过期值淘汰时被用到。如果服务器用于回收内存的算法为 volatile-lru 或者 allkeys-lru ，那么当内存到上限值时，空转时长较高的那部分键对象会优先被服务器释放，从而回收内存。
• 引用计数 ： Redis在自己的对象系统中构建了一个引用计数技术实现的内存回收机制，通过这一机制，程序可以通过跟踪对象的引用计数信息，在适当的时候自动释放对象并进行内存回收。

- <font style="color:#333333;">在</font>**<font style="color:#333333;">创建一个新对象</font>**<font style="color:#333333;">时， 引用</font>**<font style="color:#333333;">计数的值会被初始化为1</font>**<font style="color:#333333;">；</font>
- <font style="color:#333333;">当对象被一个新程序使用时，它的引用计数值会被</font>**<font style="color:#333333;">增一</font>**<font style="color:#333333;">；</font>
- <font style="color:#333333;">当对象</font>**<font style="color:#333333;">不再被一个程序使用时</font>**<font style="color:#333333;">，</font>**<font style="color:#333333;">它的引用计数值会被减一</font>**<font style="color:#333333;">；</font>
- <font style="color:#333333;">当对象的引用计数值变为0时，</font>**<font style="color:#333333;">对象所占用的内存会被释放</font>**<font style="color:#333333;">；</font>

void decrRefCount(robj *o) {
    if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0");
    // 释放对象
    if (o->refcount == 1) {
        switch(o->type) {
        case REDIS_STRING: freeStringObject(o); break;
        case REDIS_LIST: freeListObject(o); break;
        case REDIS_SET: freeSetObject(o); break;
        case REDIS_ZSET: freeZsetObject(o); break;
        case REDIS_HASH: freeHashObject(o); break;
        default: redisPanic("Unknown object type"); break;
        }
        zfree(o);
    // 减少计数
    } else {
        o->refcount--;
    }
}

• ptr，该指针指向实际存储的数据单元。

因为需要有一个验证时间！只有当共享目标和想要建立的键一样时，才会共享，验证整数只需要O(1)，字符串需要O(n），列表或hash则需要O(n^2)，这将造成CPU资源浪费。

字符串对象

源码文件：t_string.c。

void setGenericCommand(redisClient *c, int flags, robj *key, robj *val, robj *expire, int unit, robj *ok_reply, robj *abort_reply) {
    long long milliseconds = 0; /* initialized to avoid any harmness warning */
    // 取出过期时间
    if (expire) {
        // 取出 expire 参数的值
        // T = O(N)
        if (getLongLongFromObjectOrReply(c, expire, &milliseconds, NULL) != REDIS_OK)
            return;

        // expire 参数的值不正确时报错
        if (milliseconds <= 0) {
            addReplyError(c,"invalid expire time in SETEX");
            return;
        }
        // 不论输入的过期时间是秒还是毫秒
        // Redis 实际都以毫秒的形式保存过期时间
        // 如果输入的过期时间为秒，那么将它转换为毫秒
        if (unit == UNIT_SECONDS) milliseconds *= 1000;
    }
    // 如果设置了 NX 或者 XX 参数，那么检查条件是否不符合这两个设置
    // 在条件不符合时报错，报错的内容由 abort_reply 参数决定
    if ((flags & REDIS_SET_NX && lookupKeyWrite(c->db,key) != NULL) ||
        (flags & REDIS_SET_XX && lookupKeyWrite(c->db,key) == NULL))
    {
        addReply(c, abort_reply ? abort_reply : shared.nullbulk);
        return;
    }
    // 将键值关联到数据库
    setKey(c->db,key,val);
    // 将数据库设为脏
    server.dirty++;
    // 为键设置过期时间
    if (expire) setExpire(c->db,key,mstime()+milliseconds);
    // 发送事件通知
    notifyKeyspaceEvent(REDIS_NOTIFY_STRING,"set",key,c->db->id);
    // 发送事件通知
    if (expire) notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,
        "expire",key,c->db->id);
    // 设置成功，向客户端发送回复
    // 回复的内容由 ok_reply 决定
    addReply(c, ok_reply ? ok_reply : shared.ok);
}

• int编码，如果字符串对象的整数值可以被long类型表示，那么该整数值会被保存在ptr指针中（编码格式跟实际参数类型不一样）；
• raw编码，如果字符串对象保存的是字符串，且大于39字节，那就使用SDS字符串保存；
• embstr编码，如果该保存的字符串小于等于39字节，就用embstr格式保存。embstr是专门用来保存短字符串的一种优化编码方式，跟raw的区别在于，它只需要调用一次内存分配一块连续的内存空间，依次包含redisObject和sdshdr，而raw是分别调用两次内存分配来存放redisObject和sdshdr，embstr编码时实例如下图：

1. embstr 编码将创建字符串对象所需的内存分配次数从 raw 编码的两次降低为一次。
2. 释放embstr编码的字符串对象只需要调用一次内存释放函数，而释放raw编码的字符串对象需要调用两次内存释放函数。
3. 因为embstr编码的字符串对象的所有数据都保存在一块连续的内存里面，所以这种编码的字符串对象比起raw编码的字符串对象能够更好地利用缓存带来的优势。

编码转换

• int编码的对象执行命令之后变成了字符串值，那就将从int转换为raw；
• Redis没有为embstr编码的字符串对象编写任何相应的修改程序 （只有int编码的字符串对象和raw编码的字符串对象有这些程序），embstr编码的字符串对象实际上是只读的：当我们对embstr编码的字符串对象执行任何修改命令时，程序会先将对象的编码从embstr转换成raw，然后再执行修改命令；因为这个原因，embstr编码的字符串对象在执行修改命令之后，总会变成一个raw编码的字符串对象。

列表对象

源码文件：t_list.c。

其底层编码是ziplist或linkedlist。

• ziplist编码，底层是压缩列表，每个entry保存一个列表项，有两个条件：列表对象保存的所有字符串元素的长度都小于 64 字节；列表对象保存的元素数量小于 512 个；

void listTypeConvert(robj *subject, int enc) {
    listTypeIterator *li;
    listTypeEntry entry;
    redisAssertWithInfo(NULL,subject,subject->type == REDIS_LIST);
    // 转换成双端链表
    if (enc == REDIS_ENCODING_LINKEDLIST) {
        list *l = listCreate();
        listSetFreeMethod(l,decrRefCountVoid);
        // 遍历 ziplist ，并将里面的值全部添加到双端链表中
        li = listTypeInitIterator(subject,0,REDIS_TAIL);
        while (listTypeNext(li,&entry)) listAddNodeTail(l,listTypeGet(&entry));
        listTypeReleaseIterator(li);
        // 更新编码
        subject->encoding = REDIS_ENCODING_LINKEDLIST;
        // 释放原来的 ziplist
        zfree(subject->ptr);
        // 更新对象值指针
        subject->ptr = l;
    } else {
        redisPanic("Unsupported list conversion");
    }
}

• linkedlist编码，底层是双端链表来保存SDS字符串对象；

注：字符串对象是 Redis 五种类型的对象中唯一一种会被其他四种类型对象嵌套的对象。

哈希对象

源码文件t_hash.c。

• ziplist编码，底层是压缩列表，有两个条件，哈希对象保存的所有键值对的键和值的字符串长度都小于 64 字节；哈希对象保存的键值对数量小于 512 个；每当有新的键值对要加入到哈希对象时，保存了同一键值对的两个节点总是紧挨在一起，保存键的节点在前，保存值的节点在后；且这些键值对遵循先来后到的原则。

注：在使用ziplist编码的时候，获取键对应值的时间复杂度不是O(1)，而是O(N^2)(先遍历键，在遍历值是否相等)！但由于使用ziplist的时候，长度和键的长度较小，对性能影响不是很大。

• hashtable编码，当不满足上述的两个条件时，就使用字典作为底层实现。哈希对象中的每个键值对都使用一个字典键值对来保存：字典的每个键和值都是一个字符串对象；

注：当ziplist不满足两个条件其中之一时，原本保存在压缩列表里的所有键值对都会被转移并保存到字典里面。

集合对象

源码文件：t_set.c。

• intset编码，使用整数集合作为集合对象的实现，有两个条件：集合对象保存的所有元素都是整数值；集合对象保存的元素数量不超过512个；

• hashtable编码，当intset编码的两个条件不满足时，使用字典作为底层实现，字典的每个键都是一个字符串对象，每个字符串对象包含了一个集合元素，而字典的值则全部被设置为 NULL（因为没有没有键就代表了值）。

有序集合对象

源码文件t_zset.c。

• ziplist 编码，使用压缩列表作为底层实现有两个条件：有序集合保存的元素数量小于 128 个；有序集合保存的所有元素成员的长度都小于 64 字节；

注：ziplist为了保证有序性，每次插入删除操作时，都可能需要对其中的数据进行移动。

• skiplist 编码，不满足上述条件时的有序集合对象使用zset 结构作为底层实现，一个zset结构同时包含一个字典和一个跳跃表：

typedef struct zset {     
    zskiplist *zsl;     
    dict *dict; 
} zset;

为什么需要这么做呢？

Hyperloglog对象

类型检查

多态命令

四、数据库实现

struct redisServer {
    // ……
    // 保存服务器中所有的数据库
    redisDb *db；
    //……
    // 决定服务器初始化时创建的数据库数量
    // 默认16
    int dbnum；
}

注：可以通过select指令进行数据库的切换（其实就是更改db的指向）。

数据库键空间