原文：https://blog.csdn.net/men_wen/article/details/70229375

2. 压缩列表

压缩列表(ziplist)是列表键和哈希键的底层实现之一。当一个列表键只包含少量的列表项,并且每个列表项包含的数字小或者字符串短,那么Redis就会使用压缩列表来做列表键的底层实现。(注:redis在3.2之前是这样的，3.2版本之后使用quicklist实现)，这一节的学习也是基于3.2之前的，quicklist也会一一学习。

看下具体现象:

redis> rpush lst 1 2 3 4 6 "qq" "vv"
(integer)8
redis> object lst
"ziplist"

可以看到,当我们输入的都是比较小的时候使用了ziplist这种数据结构去存储,哈希键里面包含的所有键和值都是小整数值或者短字符串。

2.1 压缩列表的构成

首先需要明确,压缩列表的产生是Redis为了节约内存开发的,是一个由一系列特殊编码的连续内存块组成的顺序性数据结构。一个压缩列表可以包含任意数量个节点,每个节点可以保存自己数组或者一个整数值。如下图所示

image

zlbytes记录整个压缩列表占用的内存字节数,在对压缩列表进行内存重分配或计算zlend的位置时使用。zltail记录压缩列表尾节点距离压缩列表的起始地址有多少字节,通过这个偏移量,可以直接确定尾节点的位置。zllen记录压缩列表包含的节点数量,entryX表示各种节点,数量和长度不一定。zlend用于标记压缩列表的末端。
如图,如果有一个指针p指向该压缩列表,则尾巴节点的长度就是指针加上偏移量179(十六进制0xb3=16*11+3=179),列表的长度zllen为5,表示压缩列表包含5个节点。zlbytes为0xd2表示压缩列表的总长为210字节。

image

由上可知,每个压缩列表的节点可以保存一个字节数组或者一个整数值,那么每个节点肯定也有自己的结构。

2.2 压缩列表的节点

如图所示,每个压缩列表的节点都是由previous_entry_length、encoding、content组成的。下面分别来说一说这三个字段的含义。

image

2.2.1 previous_entry_length

previous_entry_length以自己为单位,记录的是压缩列表中前一个节点的长度,previous_entry_length自身的空间长度可以是1字节或者5字节。如果前一个字节的长度小于254自己,就是1字节(前一个节点的长度就保存在这里面,这两个值一个是本节点里这个字段本身的空间大小,存储的是前一个节点的空间大小,不要弄混了哈)。如果前一个大于254那么这个字段的空间长度就为5字节,存储的值为大于254的那个值(就是前一个节点的长度)。其中这5个字节,第一个字节会被设置为0xFF也就是254,之后的四个字节用来保存前一个节点的长度。因为前一个节点的长度被previous_entry_length属性记录了,所以程序可以通过指针的运算根据当前节点的起始地址来计算出前一个节点的起始地址。而压缩列表的从表尾向表头的遍历操作就是通过这个原理实现的,只要我们拥有了一个指向某个节点的起始地址指针,通过这个指针和这个字段,我们可以往回遍历出所有的节点，最终到达表头。如下:

image

2.2.2 encoding

节点encoding属性记录了节点的content属性所保存的数据类型及长度。可以为一字节、两字节或者五字节长,值的最高位为00、01或者10的是字节数组编码，这种编码表示节点的content属性保存着字节数组,数组的长度由编码除去最高2位之后的其他位记录。也就是说高2位其实代表的是类型是字节数组还是整数编码。值的最高位以11开头的是整数编码：这种编码表示节点的content属性保存着整数值。整数值的类型和长度由编码除去最高2位之后的其他位的记录。

2.2.3 content

节点的content属性负责保存节点的值,节点值可以是一个字节数组或者整数，值的类型和长度由节点的encoding属性决定。

2.3 连锁更新

