大家好，又见面了，我是你们的朋友全栈君。

redis6.0 源码学习（五）ziplist

一、数据结构

ziplist是经过特殊编码的双向链接列表，该列表具有很高的内存效率。 它存储字符串和整数值，其中整数被编码为实际整数，而不是一系列个字符。 它允许对列表的两侧进行push和pop操作且复杂度为O(1)。 但是由于每个操作都需要重新分配ziplist使用的内存，实际复杂度与ziplist使用的内存量有关。

下图是ziplist得示意图：

)

各个字段详细解释如下：

ziplist中的每个entry均包含两个信息：prevlen 和 encoding。prevlen 字段：存储前一个条目的长度，以便能够从后到前遍历列表。encoding表示entry类型，整数或字符串，对于字符串，还表示字符串有效负载的长度。 因此，完整的entry存储形式如下：

)

在存储较小的整数的时候，entry-data字段会不存在，因为也存储在encoding字段中了。

1、prevlen：针对prevlen的大小不同，prevlen占用的字段长度也不一样。

• prevlen < 254

  prevlen占用1个字节 8位

• prevlen > 254

  此时prevlen占用5个字节，第一个字节为 0xFE,后续四个字节为真正的长度。 0xFE表示是下一个entry一个大的值。

2、encoding 根据entry-data存入的是字符串还是整数具有不同的格式。具体如下：

• a：字符串

当字符串小于63字节时(2^6)，节点存为上图的第一种类型，高2位为00，低6位表示data的长度。

当字符串小于16383字节时(2^14)，节点存为上图的第二种类型，高2位为01，后续14位表示data的长度。

当字符串小于4294967296字节时(2^32)，节点存为上图的第三种类型，高2位为10，下一字节起连续32位表示data的长度。

• b：整数：整数节点的encoding的长度为8位，其中高2位用来区分整数节点和字符串节点（高2位为11时是整数节点），低6位用来区分整数节点的类型，具体如下：

二、代码解析

主要有以下接口，这里只分析几个，其他的大家可以自行去看源代码。

unsigned char *ziplistNew(void);
unsigned char *ziplistMerge(unsigned char **first, unsigned char **second);
unsigned char *ziplistPush(unsigned char *zl, unsigned char *s, unsigned int slen, int where);
unsigned char *ziplistIndex(unsigned char *zl, int index);
unsigned char *ziplistNext(unsigned char *zl, unsigned char *p);
unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p);
unsigned int ziplistGet(unsigned char *p, unsigned char **sval, unsigned int *slen, long long *lval);
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen);
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p);
unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num);
unsigned int ziplistCompare(unsigned char *p, unsigned char *s, unsigned int slen);
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip);
unsigned int ziplistLen(unsigned char *zl);
size_t ziplistBlobLen(unsigned char *zl);

1、创建

可以对照图1 ziplist结构示意图 看这个创建的过程。

/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) { 
   
    unsigned int bytes = ZIPLIST_HEADER_SIZE+ZIPLIST_END_SIZE;  //<zlbytes>4字节<zltail>4字节<zllen>2字节<zlend>1字节，没有entry节点
    unsigned char *zl = zmalloc(bytes); //申请内存
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);//zlbytes赋值
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);//zltail赋值
    ZIPLIST_LENGTH(zl) = 0; //zllen赋值
    zl[bytes-1] = ZIP_END; //zlend赋值
    return zl;
}

2、查找

/* Find pointer to the entry equal to the specified entry. Skip 'skip' entries * between every comparison. Returns NULL when the field could not be found. */
//返回p节点之后data与vstr(长度是vlen)相等的节点，只找p节点之后每隔skip的节点
//时间复杂度 O(n)
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip) { 
   
    int skipcnt = 0;
    unsigned char vencoding = 0;
    long long vll = 0;

    while (p[0] != ZIP_END) { 
     /* ZIP_END: Special "end of ziplist" entry. */
        unsigned int prevlensize, encoding, lensize, len;
        unsigned char *q;

        ZIP_DECODE_PREVLENSIZE(p, prevlensize); //获取prevlen的字节数1还是5
        /* 获取'ptr'中 entry encoding type 和 data length (字符串长度或者整数使用的bytes ) 。 * encoding变量： entry的encoding值 *lensize变量：encode entry所需要的bytes数 *len变量： hold the entry length. */        
        ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
        q = p + prevlensize + lensize; //当前节点的data

        if (skipcnt == 0) { 
   
            /* Compare current entry with specified entry */
            if (ZIP_IS_STR(encoding)) { 
   //判断当前节点是不是字符串节点
                if (len == vlen && memcmp(q, vstr, vlen) == 0) { 
    //字符串比较
                    return p;
                }
            } else { 
   
                /* Find out if the searched field can be encoded. Note that * we do it only the first time, once done vencoding is set * to non-zero and vll is set to the integer value. */
                if (vencoding == 0) { 
    //这个代码块只会执行一次,判断vstr能够用整数表示 
                    /* 校验'vstr'指向的内容能否转换成整数，'vll'存这个整数值 'encoding'存编码格式 */
                    if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) { 
   
                        /* If the entry can't be encoded we set it to * UCHAR_MAX so that we don't retry again the next * time. */
                        vencoding = UCHAR_MAX;
                    }
                    /* Must be non-zero by now */
                    assert(vencoding); 
                }

                /* Compare current entry with specified entry, do it only * if vencoding != UCHAR_MAX because if there is no encoding * possible for the field it can't be a valid integer. */
                if (vencoding != UCHAR_MAX) { 
    //entry地址的内容转换成整数
                    long long ll = zipLoadInteger(q, encoding);
                    if (ll == vll) { 
   
                        return p; //相等的话返回
                    }
                }
            }

            /* Reset skip count */
            skipcnt = skip;
        } else { 
   
            /* Skip entry */
            skipcnt--;
        }

        /* Move to next entry */
        p = q + len;
    }

    return NULL;
}

