redis源碼分析教程之壓縮鏈表ziplist詳解

//由元素2，5組成的壓縮鏈表
[0f 00 00 00] [0c 00 00 00] [02 00] [00 f3] [02 f6] [ff]
 |  |  | | | |
 zlbytes zltail entries "2" "5" end
//字符串"Hello World"編碼后的內容
[02] [0b] [48 65 6c 6c 6f 20 57 6f 72 6c 64]

[0f 00 00 00] [0c 00 00 00] [02 00] [00 f3] [02 f6] [02 0b 48 65 6c 6c 6f 20 57 6f 72 6c 64] [ff]
 |  |  | | |   |   |
 zlbytes zltail entries "2" "5"   "Hello World"  end

//定義由zlbytes，zltail跟zllen組成的壓縮鏈表的頭大小
#define ZIPLIST_HEADER_SIZE (sizeof(uint32_t)*2+sizeof(uint16_t))
//創建一個壓縮鏈表，并且返回指向該鏈表的指針
unsigned char *ziplistNew(void) {
 //這里之所以+1是因為尾元素占用一個字節，這也是一個壓縮鏈表最小尺寸
 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;
}

//四個參數依次是：壓縮鏈表，插入位置（新元素插入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;
 zlentry tail;

 //1. 這段邏輯目的就是獲取前置元素的長度
 if (p[0] != ZIP_END) {
 //如果插入位置的元素不是尾元素,則獲取該元素的長度
 //這里為了后面使用方便進行了拆分，prevlensize保存encoding字段的長度，prevlen保存元素本身的長度
 ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
 } else {
 //如果插入位置的元素是尾元素，那么需要把新元素插入鏈表尾端
 //獲取到鏈表最后一個元素（注：最后一個元素不等于尾元素）
 unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);

 if (ptail[0] != ZIP_END) {
  //如果最后一個元素不是尾元素，則該元素為新元素的前置元素，獲取該元素長度
  prevlen = zipRawEntryLength(ptail);
 }
 //否則說明鏈表還沒有任何元素，即新元素的前置元素長度為0
 }

 //2. 對新元素進行編碼，獲取新元素的總大小
 if (zipTryEncoding(s,slen,&value,&encoding)) {
 //如果是數字，則按數字進行編碼
 reqlen = zipIntSize(encoding);
 } else {
 //元素長度即為字符串長度
 reqlen = slen;
 }
 //新元素總長度為值的長度加上前置元素跟encoding元素的長度
 reqlen += zipStorePrevEntryLength(NULL,prevlen);
 reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

 //如果插入的位置不是鏈表尾，則要對新元素的后續元素的prevlen字段進行判斷
 //根據上面的編碼規則，該字段可能需要擴容
 int forcelarge = 0;
 nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
 if (nextdiff == -4 && reqlen < 4) {
 nextdiff = 0;
 forcelarge = 1;
 }
 //按照新計算出的數組大小進行擴容，由于新數組的地址可能會改變，因此這里記錄舊的偏移量
 offset = p-zl;
 zl = ziplistResize(zl,curlen+reqlen+nextdiff);
 //在新數組上計算插入位置
 p = zl+offset;
 //如果新插入的元素不是在鏈表末尾
 if (p[0] != ZIP_END) {
 //把新元素后繼元素復制到新的數組中，-1為尾元素
 memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);
 //新元素的后繼元素的prevlen字段
 if (forcelarge)
  zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
 else
  zipStorePrevEntryLength(p+reqlen,reqlen);

 //更新最后一個元素的偏移量
 ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
 //當新元素的后繼元素不止有一個時，新的尾元素偏移量需要加上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 {
 //新元素插入到鏈表尾端，更新尾端偏移量
 ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
 }
 //nextdiff !=0表示后繼元素的長度發生變化，因此我們需要級聯更新后繼元素的后繼元素
 if (nextdiff != 0) {
 offset = p-zl;
 zl = __ziplistCascadeUpdate(zl,p+reqlen);
 p = zl+offset;
 }
 //把新元素寫入鏈表
 p += zipStorePrevEntryLength(p,prevlen);
 p += zipStoreEntryEncoding(p,encoding,slen);
 if (ZIP_IS_STR(encoding)) {
 memcpy(p,s,slen);
 } else {
 zipSaveInteger(p,value,encoding);
 }
 //壓縮鏈表存儲元素數量+1
 ZIPLIST_INCR_LENGTH(zl,1);
 return zl;
}

