redis之压缩列表源码剖析

形象化设计模式实战             HELLO!架构                     redis命令源码解析

 

用过Redis的应该对其哈希命令不陌生，在探索其实现之前，先得了解Redis的一个内部映射数据结构——压缩列表。

 

1、zipList的结构

找到ziplist.c文件，在代码注释中：

 * The general layout of the ziplist is as follows:
 * <zlbytes><zltail><zllen><entry><entry><zlend>

 

 

//注意这三个定义，新建时会使用。

#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl))) //32位unsigned int,4个字节
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))
#define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2))) //16位unsigned int，2个字节
/* Create a new empty ziplist. 
 *
 * 创建并返回一个新的 ziplist 
 *
 * T = O(1)
 */
unsigned char *ziplistNew(void) {

    // ZIPLIST_HEADER_SIZE 是 ziplist 表头的大小
    // 1 字节是表末端 ZIP_END 的大小
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;

    // 为表头和表末端分配空间
    unsigned char *zl = zmalloc(bytes);

    // 初始化<zlbytes>,整个ziplist 占用的内存字节数，对ziplist 进行内存重分配，或者计算末端时使用。
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);
    // 初始化<zltail>,到达ziplist 表尾节点的偏移量。通过这个偏移量，可以在不遍历整个ziplist 的前提下，弹出表尾节点。
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);
    // 新的list还没有entry，所以为0
    ZIPLIST_LENGTH(zl) = 0;
    // 初始化<zlend>,用于标记ziplist的末端
    zl[bytes-1] = ZIP_END;

    return zl;
}

 

成功后，ziplist结构如图所示：

 

2、节点(entry)的结构

 

+-----------------------+---------------+-----------+-------------+

 

| pre_entry_length | encoding | length | content |

 

+-----------------------+----------------+-----------+-------------+

pre_entry_length:记录了前一个节点的长度，通过这个值，可以进行指针计算，从而跳转到上一个节点，占用1或5字节。

encoding            :content的数据类型，值为00，01，10，11，占用两bit。

length                 :content的长度，占用1-5字节（和encoding一起）。

content               :真正要保存的数据。

 

这里知道个大概就好，后面会详解。

 

 

3、几个重要的宏

 

这里先看两个宏，因为插入节点时会用到。

/* Decode the number of bytes required to store the length of the previous
 * element, from the perspective of the entry pointed to by 'ptr'. 
 *
 * 解码 ptr 指针，
 * 取出编码前置节点长度所需的字节数，并将它保存到 prevlensize 变量中。
 *
 * T = O(1)
 */
#define ZIP_DECODE_PREVLENSIZE(ptr, prevlensize) do {                          \
    if ((ptr)[0] < ZIP_BIGLEN) {                                               \
        (prevlensize) = 1;                                                     \
    } else {                                                                   \
        (prevlensize) = 5;                                                     \
    }                                                                          \
} while(0);

ZIP_BIGLEN为254。

根据编码方式的不同，pre_entry_length 域可能占用1 字节或者5 字节：

• 1 字节：如果前一节点的长度小于254 字节，那么只使用一个字节保存它的值。
• 5 字节：如果前一节点的长度大于等于254 字节，那么将第1 个字节的值设为254 ，然后用接下来的4 个字节保存实际长度。

 

/* Extract the encoding from the byte pointed by 'ptr' and set it into
 * 'encoding'. 
 *
 * 从 ptr 中取出节点值的编码类型，并将它保存到 encoding 变量中。
 *
 * T = O(1)
 */
#define ZIP_ENTRY_ENCODING(ptr, encoding) do {  \
    (encoding) = (ptr[0]); \
    if ((encoding) < ZIP_STR_MASK) (encoding) &= ZIP_STR_MASK; \
} while(0)

ZIP_STR_MASK为0xc0，二进制就是11000000，与11000000取&，只能四种结果：11000000,1000000,01000000,00000000，这就是之前说encoding类型的由来。

 

/* Decode the length encoded in 'ptr'. The 'encoding' variable will hold the
 * entries encoding, the 'lensize' variable will hold the number of bytes
 * required to encode the entries length, and the 'len' variable will hold the
 * entries length. 
 *
 * 解码 ptr 指针，取出列表节点的相关信息，并将它们保存在以下变量中：
 *
 * - encoding 保存节点值的编码类型。
 *
 * - lensize 保存编码节点长度所需的字节数。
 *
 * - len 保存节点的长度。
 *
 * T = O(1)
 */
