Redis 的底層數據結構（壓縮列表）

• 2019 年 11 月 13 日
• 筆記

上一篇我們介紹了 redis 中的整數集合這種數據結構的實現，也談到了，引入這種數據結構的一個很大的原因就是，在某些僅有少量整數元素的集合場景，通過整數集合既可以達到字典的效率，也能使用遠少於字典的內存達到同樣的效果。

我們本篇介紹的壓縮列表，相信你從他的名字里應該也能看出來，又是一個為了節約內存而設計的數據結構，它的數據結構相對於整數集合來說會複雜了很多，但是整數集合只能允許存儲少量的整型數據，而我們的壓縮列表可以允許存儲少量的整型數據或字符串。

這是他們之間的一個區別，下面我們來看看這種數據結構。

一、基本的結構定義

• ZIPLIST_BYTES：四個位元組，記錄了整個壓縮列表總共佔用了多少位元組數
• ZIPLIST_TAIL_OFFSET：四個位元組，記錄了整個壓縮列表第一個節點到最後一個節點跨越了多少個位元組，通故這個字段可以迅速定位到列表最後一個節點位置
• ZIPLIST_LENGTH：兩個位元組，記錄了整個壓縮列表中總共包含幾個 zlentry 節點
• zlentry：非固定位元組，記錄的是單個節點，這是一個複合結構，我們等下再說
• 0xFF：一個位元組，十進制的值為 255，標誌壓縮列表的結尾

其中，zlentry 在 redis 中確實有着這樣的結構體定義，但實際上這個結構定義了一堆類似於 length 這樣的字段，記錄前一個節點和自身節點佔用的位元組數等等信息，用處不多，而我們更傾向於使用這樣的邏輯結構來描述 zlentry 節點。

這種結構在 redis 中是沒有具體結構體定義的，請知悉，網上的很多博客文章都直接描述 zlentry 節點是這樣的一種結構，其實是不準確的。

簡單解釋一下這三個字段的含義：

• previous_entry_length：每個節點會使用一個或者五個位元組來描述前一個節點佔用的總位元組數，如果前一個節點佔用的總位元組數小於 254，那麼就用一個位元組存儲，反之如果前一個節點佔用的總位元組數超過了 254，那麼一個位元組就不夠存儲了，這裡會用五個位元組存儲並將第一個位元組的值存儲為固定值 254 用於區分。
• encoding：壓縮列表可以存儲 16位、32位、64位的整數以及字符串，encoding 就是用來區分後面的 content 字段中存儲於的到底是哪種內容，分別佔多少位元組，這個我們等下細說。
• content：沒什麼特別的，存儲的就是具體的二進制內容，整數或者字符串。

下面我們細說一個 encoding 具體是怎麼存儲的。

主要分為兩種，一種是字符串的存儲格式：

content 的具體長度，由編碼除去高兩位剩餘的二進制位表示。

注意，整型數據的編碼是固定 11 開頭的八位二進制，而字符串類型的編碼都是非固定的，因為它還需要通過後面的二進制位得到字符串的長度，稍有區別。

這就是壓縮列表的基本的結構定義情況，下面我們通過節點的增刪改查方法源碼實現來看看 redis 中具體的實現情況。

二、redis 的具體源碼實現

1、ziplistNew

我們先來看看壓縮列表初始化的方法實現：

