Redis源码阅读笔记

2021/3/17 青玉白露

前言

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

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

黄健宏老师的《Redis设计与实现》（第二版） 部分资料及图片摘抄自网上，并附有参考链接，侵删。

一、基本数据结构

学习Redis源码建议自底向上，从底层数据结构入手，一步一步感受Redis的设计之巧妙，源码之美妙。

动态字符串SDS

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

Redis 构建了一种名为简单动态字符串（simple dynamic string，SDS）的抽象类型，并用作默认字符串表示。而传统的C语言字符串在Redis中只用来作为字符串字面量（常量），如打印日志等等。

结构定义

先来看看SDS结构的定义：

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

由此观之，Redis的SDS其实是基于C语言传统的字符串数组进行封装。一个SDS实例的示意图如下：

值得注意的是，SDS也遵循C字符串的空字符结尾，并不计入len中，这么做的好处就是可以继续使用C相关的字符串库函数。
如:printf("%s", s->buf);

对比C字符串

接着我们思考一个问题：这么封装C字符串有什么好处呢？
其一、快速获取字符串长度。通过len属性可以在O(1)的时间复杂度下获取一个字符串的长度，而C字符串需要O（n）。
其二、杜绝缓冲区溢出。C字符串由于没有记录自身长度，很可能在一些操作（如strcat等）时，造成缓冲区溢出，见下图：

对s1执行拼接操作之后：

而SDS的API在对SDS进行修改时，会检查其空间是否充足，如果不够，会先进行扩容，之后进行相应的操作。
其三、减少修改字符串时带来的内存重分配次数。对于一个包含了 N 个字符的C字符串来说，这个C字符串的底层实现总是一个N+1个字符的数组（额外一个字符用于存空字符）。
如果执行拼接操作（append），那么在执行这个操作之前，程序需要先通过内存重分配来扩展底层数组的空间大小——如果忘了这一步就会产生缓冲区溢出。
如果执行截断操作（trim），那么在执行这个操作之后，程序需要通过内存重分配来释放字符串不再使用的那部分空间——如果忘了这一步就会产生内存泄漏。
内存重分配可能会造成系统调用，一般程序很少进行字符串长度修改，但是Redis作为数据库，会进行大量的字符串修改。为了避免这种缺陷，SDS通过未使用空间解除了字符串长度和底层数组长度之间的关联： 在SDS中，buf 数组的长度不一定就是字符数量加一，数组里面可以包含未使用的字节，而这些字节的数量就由 SDS 的 free 属性记录。
通过未使用空间，SDS 实现了空间预分配和惰性空间释放两种优化策略。
其四、空间预分配策略。当API对SDS进行修改时，不仅会分配必要空间还会分配额外空间，具体策略如下：
当修改后SDS长度小于1M（默认最大长度），则额外分配len长度；
当大于1M，则额外分配1M；
这种预分配策略，将连续增长N 次字符串所需的内存重分配次数从必定 N 次降低为最多 N 次。
其五、惰性空间释放。当SDS的API需要缩短SDS保存的字符串时， 程序并不立即使用内存重分配来回收缩短后多出来的字节，而是使用free属性将这些字节的数量记录起来，并等待将来使用。
其六、支持二进制。C字符串是以’\0’结尾的，这使得C字符串只能保存文本数据，而不能保存像图片、音频、视频、压缩文件这样的二进制数据。但是Redis需要、也具备这样的特点。
可以说正是诸如此类的小改进，造就了Redis的高效和稳定。
总结一下：

接口API
挑几个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; 
}

请大家结合sds的结构体定义，先想想，为什么这个API可以获取SDS长度？
在结构体里面， char* buf 和char buf[1]的效果差不多（对齐的情况），占4个字节；char buf[0] 和char buf[]是一样的，不占内存。链接
所以(void)(s-(sizeof(struct sdshdr)))其实就将内存地址转移到了*SDS结构体地址，于是就可以通过这个地址直接访问其成员变量。

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的free进行扩充，2倍原大小或者加上1M的额外空间。

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。

Redis实现的是双端链表，其被广泛用于实现 Redis 的各种功能，比如列表键，发布与订阅等等。通过将链表的void value设置为不同的类型，*Redis的链表可以用于保存各种不同类型的值。

结构定义

/*  * 双端链表结构定义  */ 
typedef struct list {     
    // 表头节点     
    listNode *head;     
    // 表尾节点     
    listNode *tail;          
    // 以下这些是函数指针，负责对应的功能     
    // 节点值复制函数     
    void *(*dup)(void *ptr);     
    // 节点值释放函数     
    void (*free)(void *ptr);     
    // 节点值对比函数     
    int (*match)(void *ptr, void *key);     
    // 链表所包含的节点数量     
    unsigned long len; 
} list;