之前说过,每个节点的previous_entry_length都记录了前一个节点的长度,如果长度小于254那么previous_entry_length需要用1字节来保存这个长度值。现在假设这种情况:压缩列表有多个连续的长度介于250-253之间的节点e1-eN。因为每个节点的长度都小于254字节，所以这些节点的previous_entry_length属性都是1字节长度。此时如果将一个长度大于254的新节点设置为压缩列表的头节点，那么这个新节点成为头节点，也就是e1节点的前置节点。此时将e1的previous_entry_length扩展为5字节长度,此时e1又超过了254，于是e2的previous_entry_length也超过了254··· .此时这些节点就会连锁式的更新，并重新分配空间。除了新增加的节点会引发连锁更新之外，删除也会。假设中间有一个小于250的删除了，也会连锁更新。同上面所说的类似。因为连锁更新在最坏的情况下需要对压缩列表执行N次空间重分配操作，而每次空间重分配的最坏复杂度为O(N),所以连锁更新的最坏复杂度为`O(N^2)。虽然这很耗费时间，但是实际情况下这种发生的概率非常低的。对很少一部分节点进行连锁更新绝对不会影响性能的。

小结

压缩列表是redis3.2之前为了节约内存开发的顺序性数据结构，它被用作列表键和哈希键的底层实现之一，压缩列表可以包含多个节点，每个节点保存一个字节数组或者整数值，在添加和删除的时候，可能会引发连锁更新操作，但是这种操作出现的频率不高。
ziplist效率高省内存的原因：

1. 双向链表一个节点prev和next指针就16字节了，ziplist的previous_entry_length1-5字节，ziplist根据指针偏移量就可以很方便定位每个元素
2. 分配大块连续内存减小内存碎片

3. quicklist

3.1 简介

如果使用的redis3.2版本以上的,那么就会发现在程序中quicklist基本取代了ziplist。既然取代肯定意味着有功能上有优化并且对程序更加友好。其实Redis对外暴露的list的数据结构的底层实现就是quicklist。先回忆下list这种数据结构的特点:

1. list两端的增加和删除很方便,时间复杂度为O(1)
2. list是一个双向链表
3. list可以在中间的任意位置插入,时间复杂度为O(N)
4. list可以被用来作为队列,因为它上面的特性

而quicklist是一个ziplist的双向链表(双向链表是由多个节点Node组成的,这里上面有介绍)。也就是说quicklist的每个节点都是一个ziplist。ziplist本身也记录了数据节点的顺序，而且在内存中的位置是相邻的。

3.2 quicklist的结构

上面提到过,quicklist是由ziplist组成的双向链表,链表中的每个节点都以压缩列表ziplist的结构保存着数据,而ziplist有多个entry节点保存多个数据。相当于在一个quicklist节点中保存的是一整片数据,而不是一个单独的数据。

//真正表示quicklist的数据结构
typedef struct quicklist {
    // 指向头节点的指针（最左边）
    quicklistNode *head;
    // 指向尾节点的指针(最右边)
    quicklistNode *tail;
    // 所有ziplist数据项的个数总和
    unsigned long count;       
    // quicklist节点的个数
    unsigned int len;           
    // ziplist大小设置
    int fill : 16;              
    // 节点压缩深度设置
    unsigned int compress : 16;
} quicklist;
typedef struct quicklistNode {
    //前驱节点
    struct quicklistNode *prev;
    //后继节点
    struct quicklistNode *next;
    //数据指针,如果当前节点的数据没有压缩,它就指向一个ziplist结构,否则指向quicklistLZF结构
    unsigned char *zl;
    // 表示zl指向的ziplist的总大小,如果ziplist被压缩了,它的值仍然是压缩前的大小
    unsigned int sz;            
    // 表示ziplist里面包含的数据项个数,这个字段16bit
    unsigned int count : 16;    
    // 表示ziplist是否压缩了,1代表没有压缩  2代表使用LZF压缩
    unsigned int encoding : 2;  
    // 预留字段,固定值2,表示使用ziplist作为数据容器
    unsigned int container : 2; 
    // 此节点之前是否已经压缩过
    unsigned int recompress : 1; 
    // 测试用的,暂时用不上
    unsigned int attempted_compress : 1; 
    // 扩展字段,暂时无用
    unsigned int extra : 10;
} quicklistNode;

// 此结构表示一个被压缩过的ziplist
typedef struct quicklistLZF {
    // 压缩后的ziplist大小
    unsigned int sz;
    // 存放压缩后的ziplist字节数组
    char compressed[];
} quicklistLZF;

源码看了一遍,参照上面的表述,基本可得图下图的quicklist的数据结构.

image

3.2 创建quicklist及其节点

quicklist *quicklistCreate(void) {
    struct quicklist *quicklist;  
    // 为quicklist分配内存
    quicklist = zmalloc(sizeof(*quicklist)); 
    // 初始条件下,头和尾都是null
    quicklist->head = quicklist->tail = NULL;
    // quicklist初始长度0
    quicklist->len = 0;  // 设定长度
    // 数据项的总和,初始也是0
    quicklist->count = 0;
    // 压缩深度
    quicklist->compress = 0;
    // 设定ziplist大小限定
    quicklist->fill = -2;  
    return quicklist;
}

3.3 quicklist查找迭代器实现

现在基本已经知道了quicklist的基本结构,Redis为这个结构特意实现了一个迭代器,看下源码.

//quicklist的迭代器
typedef struct quicklistIter {
    //指向所属的quicklist的指针
    const quicklist *quicklist;
    // 当前quicklistNode节点
    quicklistNode *current;
    // 当前quicklist节点中的ziplist
    unsigned char *zi;
    // 当前ziplist结构中的偏移量,这个用处前面有介绍过,通过这个值和zllen可以得出ziplist的尾节点
    long offset;  
    // 迭代的方向标识前序还是后序,因为是双向列表
    int direction;           
} quicklistIter;

3.4 PUSH操作

push一个entry到quicklist的头节点或尾节点中的ziplist的头部或尾部。底层调用了ziplistPush操作。具体push过程如下代码:

//返回0可能插在尾节点或者中间的某个位置
//返回1代表节点插入在头部,插入的节点就是头节点
int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
    //暂存头结点的地址
    quicklistNode *orig_head = quicklist->head; 
    //判断ziplist允许插入entry节点
    if (likely(
            _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        quicklist->head->zl =
            ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);  
        //将节点push到头部并更新quicklistNode记录ziplist大小的属性sz
        quicklistNodeUpdateSz(quicklist->head); 
    } else {        
        //如果不允许插入entry节点到ziplist就新创建一个节点
        quicklistNode *node = quicklistCreateNode(); 
        //将entry节点push到新创建的quicklistNode节点中
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        //更新sz并把新穿件的节点插入到头节点的位置
        quicklistNodeUpdateSz(node);    
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);   
    }
    quicklist->count++;                     
    quicklist->head->count++;  
    // 判断整个更新过程中头节点是否变化,没有变化返回0,变化返回1
    return (orig_head != quicklist->head);  
}

/* Add new entry to tail node of quicklist.
 * push到尾节点,返回0表示尾节点指针没有改变,返回1表示改变了
 * Returns 0 if used existing tail.
 * Returns 1 if new tail created. 
 */
int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_tail = quicklist->tail;
    if (likely(
            _quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) {
        quicklist->tail->zl =
            ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL);  
        //push到尾部,更新sz
        quicklistNodeUpdateSz(quicklist->tail); 
    } else {
        quicklistNode *node = quicklistCreateNode();
        //新建一个quicklistNode,将entry节点push到新创建的quicklistNode节点中
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL);
        quicklistNodeUpdateSz(node);
        //将刚刚新创建的节点插入到尾节点后
        _quicklistInsertNodeAfter(quicklist, quicklist->tail, node);
    }
    quicklist->count++;
    quicklist->tail->count++;
    return (orig_tail != quicklist->tail); 
}

3.5 POP

从quicklist的头节点或尾节点的ziplist中pop出一个entry，分2种情况,主要看entry保存的是字符串还是整数。如果字符串的话，需要传入一个函数指针，这个函数叫_quicklistSaver()，真正的pop操作还是在这两个函数基础上在封装了一次，来操作拷贝字符串的操作。如下：

//从quicklist的头节点或尾节点pop弹出出一个entry，并将value保存在传入传出参数
//返回0表示没有可pop出的entry
//返回1表示pop出了entry，存在data或sval中
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                       unsigned int *sz, long long *sval,
                       void *(*saver)(unsigned char *data, unsigned int sz)) {
    unsigned char *p;
    unsigned char *vstr;
    unsigned int vlen;
    long long vlong;
    int pos = (where == QUICKLIST_HEAD) ? 0 : -1;   //位置下标

    if (quicklist->count == 0)  //entry数量为0，弹出失败
        return 0;

    //初始化
    if (data)
        *data = NULL;
    if (sz)
        *sz = 0;
    if (sval)
        *sval = -123456789;

    quicklistNode *node;
    //记录quicklist的头quicklistNode节点或尾quicklistNode节点
    if (where == QUICKLIST_HEAD && quicklist->head) {
        node = quicklist->head;
    } else if (where == QUICKLIST_TAIL && quicklist->tail) {
        node = quicklist->tail;
    } else {
        return 0;           //只能从头或尾弹出
    }

    p = ziplistIndex(node->zl, pos);    //获得当前pos的entry地址
    if (ziplistGet(p, &vstr, &vlen, &vlong)) {  //将entry信息读入到参数中
        if (vstr) {     //entry中是字符串值
            if (data)
                *data = saver(vstr, vlen);  //调用特定的函数将字符串值保存到*data
            if (sz)
                *sz = vlen;                 //保存字符串长度
        } else {        //整数值
            if (data)
                *data = NULL;
            if (sval)
                *sval = vlong;  //将整数值保存在*sval中
        }
        quicklistDelIndex(quicklist, node, &p); //将该entry从ziplist中删除
        return 1;
    }
    return 0;
}

/* Return a malloc'd copy of data passed in */
//将data内容拷贝一份并返回地址
REDIS_STATIC void *_quicklistSaver(unsigned char *data, unsigned int sz) {
    unsigned char *vstr;
    if (data) {
        vstr = zmalloc(sz);     //分配空间
        memcpy(vstr, data, sz); //拷贝
        return vstr;
    }
    return NULL;
}