Redis深度历险-压缩列表

在zset和hash在元素个数较少时会采用压缩列表来存储以节省空间，主要代码在ziplist.c和ziplist.h中；这是非常重要的数据结构，在zset、hash、list的底层数据结构之一

设计思想

压缩列表并没有定义成结构体，而是以字符串的形式返回出去，必须要先明白压缩列表的空间排布

压缩列表

空间排布.png

• zlbytes：用以获取整个压缩列表的长度，4字节长
• zltail_offset：用以获取最后一个元素的地址，4字节长；这个字段是用来方便倒序查找的
• zllength：存储的元素个数，2字节
• entry：元素，长度不固定
• zlend：存储的是特殊字符0xff，1字节长，用于标记压缩列表末端

元素

即使是单个元素entry内部的排布也是很复杂的，主要是为了尽可能节省空间

存储类型

压缩列表支持保存一个整数或者一个字符串，支持以下类型

• ZIP_STR_06B：长度小于2^6的字符串
• ZIP_STR_14B：长度小于2^14的字符串
• ZIP_STR_32B：长度小于2^32的字符串
• ZIP_INT_16B：2字节长的有符号整数
• ZIP_INT_32B：4字节长的有符号整数
• ZIP_INT_64B：8字节长的有符号整数
• ZIP_INT_24B：3字节长的有符号整数
• ZIP_INT_8B：1字节长的有符号整数
• 0~12的极小值

不同类型的内存存储方式是不一样的

内存排布

元素空间排布.png

• prelen：存储的是前一个元素的内容长度，分两种情况
  • 前一个元素内容长度小于254字节，则使用1个字节存储
  • 大于254字节的情况使用5个字节存储，第1个字节设置为标志位oxfe，后4字节保存真正的长度
• content：包含两部分，数据的类型和长度encoding、数据；encoding长度不定，通过位标识类型
  • 00xxxxxx：对应于ZIP_STR_06B，后6位表示长度
  • 01xxxxxx xxxxxxxx：对应于ZIP_STR_14B，后14位表示长度
  • 10000000：对应于ZIP_STR_32B，10后6位没有使用，该字节的后4个字节共32位来表示长度
  • 11000000、11010000、11100000、11110000、11111110用来表示2、4、8、3、1字节长度的数字，11111111表示结束即oxff
  • encoding除了上述的值被占用了，后四位还有0001~1101可以用来表示极小值，即0～12

压缩列表固然设计的很精细巧妙，但是是否有点过度设计了？？？代码中的zlentry并不代表内存排布，只是便于操作而已；prelen的设计主要是方便从后往前的便利操作

集合中的压缩列表

集合在数据个数小于128zset_max_ziplist_entries个且插入的字符串小于64zset_max_ziplist_value时会使用跳表存储，这两个参数均可配

创建压缩列表

/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
    //空的压缩列表只有：zlbyte、zltail_offset、zllength、zlend
    unsigned int bytes = ZIPLIST_HEADER_SIZE+ZIPLIST_END_SIZE;
    //分配空间
    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;
}

//获取zlbyte的地址
#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl)))
//获取zltail_offset的地址
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))
//获取zllength的地址
#define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))

压缩列表没有固定的数据结构而是通过unsigned char*来表示，这里用到了很多宏来计算空间

插入数据

在zset中需要先根据score进行查找后插入到指定位置

unsigned char *zzlInsert(unsigned char *zl, sds ele, double score) {
    //定位到压缩列表头部
    unsigned char *eptr = ziplistIndex(zl,0), *sptr;
    double s;
        
    //元素在压缩列表中从大到小按需存储，这里从前往后遍历元素
    while (eptr != NULL) {
        //从前往后遍历压缩列表所有元素
        sptr = ziplistNext(zl,eptr);
        serverAssert(sptr != NULL);
        s = zzlGetScore(sptr);
                
        //优先比较分数、其次比较值
        if (s > score) {
            zl = zzlInsertAt(zl,eptr,ele,score);
            break;
        } else if (s == score) {
            if (zzlCompareElements(eptr,(unsigned char*)ele,sdslen(ele)) > 0) {
                zl = zzlInsertAt(zl,eptr,ele,score);
                break;
            }
        }

        eptr = ziplistNext(zl,sptr);
    }

    //如果插入的数据是最小的就插入在末尾
    if (eptr == NULL)
        zl = zzlInsertAt(zl,NULL,ele,score);
    return zl;
}

在有序集合中需要在业务上保证压缩列表是有序的

哈希表中的压缩列表

查找元素

int hashTypeGetFromZiplist(robj *o, sds field,
                           unsigned char **vstr,
                           unsigned int *vlen,
                           long long *vll)
{
    unsigned char *zl, *fptr = NULL, *vptr = NULL;
    int ret;

    serverAssert(o->encoding == OBJ_ENCODING_ZIPLIST);
        
    //调用ziplistFind查找元素
    zl = o->ptr;
    fptr = ziplistIndex(zl, ZIPLIST_HEAD);
    if (fptr != NULL) {
        fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
        if (fptr != NULL) {
            
            vptr = ziplistNext(zl, fptr);
            serverAssert(vptr != NULL);
        }
    }
        
    //获取元素的数据
    if (vptr != NULL) {
        ret = ziplistGet(vptr, vstr, vlen, vll);
        serverAssert(ret);
        return 0;
    }

    return -1;
}

内部调用了压缩列表的接口，本质就是遍历对比所有的元素

总结

这里不对代码做详细的分析，因为涉及到大量的内存操作，逻辑必然非常复杂
学习一下设计思想和空间排布就可以了，压缩列表仅在数据较少时可用，因为涉及到大量的遍历、数据复制移动操作