unsigned char *ziplistNew(void) {      //bytes=2*4+2      //分配壓縮列表結構所需要的位元組數      //ZIPLIST_BYTES + ZIPLIST_TAIL_OFFSET + ZIPLIST_LENGTH      unsigned int bytes = ZIPLIST_HEADER_SIZE+1;      unsigned char *zl = zmalloc(bytes);      //初始化 ZIPLIST_BYTES 字段      ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);      //初始化 ZIPLIST_TAIL_OFFSET      ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);      //初始化 ZIPLIST_LENGTH 字段      ZIPLIST_LENGTH(zl) = 0;      //為壓縮列表最後一個位元組賦值 255      zl[bytes-1] = ZIP_END;      return zl;  }

2、ziplistPush

接着我們看新增節點的源碼實現：

unsigned char *ziplistPush(unsigned char *zl, unsigned char *s          ,unsigned int slen, int where) {      unsigned char *p;      //找到待插入的位置，頭部或者尾部      p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);      return __ziplistInsert(zl,p,s,slen);  }

解釋一下 ziplistPush 的幾個入參的含義。

zl 指向一個壓縮列表的首地址，s 指向一個字符串首地址），slen 指向字符串的長度（如果節點存儲的值是整型，存儲的就是整型值），where 指明新節點的插入方式，頭插亦或尾插。

ziplistPush 方法的核心是 __ziplistInsert：

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;      //prevlensize 存儲前一個節點長度，本節點使用了幾個位元組 1 or 5      //prelen  存儲前一個節點實際佔用了幾個位元組      if (p[0] != ZIP_END) {          ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);      } else {          unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);          if (ptail[0] != ZIP_END) {              prevlen = zipRawEntryLength(ptail);          }      }      if (zipTryEncoding(s,slen,&value,&encoding)) {          //s 指針指向一個整數，嘗試進行一個轉換並得到存儲這個整數佔用了幾個位元組          reqlen = zipIntSize(encoding);      } else {          //s 指針指向一個字符串（字符數組），slen 就是他佔用的位元組數          reqlen = slen;      }      //當前節點存儲數據佔用 reqlen 個位元組，加上存儲前一個節點長度佔用的位元組數      reqlen += zipStorePrevEntryLength(NULL,prevlen);      //encoding 字段存儲實際佔用位元組數      reqlen += zipStoreEntryEncoding(NULL,encoding,slen);      //至此，reqlen 保存了存儲當前節點數據佔用位元組數和 encoding 編碼佔用的位元組數總和      int forcelarge = 0;      //當前節點佔用的總位元組減去存儲前一個節點字段佔用的位元組      //記錄的是這一個節點的插入會引起下一個節點佔用位元組的變化量      nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;      if (nextdiff == -4 && reqlen < 4) {          nextdiff = 0;          forcelarge = 1;      }      //擴容有可能導致 zl 的起始位置偏移，故記錄 p 與 zl 首地址的相對偏差數，事後還原 p 指針指向      offset = p-zl;      zl = ziplistResize(zl,curlen+reqlen+nextdiff);      p = zl+offset;      if (p[0] != ZIP_END) {          memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);          //把當前節點佔用的位元組數存儲到下一個節點的頭部字段          if (forcelarge)              zipStorePrevEntryLengthLarge(p+reqlen,reqlen);          else              zipStorePrevEntryLength(p+reqlen,reqlen);            //更新 tail_offset 字段，讓他保存從頭節點到尾節點之間的距離          ZIPLIST_TAIL_OFFSET(zl) =              intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);          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);      }      //是否觸發連鎖更新      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);      }      ZIPLIST_INCR_LENGTH(zl,1);      return zl;  }

具體細節我不再贅述，總結一下整個插入節點的步驟。

1. 計算並得到前一個節點的總長度，並判斷得到當前待插入節點保存前一個節點長度的 previous_entry_length 佔用位元組數
2. 根據傳入的 s 和 slen，計算並保存 encoding 字段內容
3. 構建節點並將數據寫入節點添加到壓縮列表中

ps：重點要去理解壓縮列表節點的數據結構定義，previous_entry_length、encoding、content 字段，這樣才能比較容易理解節點新增操作的實現。

三、連鎖更新

談到 redis 的壓縮列表，就必然會談到他的連鎖更新，我們先引一張圖：

假設原本 entry1 節點佔用位元組數為 211（小於 254），那麼 entry2 的 previous_entry_length 會使用一個位元組存儲 211，現在我們新插入一個節點 NEWEntry，這個節點比較大，佔用了 512 個位元組。

那麼，我們知道，NEWEntry 節點插入後，entry2 的 previous_entry_length 存儲不了 512，那麼 redis 就會重分配內存，增加 entry2 的內存分配，並分配給 previous_entry_length 五個位元組存儲 NEWEntry 節點長度。

看似沒什麼問題，但是如果極端情況下，entry2 擴容四個位元組後，導致自身佔用位元組數超過 254，就會又觸發後一個節點的內存佔用空間擴大，非常極端情況下，會導致所有的節點都擴容，這就是連鎖更新，一次更新導致大量甚至全部節點都更新內存的分配。

如果連鎖更新發生的概率很高的話，壓縮列表無疑就會是一個低效的數據結構，但實際上連鎖更新發生的條件是非常苛刻的，其一是需要大量節點長度小於 254 連續串聯連接，其二是我們更新的節點位置恰好也導致後一個節點內存擴充更新。

基於這兩點，且少量的連鎖更新對性能是影響不大的，所以這裡的連鎖更新對壓縮列表的性能是沒有多大的影響的，可以忽略，但需要知曉。

同樣的，如果覺得我寫的對你有點幫助的話，順手點一波關注吧，也歡迎加作者微信深入探討，我們逐漸開始走近 redis 比較實用性的相關內容了，盡請關注。