其中链表节点定义如下：

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

一个链表的实例如下图所示：

接口分析

来挑几个API看看其具体实现：
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迭代器有什么用？

从源码上来看，list的迭代器主要是为了方便实现对list双向前进。迭代器作为一种设计模式，可以使得list适应不同的算法。

字典Dict

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

Redis中字典使用哈希表实现。同时，为了实现渐进式Rehash的操作，每一个字典都有两个hash表（新/旧）。

结构定义

Redis中字典使用哈希表实现。同时，为了实现渐进式Rehash的操作，每一个字典都有两个hash表（新/旧）。

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

其中的hash表定义如下：

//每个字典都使用两个哈希表，从而实现渐进式 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;

一个hash表的实例见下：

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

一个普通状态下的字典实例见下：

接下来我们看看其中的重要API，比如计算hash值、处理hash冲突或者是rehash。

接口分析

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

在其中有个函数：dictkeyindex()——计算该字典中可以插入该键值对的index，如果标志符号dict_can_resize为正，那么会在hash表的size和used比率大于1（即负载因子）时(没有执行BGSAVE)进行rehash或者大于5（执行BGSAVE）时进行强制rehash。
计算hash值时是这样的：index = hash & dict->ht[0].sizemask
即是利用计算出的hash值跟sizemask相与，这个hash算法被称为MurmurHash2（目前有3了，但是Redis没用），这种算法的优点在于，即使输入的键是有规律的，算法仍能给出一个很好的随机分布性，并且算法的计算速度也非常快。

为什么负载因子一个是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的执行时间。

  迭代器

  跟链表一样，Redis字典中也存在迭代器，主要是为了实现遍历一个字典。
  其定义如下：
  安全迭代器

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

  两种迭代器都共用同一个函数接口：dictEntry *dictNext(dictIterator *iter)
  该接口返回迭代器指向的当前节点。
  如果是安全迭代器调用该函数，会更新字典的iterator计数器（安全迭代器）；
  如果是非安全迭代器调用该函数，会计算此时字典的fingerprint，以确定用户没有违规操作。有关迭代器的重点函数是：dictScan（）。详见源码注释。

  跳跃表skiplist

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

Redis在两个地方用到了跳跃表，一个是实现有序集合，另一个是在集群节点中用作内部数据结构（后文会将，见Redis多机数据库）。
跳跃表是一种可以对有序链表进行近似二分查找的数据结构，它通过在每个节点中维持多个指向其他节点的指针，从而达到快速访问节点的目的。
跳跃表支持平均O(logN) 、最坏O(N)复杂度的节点查找，还可以通过顺序性操作来批量处理节点。
在大部分情况下，跳跃表的效率可以和平衡树相媲美，并且因为跳跃表的实现比平衡树要来得更为简单，所以有不少程序都使用跳跃表来代替平衡树。
通过下面这个图，来看看跳表这个数据结构：

跳表其实是空间换时间的结构，因为每隔一定的点都需要建立一个链表索引。
它的查询过程如下：

即先同层查找，而后往下一层找，直到最后的找到节点，类似一个二分的过程。
另外，我们想要为跳表插入或者删除数据，我们首先需要找到插入或者删除的位置，然后执行插入或删除操作，前边我们已经知道了，跳表的查询的时间复杂度为 O(logn），因为找到位置之后插入和删除的时间复杂度很低，为 O(1)，所以最终插入和删除的时间复杂度也为 O(longn)。
跳表的代码实现参考：链接

结构定义

跳跃表包含头尾节点、节点数目以及最大的层数，定义如下：

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;

跳表每个节点都具有多个层标记，每一个层标记带有两个属性：前进指针和跨度，前进指针表示访问其之后（score比它大）的其他节点，跨度表示当前节点跟后续节点的距离。跳跃表 level 层级完全是随机的（插入节点时会随机生成1-32的数字）。一般来说，层级越多，访问节点的速度越快。
在跳跃表中，节点按各自所保存的分值从小到大排列。
obj是指向一个字符串对象，而字符串对象则保存着一个SDS值。

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

其中的Redis成员对象定义如下：

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源码阅读（三）。

查找、删除、添加

我们以一个实际例子，来展现在跳跃表中，究竟是以何种方式进行节点的查找、删除和添加的。

• 添加节点

假设一个跳跃表刚开始为空，那么其实它就是一个简单的空链表结构：

当需要插入一个key=3的节点，得到一个随机值level = 3（随机抛出来的），那么该节点就具有level3的属性，此时跳跃表的结构如下：

·插入key = 2，随机的level = 1，如下：

···
·插入key = 100， 随机的level=2，如下：

okay，以上我们就完成了整个跳跃表的插入过程，那如果我们想查找66这个值，如何进行？
跳跃表的查询是从顶层往下找，那么会先从第顶层（即最高的level）开始找，查找方式就是一个二分查找，如过该层找不到指定值，就会跳到下一层，继续遍历，直到找到对应节点。查找具体过程如下：
level3）66比1大，那就往同层的右边走，5还是比66小，再往右走就倒表尾了，因此下沉，从5的level3下沉到level2。
level2）5的右边100比66大，所以从5的level2下沉到level1。
level1）5的右边66恰好等于，于是返回这个节点。

• 删除操作呢？

其实跟查找操作差不多，找到删除节点之后，进行单向链表的删除操作，唯一的区别就是需要在多个level进行列表节点的删除。以上部分参考链接。

接口分析

其中，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;
}