//參數依次為：壓縮鏈表，刪除元素的其實位置，刪除元素的個數
unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsigned int num) {
 unsigned int i, totlen, deleted = 0;
 size_t offset;
 int nextdiff = 0;
 zlentry first, tail;
 //讀取p指向的元素保存在first中
 zipEntry(p, &first);
 for (i = 0; p[0] != ZIP_END && i < num; i++) {
  //統計總共刪除的長度
  p += zipRawEntryLength(p);
  //統計實際刪除元素的個數
  deleted++;
 }
 //需要刪除元素的字節數
 totlen = p-first.p;
 if (totlen > 0) {
  if (p[0] != ZIP_END) {
   //判斷元素大小是否有改變
   nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
   //修改刪除元素之后的元素的prevlen字段
   p -= nextdiff;
   zipStorePrevEntryLength(p,first.prevrawlen);
   //更新末尾元素的偏移量
   ZIPLIST_TAIL_OFFSET(zl) =intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);
   //當刪除元素的后繼元素不止有一個時，新的末尾元素偏移量需要加上nextdiff
   zipEntry(p, &tail);
   if (p[tail.headersize+tail.len] != ZIP_END) {
    ZIPLIST_TAIL_OFFSET(zl) =
     intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
   }
   //把后面剩余的元素移動至前面
   memmove(first.p,p,
    intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);
  } else {
   //直接刪除到鏈表末尾，因此不需要內存拷貝，只需修改最后一個元素的偏移量
   ZIPLIST_TAIL_OFFSET(zl) =
    intrev32ifbe((first.p-zl)-first.prevrawlen);
  }
  //resize數組大小
  offset = first.p-zl;
  zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
  //修改鏈表元素個數
  ZIPLIST_INCR_LENGTH(zl,-deleted);
  p = zl+offset;
  //nextdiff != 0表示元素大小發生變化，需要進行級聯更新
  if (nextdiff != 0)
   zl = __ziplistCascadeUpdate(zl,p);
 }
 return zl;
}

//參數依次為：壓縮鏈表，要查找元素的值，要查找元素的長度，每次比較之間跳過的元素個數
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) {
  unsigned int prevlensize, encoding, lensize, len;
  unsigned char *q;
  //查詢鏈表當前元素的prevlen字段
  ZIP_DECODE_PREVLENSIZE(p, prevlensize);
  //查詢鏈表當前元素的encoding字段
  ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
  q = p + prevlensize + lensize;
  //已到達需要比較的元素位置
  if (skipcnt == 0) {
   //如果鏈表中的當前元素時字符串
   if (ZIP_IS_STR(encoding)) {
    //跟要查找的字符串進行比較
    if (len == vlen && memcmp(q, vstr, vlen) == 0) {
     //匹配成功，則要查找元素的指針
     return p;
    }
   } else {
    //如果當前元素為數字且vencoding為0
    if (vencoding == 0) {
     //嘗試對要查找的值進行數字編碼
     if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {
      //如果編碼失敗，則說明要查找的元素根本不是數字
      //然后把vencoding設置為最大值，起一個標記作用
      //也就是說后面就不用嘗試把要查找的值編碼成數字了
      vencoding = UCHAR_MAX;
     }
     assert(vencoding);
    }
    //如果vencoding != UCHAR_MAX，則說明要查找的元素成功編碼為數字
    if (vencoding != UCHAR_MAX) {
     //按數字取出當前鏈表中的元素
     long long ll = zipLoadInteger(q, encoding);
     if (ll == vll) {
      //如果兩個數字相等，則返回元素指針
      return p;
     }
    }
   }
   //重置需要跳過的元素個數
   skipcnt = skip;
  } else {
   //跳過元素
   skipcnt--;
  }
  //遍歷下個元素
  p = q + len;
 }
 //遍歷完整個鏈表，沒有找到元素
 return NULL;
}