#define ZIP_DECODE_LENGTH(ptr, encoding, lensize, len) do {                    \
                                                                               \
    /* 取出值的编码类型 */                                                     \
    ZIP_ENTRY_ENCODING((ptr), (encoding));                                     \
                                                                               \
    /* 字符串编码 */                                                           \
    if ((encoding) < ZIP_STR_MASK) {                                           \
        if ((encoding) == ZIP_STR_06B) {                                       \
            (lensize) = 1;                                                     \
            (len) = (ptr)[0] & 0x3f;                                           \
        } else if ((encoding) == ZIP_STR_14B) {                                \
            (lensize) = 2;                                                     \
            (len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1];                       \
        } else if (encoding == ZIP_STR_32B) {                                  \
            (lensize) = 5;                                                     \
            (len) = ((ptr)[1] << 24) |                                         \
                    ((ptr)[2] << 16) |                                         \
                    ((ptr)[3] <<  8) |                                         \
                    ((ptr)[4]);                                                \
        } else {                                                               \
            assert(NULL);                                                      \
        }                                                                      \
                                                                               \
    /* 整数编码 */                                                             \
    } else {                                                                   \
        (lensize) = 1;                                                         \
        (len) = zipIntSize(encoding);                                          \
    }                                                                          \
} while(0);

</pre><pre name="code" class="cpp">/*
 * 字符串编码类型
 */
#define ZIP_STR_06B (0 << 6)
#define ZIP_STR_14B (1 << 6)
#define ZIP_STR_32B (2 << 6)

ZIP_STR_06B就是00000000，ZIP_STR_14B就是01000000,ZIP_STR_32B就是10000000,

 

00开头：1 byte ，长度小于等于63 字节的字符数组。

01开头：2 byte， 长度小于等于16383 字节的字符数组。

10开头：5 byte ，长度小于等于4294967295 的字符数组。

11开头：表示content 部分保存着整数。

 

结构略显复杂，感觉好省内存啊，哈哈。

 

4、插入节点

 

由于代码量太大，所以代码不贴出。

大致步骤：

1. 记录到达ziplist 末端所需的偏移量（因为之后的内存重分配可能会改变ziplist 的地址，因此记录偏移量而不是保存指针）
2. 根据新节点要保存的值，计算出编码这个值所需的空间大小，以及编码它前一个节点的长度所需的空间大小，然后对ziplist 进行内存重分配（如果不是末端，还需要计算出节点的相对偏移量:nextdiff，如果nextdiff不等于0，需要对list进行联动更新）。
3. 设置新节点的各项属性：pre_entry_length 、encoding 、length 和content 。

4. 更新ziplist 的各项属性，比如记录空间占用的zlbytes ，到达表尾节点的偏移量zltail，以及记录节点数量的zllen 。

 

大致图解：

 

这里解释一下__ziplistCascadeUpdate：就是在nextdiff不等于0时触发，新节点后面节点的pre_entry_length是1字节，但新节点的长度需要5字节，这样就需要更改新节点后面的节点的pre_entry_length，然而，因为新节点后面节点的空间长度改变了，所以程序又必须检查它的后继节点，看它的pre_entry_length 能否编码新长度，如果不能的话，程序又需要继续对下一个节点进行扩容，这就是__ziplistCascadeUpdate要做的。这种检查的时间复杂度是O(N2)，但是这种概率比较小。

值得注意的是在抖动过程中，只扩展，不收缩，因为可能会出现重复的扩展——收缩——再扩展——再收缩的抖动（flapping）效果，这样只会降低Redis的性能。

 

内存映射数据结构可以节省大量的内存，不过，因为内存映射数据结构的编码和操作方式要比内部数据结构要复杂得多，所以内存映射数据结构所占用的CPU 时间会比作用类似的内部数据结构要多。

压缩列表解析到这了，查找（遍历列表查找），删除（移除后和insert一样进行__ziplistCascadeUpdate）没什么难点，有兴趣可以自己看下源码。