从上诉代码可以看出，跳跃表的表头节点跟其他的跳跃表节点一样，只不过忽略了BW指针和分值、成员对象等信息。
创建新节点如下（请仔细思考源码的实现）：

/*
 * 创建一个成员为 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;
}

其中计算节点随机level的函数如下：

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

返回值介乎 1 和 ZSKIPLIST_MAXLEVEL 之间（包含ZSKIPLIST_MAXLEVEL），根据随机算法所使用的幂次定律，越大的值生成的几率越小。

补充：有序集合

Redis的有序集合（Zset）数据结构是结合字典跟跳跃表实现的，其定义如下：

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

ZSET同时使用两种数据结构来持有同一个元素，从而提供 O(log(N)) 复杂度的有序数据结构的插入和移除操作。哈希表将 Redis 对象映射到分值上。而跳跃表则将分值映射到 Redis 对象上，以跳跃表的视角来看，可以说 Redis 对象是根据分值来排序的。更详细的介绍参见第三章。

HyperLogLog

源码文件：hyperloglog.c。

Redis 在 2.8.9 版本添加了 HyperLogLog 结构。HyperLogLog 是用来做基数统计的算法，优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。
在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。
但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素。

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

算法细节

它内部维护了 16384 个桶（bucket）来记录各自桶的元素数量。当一个元素到来时，它会散列到其中一个桶，以一定的概率影响这个桶的计数值。因为是概率算法，所以单个桶的计数值并不准确，但是将所有的桶计数值进行调合均值累加起来，结果就会非常接近真实的计数值。

为了便于理解HyperLogLog算法，我们先简化它的计数逻辑。因为是去重计数，如果是准确的去重，肯定需要用到 set 集合，使用集合来记录所有的元素，然后获取集合大小就可以得到总的计数。因为元素特别多，单个集合会特别大，所以将集合打散成 16384 个小集合。当元素到来时，通过 hash 算法将这个元素分派到其中的一个小集合存储，同样的元素总是会散列到同样的小集合。这样总的计数就是所有小集合大小的总和。使用这种方式精确计数除了可以增加元素外，还可以减少元素。
但是，集合打散并没有什么明显好处，因为总的内存占用并没有减少。HyperLogLog算法中每个桶所占用的空间实际上只有 6 个 bit，这 6 个 bit 自然是无法容纳桶中所有元素的，它记录的是桶中元素数量的对数值。
对数？怎么突然提到对数了？等等···Hyperloglog，log···难道对数才是这个数据结构的灵魂？！
先想想：一个随机的整数值，这个整数的尾部有一个 0 的概率是 50%，要么是 0 要么是 1（这里说的是二进制）。同样，尾部有两个 0 的概率是 25%，有三个零的概率是 12.5%，以此类推，有 k 个 0 的概率是 2^(-k)。如果我们随机出了很多整数，整数的数量我们并不知道，但是我们记录了整数尾部连续 0 的最大数量 K。我们就可以通过这个 K 来近似推断出整数的数量，这个数量就是 2^K！！
当然结果是非常不准确的，因为可能接下来你随机了非常多的整数，但是末尾连续零的最大数量 K 没有变化，但是估计值还是 2^K。你也许会想到要是这个 K 是个浮点数就好了，每次随机一个新元素，它都可以稍微往上涨一点点，那么估计值应该会准确很多。
HyperLogLog通过分配 16384 个桶，然后对所有的桶的最大数量 K 进行调合平均来得到一个平均的末尾零最大数量 K# ，K# 是一个浮点数，使用平均后的 2^K# 来估计元素的总量相对而言就会准确很多。不过这只是简化算法，真实的算法还有很多修正因子，因为涉及到的数学理论知识过于繁多，这里就不再精确描述。
下面我们看看Redis HyperLogLog 算法的具体实现。
我们知道一个HyperLogLog实际占用的空间大约是 16384 * 6bit / 8 = 12k 字节。但是在计数比较小的时候，大多数桶的计数值都是零。如果 12k 字节里面太多的字节都是零，那么这个空间是可以适当节约一下的。
Redis 在计数值比较小的情况下采用了稀疏存储，稀疏存储的空间占用远远小于 12k 字节。相对于稀疏存储的就是密集存储，密集存储会恒定占用 12k 字节。
不论是稀疏存储还是密集存储，Redis 内部都是使用字符串位图来存储 HyperLogLog 所有桶的计数值。

密集存储

密集存储的结构非常简单，就是连续 16384 个 6bit 串成的字符串位图。

那么给定一个桶编号，如何获取它的 6bit 计数值呢？这 6bit 可能在一个字节内部，也可能会跨越字节边界。我们需要对这一个或者两个字节进行适当的移位拼接才可以得到计数值。
假设桶的编号为idx，这个 6bit 计数值的起始字节位置偏移用 offset_bytes表示，它在这个字节的起始比特位置偏移用 offset_bits 表示。我们有：
offset_bytes = (idx 6) / 8 offset_bits = (idx 6) % 8
需要注意的是字节位序是左边低位右边高位，而通常我们使用的字节都是左边高位右边低位，我们需要在脑海中进行倒置。

稀疏存储

稀疏存储适用于很多计数值都是零的情况。下图表示了一般稀疏存储计数值的状态。

当多个连续桶的计数值都是零时，Redis 使用了一个字节来表示接下来有多少个桶的计数值都是零：00xxxxxx。前缀两个零表示接下来的 6bit 整数值加 1 就是零值计数器的数量，注意这里要加 1 是因为数量如果为零是没有意义的。比如 00010101表示连续 22 个零值计数器。6bit 最多只能表示连续 64 个零值计数器，所以 Redis 又设计了连续多个多于 64 个的连续零值计数器，它使用两个字节来表示：01xxxxxx yyyyyyyy，后面的 14bit 可以表示最多连续 16384 个零值计数器。这意味着 HyperLogLog 数据结构中 16384 个桶的初始状态，所有的计数器都是零值，可以直接使用 2 个字节来表示。
稀疏存储的一般状态如下：

回答上面的问题，何时从稀疏到非稀疏转换？
当稀疏存储的某个计数值需要调整到大于 32 （如33）时，Redis 就会立即转换 HyperLogLog 的存储结构，将稀疏存储转换成密集存储。
或者，稀疏存储占用的总字节数超过 3000 字节，这个阈值可以通过 hll_sparse_max_bytes 参数进行调整。
以上两种情况发生一种，就会触发稀疏到密集的转换，且不可逆。

计数缓存

前面提到 HyperLogLog 表示的总计数值是由 16384 个桶的计数值进行调和平均后再基于因子修正公式计算得出来的。它需要遍历所有的桶进行计算才可以得到这个值，中间还涉及到很多浮点运算。这个计算量相对来说还是比较大的。
所以 Redis 使用了一个额外的字段来缓存总计数值，这个字段有 64bit，最高位如果为 1 表示该值是否已经过期，如果为 0， 那么剩下的 63bit 就是计数值。
当 HyperLogLog 中任意一个桶的计数值发生变化时，就会将计数缓存设为过期，但是不会立即触发计算（惰性计算）。而是要等到用户显示调用 pfcount 指令时才会触发重新计算刷新缓存。缓存刷新在密集存储时需要遍历 16384 个桶的计数值进行调和平均，但是稀疏存储时没有这么大的计算量。也就是说只有当计数值比较大时才可能产生较大的计算量。另一方面如果计数值比较大，那么大部分 pfadd 操作根本不会导致桶中的计数值发生变化。
这意味着在一个极具变化的 HLL 计数器中频繁调用 pfcount 指令可能会有少许性能问题。关于这个性能方面的担忧在 Redis 作者 antirez 的博客中也提到了。不过作者做了仔细的压力的测试，发现这是无需担心的，pfcount 指令的平均时间复杂度就是 O(1)。

HHL数据结构暂时只需要理解其原理，基本介绍链接
文中具体算法细节转载自（强烈推荐）链接

二、内存编码数据结构的实现

这一部分主要介绍Redis特制的内存编码数据结构，建议结合图像来理解。

整数集合

源码：intset.h和intset.c。

整数集合（intset）是集合键的底层实现之一： 当一个集合只包含整数值元素，并且这个集合的元素数量（见后）不多时，Redis就会使用整数集合作为集合键的底层实现。

结构定义

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

其中，content是整数集合的底层实现，所有的元素都是该contents数组的一项，值得注意的是，虽然该数组声明为int8_t类型，但是该数组中存储的数据类型由encoding来决定：
encoding有三个选择：INTSET_ENC_INT16/32/64。

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

升级操作

当向当前的整数集合插入一个大于当前encodeing类型的数据时，必须先进行一次不可逆的升级操作：即将所有的元素编码升级为足以容纳新元素的encoding。
具体分为三步：
1.根据新元素的类型， 扩展整数集合底层数组的空间大小。
2.将底层数组现有的所有元素都转换成与新元素相同的类型，并将类型转换后的元素放置到正确的位上，而且在放置元素的过程中，需要继续维持底层数组的有序性质不变。
3.将新元素添加到底层数组里面。

举个例子：
开始时，整数集合的contents数据类型为INT16，其中存储的元素如下：

现在需要插入一个数字：65535，超过了INT16的数据范围，故需要进行升级操作，升级到32位。先进行空间的拓展，644=256位，现在只有48位，更新数组空间之后：

空间分配完毕，现在需要对原有的元素进行转移（因为原有的数字是INT16位格式存储的），为了保证元素有序性，从最后一位3进行转移：

转移原有字符之后，插入新元素65535：

升级操作的时间复杂度为O（N）。
而且值得注意的是，引起升级操作的新元素要么存在索引0位置（负的太离谱），要么在length-1（大的太离谱）。之所以Redis的整形数组采用这种升级的方式，有两个好处：
一是节约内存，如果想容纳INT64的数字，那么传统做法是直接声明INT64的数组，但有时其实并不需要INT64的格式存储，浪费内存；只有当真正需要INT64时，才会分配相应的内存空间。并且使用整数集合时元素数量较少，不会造成太高的时间成本。
二是高灵活性，整数集合可以通过自动升级底层数组来适应新元素， 所以我们可以随意地将 int16_t 、int32_t 或者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));
}

不对，上述的操作只是在指定的pos插入，那么如何确定这个pos呢？其实整数集合中的整数都是有序的，从小到大排列，因此在插入或者查找某值时都是先使用二分查找的方式进行查找，以下是往整数集合中插入一个value的函数：

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

看到其中有个memrev16/32/64ifbe函数，有些好奇，遂F12，找到了源码：

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

补充关于大端和小端序的转换，值得注意的是，大端小端在内存的存储的顺序是按照字节，而不是按位！
例如小端中的：00000000 10000000，好
在大端中存储不是：00000001 00000000 （即按位逆序）
而是按字节相反：10000000 00000000。
示意图：

转换方式：

所以现在来看memrev16/32就明白了，先用char来获取单一字节的数据，而后两两换位，即实现大端到小端的转变。不妨试试不看源码，写写memrev64？
那么集合是怎么进行删除的呢？*盲猜是靠内存的拷贝，将后续的数据拷贝至要删除的位置，看了一下果然如此：

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 为了节约内存而开发的，由一系列特殊编码的连续内存块组成的顺序型（sequential）数据结构。
Redis的有序集合、哈希以及列表都直接或者间接使用了压缩列表。
当有序集合或哈希的元素数目比较少，且元素都是短字符串或整形时，Redis便使用压缩列表作为其底层数据存储方式。列表使用快速链表（quicklist）数据结构存储，而快速链表就是双向链表与压缩列表的组合(见后)。
一个压缩列表可以包含任意多个节点（entry）， 每个节点可以保存一个字节数组或者一个整数值。以下就是一个压缩列表的结构：

1、zlbytes：压缩列表的字节长度，占4个字节，因此压缩列表最长(2^32)-1字节；
2、zltail：压缩列表尾元素相对于压缩列表起始地址的偏移量，占4个字节；
3、zllen：压缩列表的元素数目，占两个字节；那么当压缩列表的元素数目超过(2^16)-2怎么处理呢？此时通过zllen字段无法获得压缩列表的元素数目，必须遍历整个压缩列表才能获取到元素数目；
4、entry：压缩列表存储的元素，可以为字节数组或者整数；
5、zlend：压缩列表的结尾(不是尾元素的意思)，占一个字节，恒为0xFF。
在Redis中，压缩列表定义如下：

/*  
 * 创建并返回一个新的 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;
}

为便于快速操作压缩列表获取各字段的数据，redis有以下宏定义(char * 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;

节点细节

由上文节点定义代码可知，压缩节点信息可以分为三个部分：previous_entry_length，encoding，content，如下图：

现在就来详细看看这三个部分。
1.previous_entry_lentry：记录前一个节点的长度，1或5字节——前一节点小于254字节，那么就用1字节保存前一节点信息否则用五字节表示（那么第一字节设置为0xFE（254）后四字节为长度）。
当前节点指针和previous字段可以实现快速访问上一节点，从而实现列表节点回溯。

2.encoding字段：表示content所保存的数据类型及长度。可选1/2/5字节，分别表示字节数组或整形。详细编码可见下表：
字节数组编码如下：

整数编码如下：

3.content：负责保存节点的值，可以为一个字节数组或整数。如下两个例子分别表示一个11字节的字节数组和一个保存int16的整数值，不妨猜猜看哪一个是表示11字节的字节数组呢？


有关压缩列表指针所指地址的示例如下：

接口分析

创建压缩列表

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

插入元素
函数输入参数zl表示压缩列表首地址，p指向新元素的插入位置，s表示数据内容，slen表示数据长度，返回参数为压缩列表首地址。

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

插入元素时，可以简要分为三个步骤：
第一步需要将元素内容编码为压缩列表的元素，
第二步重新分配空间，
第三步拷贝数据。
下面分别讨论每个步骤的实现逻辑。
1)编码
编码即计算previous_entry_length字段、encoding字段和content字段的内容。如何获取前一个元素的长度呢？这时候就需要根据插入元素的位置分情况讨论了，如图所示：

当压缩列表为空插入位置为P0时，此时不存在前一个元素，即前一个元素的长度为0；
当插入位置为P1时，此时需要获取entryX元素的长度，而entryX+1元素的previous_entry_length字段存储的就是entryX元素的长度，比较容易获取；
当插入位置为P2时，此时需要获取entryN元素的长度，entryN是压缩列表的尾元素，计算其元素长度需要将其三个字段长度相加，函数实现如下：

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

2）重分配空间
空间大小是否是添加元素前的压缩列表长度与新添加元素元素长度之和呢？并不完全是，如图中所示的例子。

插入元素前，entryX元素长度为128字节，entryX+1元素的previous_entry_length字段占1个字节；
添加元素entryNEW元素，元素长度为1024字节，此时entryX+1元素的previous_entry_length字段需要占5个字节；
即压缩列表的长度不仅仅是增加了1024字节，还有entryX+1元素扩展的4字节。
很容易知道，entryX+1元素长度可能增加4字节，也可能减小4字节，也可能不变。而由于重新分配空间，新元素插入的位置指针P会失效，因此需要预先计算好指针P相对于压缩列表首地址的偏移量，待分配空间之后再偏移即可。

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在这里有什么用呢？
分析ziplistResize函数的3个输入参数，curlen表示插入元素前压缩列表的长度，reqlen表示插入元素元素的长度，而nextdiff表示的是entryX+1元素长度的变化，取值可能为0（长度不变）、4（长度增加4）和-4（长度减小4）。
我们再思考下，当nextdiff等于-4，而reqlen小于4时会发生什么呢？没错，插入元素导致压缩列表所需空间减少了，即函数ziplistResize底层调用realloc重新分配的空间小于指针zl指向的空间。这可能会存在问题，我们都知道realloc重新分配空间时，返回的地址可能不变，当重新分配的空间大小反而减少时，realloc底层实现可能会将多余的空间回收，此时可能会导致数据的丢失。因此需要避免这种情况的发生，即重新赋值nextdiff等于0，同时使用forcelarge标记这种情况。
可以再思考下，nextdiff等于-4时，reqlen会小于4吗？答案是可能的，连锁更新可能会导致这种情况的发生。连锁更新将在之后介绍。
3）数据拷贝
重新分配空间之后，需要将位置P后的元素移动到指定位置，将新元素插入到位置P。我们假设entryX+1元素的长度增加4（即nextdiff等于4），此时数据拷贝示意图如图所示：

从图中可以看到，位置P后的所有元素都需要移动，移动的偏移量是插入元素entryNew的长度，移动的数据块长度是位置P后所有元素长度之和再加上nextdiff的值，数据移动之后还需要更新entryX+1元素的previous_entry_length字段。

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

思考一下，第一次更新尾元素偏移量之后，为什么指向的元素可能不是尾元素呢？很显然，当entryX+1元素就是尾元素时，只需要更新一次尾元素的偏移量；但是当entryX+1元素不知尾元素，且entryX+1元素长度发生了改变，此时尾元素偏移量还需要加上nextdiff的值。
以上参考链接：强烈推荐

连锁更新

当往ziplist中插入或删除节点时，由于previous节点字节数可为1或5，保存的前置节点大小不一致，可能就会引发后续节点依次影响，从而发生多次空间重分配，这就是连锁更新。
比如插入的new恰好大于254字节，而原本entry都是介于250-253之间：

那么插入是如何导致的呢？先想再看：

解释——当e1-en都是250-253字节时，big大于254，small小于254，那么删除small就会造成e1之后节点的连锁更新。
在最坏的情况下，需要执行N次重分配操作，所以每次空间重分配的最坏复杂度为O(n^2）。虽然有如此严重的性能损耗，但是实际场景中发生的概率极低，所以ziplistPush等命令平均复杂度为O(n)。

三、数据类型的实现

在前面,我们陆续介绍了 Redis 用到的所有主要数据结构。
Redis 并没有直接使用这些数据结构来实现键值对数据库，而是基于这些数据结构创建了一个对象系统，这个系统包含字符串对象、列表对象、哈希对象、集合对象和有序集合对象这五种类型的对象。

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

每次当我们在Redis的数据库中新创建一个键值对时，我们至少会创建两个对象，一个对象用作键值对的键（键对象），另一个对象用作键值对的值（值对象）。
Redis中的每个对象都由一个redisObject结构表示：

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_STRING | 字符串对象 | | REDIS_LIST | 列表对象 | | REDIS_HASH | 哈希对象 | | REDIS_SET | 集合对象 | | REDIS_ZSET | 有序集合对象 |

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

• 编码 对象的ptr指针指向对象的底层实现数据结构，而这些数据结构由对象的encoding属性决定。encoding属性记录了对象所使用的编码， 也即是说使用了什么数据结构作为对象的底层实现，可选的对象编码如下。 | 编码常量 | 编码所对应的底层数据结构 | | —- | —- | | REDIS_ENCODING_INT | long 类型的整数 | | REDIS_ENCODING_EMBSTR | embstr 编码的简单动态字符串 | | REDIS_ENCODING_RAW | 简单动态字符串 | | REDIS_ENCODING_HT | 字典 | | REDIS_ENCODING_LINKEDLIST | 双端链表 | | REDIS_ENCODING_ZIPLIST | 压缩列表 | | REDIS_ENCODING_INTSET | 整数集合 | | REDIS_ENCODING_SKIPLIST | 跳跃表和字典 |

每种对象都至少（除了Hyperloglog）可以使用两种不同的编码，见下图：

通过encoding属性来设定对象所使用的编码，而不是为特定类型的对象关联一种固定的编码，极大地提升了 Redis 的灵活性和效率，因为 Redis可以根据不同的使用场景来为一个对象设置不同的编码，从而优化对象在某一场景下的效率。

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

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

对象的引用计数信息会随着对象的使用状态而不断变化：

• 在创建一个新对象时， 引用计数的值会被初始化为1；
• 当对象被一个新程序使用时，它的引用计数值会被增一；
• 当对象不再被一个程序使用时，它的引用计数值会被减一；
• 当对象的引用计数值变为0时，对象所占用的内存会被释放；

修改引用对象计数器有专门的API，如下：

对象的整个生命周期可以划分为创建对象、操作对象、释放对象三个阶段。释放之前最后调用的API即是decRefcount：

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，该指针指向实际存储的数据单元。

对象的引用计数属性带有对象共享（只对整数值）的作用。举个例子，假设键A创建了一个包含整数值100的字符串对象作为值对象:

如果这时键B也要创建一个同样保存了整数值100的字符串对象作为值对象，那么服务器有以下两种做法：
·为键 B 新创建一个包含整数值100的字符串对象；
·让键 A 和键 B 共享同一个字符串对象；
以上两种方法很明显是第二种方法更节约内存。
在 Redis 中，让多个键共享同一个值对象需要执行以下两个步骤：
·将数据库键的值指针指向一个现有的值对象；
·将被共享的值对象的引用计数增一。
如果键A和键B指向同一对象，现如今的情况如下：

共享对象机制对于节约内存非常有帮助，数据库中保存的相同值对象越多，对象共享机制就能节约越多的内存。
Redis会在初始化服务器时，创建一万个字符串对象，这些对象包含了从0到9999的所有整数值，当服务器需要用到值为0到9999的字符串对象时，服务器就会使用这些共享对象，而不是新创建对象。
为什么共享对象只对整数值，而不共享包含字符串的对象呢？
因为需要有一个验证时间！只有当共享目标和想要建立的键一样时，才会共享，验证整数只需要O(1)，字符串需要O(n），列表或hash则需要O(n^2)，这将造成CPU资源浪费。

字符串对象

源码文件：t_string.c。

字符串对象是Redis数据库所有键值对的根本，因为键都是字符串对象。如下是Redis中string对象SET、SETEX等命令的底层API：

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、raw、embstr。

• 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编码的字符串对象能够更好地利用缓存带来的优势。

若利用命令set msg “hello”，其值对象如下图所示：

值得注意，long double类型的浮点数在Redis中也是作为字符串来保存的，比如6.666，那么会将其转换为字符串保存。在需要用到该浮点数的时候又会先将其从字符串转成浮点数再计算。
编码转换
int/embstr在特定的情况下会转换为raw编码的对象，且不可逆。

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

列表对象

源码文件：t_list.c。

其底层编码是ziplist或linkedlist。

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

下图是一个ziplist编码的实例：

两个条件的任意一个不能被满足时，对象的编码转换操作就会被执行，从ziplist转为linkedlist。转换过程见该函数：

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或者hashtable。

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

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

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

以下是一个以字典为底层的哈希对象实例：

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

集合对象

源码文件：t_set.c。

集合对象的编码可以是intset或者hashtable 。

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

集合对象包含的所有元素都被保存在整数集合里面，见下：

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

  有序集合对象

  源码文件t_zset.c。

有序集合的编码可以是ziplist和skiplist。

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

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

每个集合元素使用两个紧挨在一起的压缩列表节点来保存，第一个节点保存元素的成员，而第二个元素则保存元素的分值（score）（排序）。

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

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

  
  
  zset结构中的跳跃表按分值从小到大保存了所有集合元素，每个跳跃表节点都保存了一个集合元素：跳跃表节点的 object 属性保存了元素的成员，而跳跃表节点的 score 属性则保存了元素的分值。

通过这个跳跃表， 程序可以对有序集合进行范围型操作，比如 ZRANK 、ZRANGE 等命令就是基于跳跃表 API 来实现的。
除此之外，zset结构中的dict字典为有序集合创建了一个从成员到分值的映射，字典中的每个键值对都保存了一个集合元素：字典的键保存了元素的成员，而字典的值则保存了元素的分值。

通过这个字典，程序可以用O(1)复杂度查找给定成员的分值,ZSCORE命令就是根据这一特性实现的，而很多其他有序集合命令都在实现的内部用到了这一特性。
有序集合每个元素的成员都是一个字符串对象，而每个元素的分值都是一个 double 类型的浮点数。 值得一提的是，虽然zset结构同时使用跳跃表和字典来保存有序集合元素，但这两种数据结构都会通过指针来共享相同元素的成员和分值，所以同时使用跳跃表和字典来保存集合元素不会产生任何重复成员或者分值，也不会因此而浪费额外的内存。

为什么需要这么做呢？
在理论上来说，有序集合可以单独使用字典或者跳跃表的其中一种数据结构来实现，但无论单独使用字典还是跳跃表，在性能上对比起同时使用字典和跳跃表都会有所降低。
举个例子，如果我们只使用字典来实现有序集合，那么虽然以O(1)复杂度查找成员的分值这一特性会被保留，但是，因为字典以无序的方式来保存集合元素，所以每次在执行范围型操作一一一比如ZRANK、ZRANG等命令时，程序都需要对字典保存的所有元素进行排序，完成这种排序需要至少O(NlogN)时间复杂度，以及额外的O(N)内存空间（因为要创建一个数组来保存排序后的元素）

另一方面，如果我们只使用跳跃表来实现有序集合，那么跳跃表执行范围型操作的所有优点都会被保留，但因为没有了字典，所以根据成员查找分值这一操作的复杂度将上升到到O(logn)。

因为以上原因，为了让有序集合的查找和范围型操作都尽可能快地执行，Redis选择了同时使用字典和跳跃表两种数据结构来实现有序集合。
注：常用命令

Hyperloglog对象

源码文件：hyperloglog.c。
建议直接参考：链接

类型检查和多态

Redis中用于操作键的命令基本分为两种，一种是键共有操作，比如说DEL命令、EXPIRE命令、RENAME命令、TYPE 命令、OBJECT命令等。另一种是键特定操作，如SET、GET、APPEND、STRLEN等命令只能对字符串键执行。

类型检查
在执行一个类型特定的命令之前， Redis 会先检查输入键的类型是否正确， 然后再决定是否执行给定的命令。通过 redisObject 结构的 type 属性来实现的：

多态命令
这里指的是同一个对象一般具有两种编码方式，比如hash对象是hashtable和ziplist，那么执行命令的时候是怎么确定的呢？也是根据其对象的编码格式来的：

四、数据库实现

在Redis中，服务器中所有的数据库都保存在redis.h/redisServer结构中的db数组中：

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

默认配置下的服务器启动之后的状态如下：

对于redis的客户端而言，其结构中的db属性指向了当前正在操作的目标数据库，假设一个客户端的目标数据库为1，实例如下：

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

数据库键空间

Redis服务器中的每个数据库都由redis.h/redisDb结构表示，如下：

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