3、插入

/* Insert an entry at "p". */
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) { 
   
    return __ziplistInsert(zl,p,s,slen);
}

/* Insert item at "p". */
unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) { 
   
    size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)), reqlen; // 当前长度和插入节点后需要的长度
    unsigned int prevlensize, prevlen = 0; // 前置节点长度和编码该长度值所需的长度
    size_t offset;
    int nextdiff = 0;
    unsigned char encoding = 0;
    long long value = 123456789; /* initialized to avoid warning. Using a value that is easy to see if for some reason we use it uninitialized. */
    zlentry tail;

    /* Find out prevlen for the entry that is inserted. */
    // 找出待插入节点的前置节点长度
    // 如果p[0]不指向列表末端，说明列表非空，并且p指向其中一个节点
    if (p[0] != ZIP_END) { 
   
         // 获取前置节点p的长度和编码该长度需要的字节
        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
    } else { 
           
        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
        if (ptail[0] != ZIP_END) { 
   // 如果p指向列表末端，表示列表为空
            prevlen = zipRawEntryLength(ptail);
        }
    }

    /* See if the entry can be encoded */
    if (zipTryEncoding(s,slen,&value,&encoding)) { 
   //尝试encoding成整数
        /* 'encoding' is set to the appropriate integer encoding */
        reqlen = zipIntSize(encoding); //获取编码长度
    } else { 
   
        /* 'encoding' is untouched, however zipStoreEntryEncoding will use the * string length to figure out how to encode it. */
        reqlen = slen;
    }
    /* We need space for both the length of the previous entry and * the length of the payload. */
    reqlen += zipStorePrevEntryLength(NULL,prevlen); // 获取前置节点的编码长度
    reqlen += zipStoreEntryEncoding(NULL,encoding,slen); // 获取当前节点的编码长度

    /* When the insert position is not equal to the tail, we need to * make sure that the next entry can hold this entry's length in * its prevlen field. */
    // 只要不是列表的末端插入，需要计算出下一个元素保存本元素prevlen字段空间是否足够, 不够时计算出欠缺的差值 
    // nextdiff大于0，那就说明需要对当前p指向的节点的header进行扩展
    int forcelarge = 0;
    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
    if (nextdiff == -4 && reqlen < 4) { 
   
        nextdiff = 0;
        forcelarge = 1;
    }

    /* Store offset because a realloc may change the address of zl. */
    offset = p-zl;// 存储p相对于列表zl的偏移地址
    zl = ziplistResize(zl,curlen+reqlen+nextdiff); // 重新分配空间，curlen当前列表的长度
    p = zl+offset; // 重新获取p的值

    /* Apply memory move when necessary and update tail offset. */
    if (p[0] != ZIP_END) { 
    // 非表尾插入，需要重新计算表尾的偏移量
        /* Subtract one because of the ZIP_END bytes */
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);/* 移动p原有位置和后面的内容到新的位置 */      

        /* Encode this entry's raw length in the next entry. */   
        if (forcelarge)
            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
        else
            zipStorePrevEntryLength(p+reqlen,reqlen);

        /* Update offset for tail */
        // 更新列表尾相对于表头的偏移量，将新节点的长度算上
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);

        /* When the tail contains more than one entry, we need to take * "nextdiff" in account as well. Otherwise, a change in the * size of prevlen doesn't have an effect on the *tail* offset. */
        // 如果新节点后面有多个节点，那么表尾的偏移量需要算上nextdiff的值
        zipEntry(p+reqlen, &tail);
        if (p[reqlen+tail.headersize+tail.len] != ZIP_END) { 
   
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
        }
    } else { 
   
        // 表尾插入，直接计算偏移量
        /* This element will be the new tail. */
        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
    }

    /* When nextdiff != 0, the raw length of the next entry has changed, so * we need to cascade the update throughout the ziplist */
     // 当nextdiff不为0时，表示需要新节点的后继节点对头部进行扩展
    //说明下一个元素需要扩展空间存放prevlen字段, 由于下一个元素空间变大, 有可能引起下下一个元素空间需要扩展, 下面函数检测后面元素, 并在需要时重置元素prevlen长度
    if (nextdiff != 0) { 
   
        // 需要对p所指向的header进行扩展更新
        // 有可能会引起连锁更新
        offset = p-zl;
        zl = __ziplistCascadeUpdate(zl,p+reqlen);
        p = zl+offset;
    }

    /* Write the entry */
    p += zipStorePrevEntryLength(p,prevlen); // 将新节点前置节点的长度写入新节点的header
    p += zipStoreEntryEncoding(p,encoding,slen); // 将新节点的值长度写入新节点的header
    if (ZIP_IS_STR(encoding)) { 
    // 写入节点值
        memcpy(p,s,slen);
    } else { 
   
        zipSaveInteger(p,value,encoding);
    }
    ZIPLIST_INCR_LENGTH(zl,1);  // 更新列表节点计数
    return zl;
}

三、总结

• ziplist是为节省内存空间而生的。
• ziplist是一个为Redis专门提供的底层数据结构之一，本身可以有序也可以无序。当作hash的底层实现时，节点之间没有顺序；当作为zset的底层实现时，节点之间会按照大小顺序排列。