? ? ? ? 本文介紹的是Redis中Zipmap的原理和實現。（轉載請指明出于breaksoftware的csdn博客）

基礎結構

? ? ? ? Zipmap是為了實現保存Pair(String,String)數據的結構，該結構包含一個頭信息、一系列字符串對（之后把一個“字符串對”稱為一個“元素”（ELE））和一個尾標記。用圖形表示該結構就是：

? ? ? ? Redis源碼中并沒有使用結構體來表達該結構。因為這個結構在內存中是連續的，而除了HEAD和紅色背景的尾標記END（恒定是0xFF）是固定的8位，其他部分都是不定長的。

? ? ? ? 雖然HEAD信息是定長的，但是其內容表達卻是兩層意思。HEAD信息是用于保存元素個數信息的。比如該Zipmap只有0x12個元素，則HEAD的內容是0x12。而如果元素個數是0x1234，則內容是0xFE。那這種變化的分界嶺是多少？是0xFE。如果元素個數小于0xFE，HEAD內容就是個數值。如果元素個數大于等于0xFE，HEAD內容是0xFE，表示該8位已經不表示元素個數，這個是否如果需要計算元素個數就需要遍歷整個結構了。

? ? ? ? 元素的長度也是不確定的。這點比較好理解，因為元素保存的是一個字符串對，而字符串長度是無法確定的。那么我們再看看元素的結構

? ? ? ? 元素內容中一開始時記錄的Key的長度信息——KeyLen Struct。如果長度小于0xFE，則該結構只有8位長，內容是長度值；如果大于等于0xFE，則該結構是40位長。其中前8位是0xFE，表示本位已經不能表示長度了。后32位則保存長度值

? ? ? ? KeyData包含字符串的內容，其內容可以包含NULL，但是不會自動在末尾追加NULL。該規則也符合ValueData。

? ? ? ? Key內容之后跟著的是Value的長度信息。其組織方式和Key長度信息的組織方式是一樣的。

? ? ? ? 最后再來說說比較神奇的Free字段。因為Zipmap會提供一個接口，讓用戶可以通過Key去修改Value的值，如果Value的值被改短了，則會有一定的空余空間，這個空余空間的長度就是Free的值。但是Zipmap就給Free字段留了8位的空間，然而Value修改后空余的長度可能比0xFF還長。其實不用擔心，因為如果Zipmap發現如果空余長度超過一定的值就會將之后的空間向前平移以節約空間，而這個閾值比0xFF小

#define ZIPMAP_VALUE_MAX_FREE 4

? ? ? ? 明白上述結構后，閱讀后面的代碼就變得很簡單了。

?

創建Zipmap

? ? ? ? Redis提供了下面方法創建一個空的Zipmap結構。

unsigned char *zipmapNew(void) {unsigned char *zm = zmalloc(2);zm[0] = 0; /* Length */zm[1] = ZIPMAP_END;return zm;
}

? ? ? ? 因為沒有元素，所以長度信息為0。而緊跟其后的便是結尾標記

長度信息編碼

? ? ? ? Zipmap中元素的Key長度信息和Value長度信息都是需要根據值的大小而動態改變。如果值小于0xFE，則只有8位表示長度，且內容就是長度值。而如果長度大于0xFE，則有40位表示長度信息，其前8位內容是0xFE，后32位是值內容。

static unsigned int zipmapEncodeLength(unsigned char *p, unsigned int len) {if (p == NULL) {return ZIPMAP_LEN_BYTES(len);} else {if (len < ZIPMAP_BIGLEN) {p[0] = len;return 1;} else {p[0] = ZIPMAP_BIGLEN;memcpy(p+1,&len,sizeof(len));memrev32ifbe(p+1);return 1+sizeof(len);}}
}

? ? ? ? 如果第一個參數傳NULL，則該函數是用于根據第二個參數決定需要多長的空間去存儲長度信息。如果第一個參數有值，則根據第二個參數決定在該地址的不同偏移位置設置相應的值。

#define ZIPMAP_BIGLEN 254
#define ZIPMAP_END 255
#define ZIPMAP_LEN_BYTES(_l) (((_l) < ZIPMAP_BIGLEN) ? 1 : sizeof(unsigned int)+1)

長度信息解碼

? ? ? ? 長度信息解碼是對編碼做的逆向操作。它判斷傳入的長度信息起始地址的內容是否小于0xFE。如果是則該8位就是長度值；否則后移8位，之后的32位才是長度值。

static unsigned int zipmapDecodeLength(unsigned char *p) {unsigned int len = *p;if (len < ZIPMAP_BIGLEN) return len;memcpy(&len,p+1,sizeof(unsigned int));memrev32ifbe(&len);return len;
}

計算Key長度信息和Key內容的整體長度

? ? ? ? 對應于上圖就是計算KeyLen Struct和KeyData的長度和

? ? ? ? 計算方法就是通過獲取的KeyData長度計算出KeyLen Struct的長度，然后將兩個長度相加

static unsigned int zipmapRawKeyLength(unsigned char *p) {unsigned int l = zipmapDecodeLength(p);return zipmapEncodeLength(NULL,l) + l;
}

計算Value長度信息，Free和Value內容的整體長度

? ? ? ? 對應于上圖就是計算ValueLen Struct、Free、FreeData和ValueData的長度和

static unsigned int zipmapRawValueLength(unsigned char *p) {unsigned int l = zipmapDecodeLength(p);unsigned int used;used = zipmapEncodeLength(NULL,l);used += p[used] + 1 + l;return used;
}

? ? ? ? used參數第一次賦值時，它代表ValueLen Struct的長度。于是p[used]取出的就是Free的內容，即FreeData的長度。計算和的操作中再加上1，是Free字段的長度——sizeof(char)。

計算元素長度

? ? ? ? 只要把上面兩個方法一疊加便是元素長度。唯一要做變通的是計算Value相關長度時要讓指針指向Value信息的首地址

static unsigned int zipmapRawEntryLength(unsigned char *p) {unsigned int l = zipmapRawKeyLength(p);return l + zipmapRawValueLength(p+l);
}

通過Key和Value長度計算保存該元素時需要的最短長度

? ? ? ? 因為KeyLen Struct和ValueLen Struct的最低長度是一個字節，Free字段占一個字節。于是至少需要如下算法的長度

static unsigned long zipmapRequiredLength(unsigned int klen, unsigned int vlen) {unsigned int l;l = klen+vlen+3;

? ? ? ? 但是如果Key或Value的長度大于等于0xFE，則還需要4字節表示真實長度

    if (klen >= ZIPMAP_BIGLEN) l += 4;if (vlen >= ZIPMAP_BIGLEN) l += 4;return l;
}

查找元素、計算Zipmap總長

? ? ? ? Zipmap提供了一個方法來完成兩個功能。因為這兩種方法都需要進行遍歷操作，所以索性放在一塊。

static unsigned char *zipmapLookupRaw(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned int *totlen) {unsigned char *p = zm+1, *k = NULL;unsigned int l,llen;

? ? ? ? 如果Key字段指向一個待對比的字符串，則之后會通過對比查找Key對應的元素首地址。如果totlen不為NULL，則會順帶計算出Zipmap結構的總長。一開始時，p在Zipmap首地址上前進8位是為了過掉HEAD結構，直接指向元素首地址。相似的方法還有

unsigned char *zipmapRewind(unsigned char *zm) {return zm+1;
}

? ? ? ? 之后要獲取Key長度。如果Key長度和傳入的klen不一樣，則可以肯定的是Key不同，則不需要進行之后的字符串對比。著算是一種優化

    while(*p != ZIPMAP_END) {unsigned char free;/* Match or skip the key */l = zipmapDecodeLength(p);llen = zipmapEncodeLength(NULL,l);

? ? ? ? 對比操作要看Key長度和內容是否都一致。如果一致，則視totlen是否為NULL，決定是否繼續遍歷。因為totlen為NULL，則說明不需要計算Zipmap總長，這個時候直接返回元素首地址即可。如果不為NULL，則要記錄下當前找到的元素首地址到k變量中，這樣在之后遍歷中，就可以通過變量k知道元素已經找到，不需要進行對比操作了。

        if (key != NULL && k == NULL && l == klen && !memcmp(p+llen,key,l)) {/* Only return when the user doesn't care* for the total length of the zipmap. */if (totlen != NULL) {k = p;} else {return p;}}

? ? ? ? 如果沒有匹配上，則計算Value相關信息的長度。然后跳到下個元素的首地址。

        p += llen+l;/* Skip the value as well */l = zipmapDecodeLength(p);p += zipmapEncodeLength(NULL,l);free = p[0];p += l+1+free; /* +1 to skip the free byte */}

? ? ? ? 最后根據totlen是否為NULL，計算Zipmap總長。計算時加1是為了把之前過掉HEAD結構的長度給補上。

    if (totlen != NULL) *totlen = (unsigned int)(p-zm)+1;return k;
}

? ? ? ? Zipmap還提供了下面的方法計算結構總長，當然它只是對zipmapLookupRaw的封裝

size_t zipmapBlobLen(unsigned char *zm) {unsigned int totlen;zipmapLookupRaw(zm,NULL,0,&totlen);return totlen;
}

檢測元素是否存在

? ? ? ? 檢測的方法也是封裝了zipmapLookupRaw，然后判斷其是否找到元素的首地址

int zipmapExists(unsigned char *zm, unsigned char *key, unsigned int klen) {return zipmapLookupRaw(zm,key,klen,NULL) != NULL;
}

通過Key獲取Value值

? ? ? ? 首先通過zipmapLookupRaw方法確定該Key在Zipmap中。如果不存在則返回NULL，如果存在則讓指針指向Value信息處

int zipmapGet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char **value, unsigned int *vlen) {unsigned char *p;if ((p = zipmapLookupRaw(zm,key,klen,NULL)) == NULL) return 0;p += zipmapRawKeyLength(p);*vlen = zipmapDecodeLength(p);*value = p + ZIPMAP_LEN_BYTES(*vlen) + 1;return 1;
}

? ? ? ? 注意這個函數也返回了Value的長度，所以說Value可以存儲包含NULL的數據。

遍歷Zipmap

? ? ? ? 遍歷前需要調用zipmapRewind讓指針指向元素首地址，然后調用下面方法

unsigned char *zipmapNext(unsigned char *zm, unsigned char **key, unsigned int *klen, unsigned char **value, unsigned int *vlen) {if (zm[0] == ZIPMAP_END) return NULL;if (key) {*key = zm;*klen = zipmapDecodeLength(zm);*key += ZIPMAP_LEN_BYTES(*klen);}zm += zipmapRawKeyLength(zm);if (value) {*value = zm+1;*vlen = zipmapDecodeLength(zm);*value += ZIPMAP_LEN_BYTES(*vlen);}zm += zipmapRawValueLength(zm);return zm;
}

? ? ? ? 上述方法通過用于接收Key和Value的指針是否為NULL決定是否需要把它們的信息返回出去。其中第10行加1操作實則是過掉Free占用的一個字節。

? ? ? ? 調用上面函數遍歷的方法是：

  unsigned char *i = zipmapRewind(my_zipmap);while((i = zipmapNext(i,&key,&klen,&value,&vlen)) != NULL) {printf("%d bytes key at $p\n", klen, key);printf("%d bytes value at $p\n", vlen, value);}

獲取元素個數

? ? ? ? 之前我們介紹基礎結構時說過。如果元素個數少于0xFE，則結構首地址保存的數值就是元素個數值。如果大于等于0xFE，則要遍歷整個結構

unsigned int zipmapLen(unsigned char *zm) {unsigned int len = 0;if (zm[0] < ZIPMAP_BIGLEN) {len = zm[0];} else {unsigned char *p = zipmapRewind(zm);while((p = zipmapNext(p,NULL,NULL,NULL,NULL)) != NULL) len++;/* Re-store length if small enough */if (len < ZIPMAP_BIGLEN) zm[0] = len;}return len;
}

重分配Zipmap空間

? ? ? ? 重分配操作除了重新分配空間外，還要將結尾符設置。

static inline unsigned char *zipmapResize(unsigned char *zm, unsigned int len) {zm = zrealloc(zm, len);zm[len-1] = ZIPMAP_END;return zm;
}

刪除元素

? ? ? ? 刪除元素前先需要找到該元素的起始地址

unsigned char *zipmapDel(unsigned char *zm, unsigned char *key, unsigned int klen, int *deleted) {unsigned int zmlen, freelen;unsigned char *p = zipmapLookupRaw(zm,key,klen,&zmlen);if (p) {

? ? ? ? 如果找到，則計算這個元素的長度

        freelen = zipmapRawEntryLength(p);

? ? ? ? 然后將該元素之后的內容，除了結尾符0xFE，向前移動到該元素的起始地址

        memmove(p, p+freelen, zmlen-((p-zm)+freelen+1));

? ? ? ? 接著重分配Zipmap結構的空間，以節約空間。zipmapResize方法還輔助性的把結尾符給設置上了。

        zm = zipmapResize(zm, zmlen-freelen);

? ? ? ? 接著判斷元素個數是否在0xFE之內。如果在，則讓Zipmap結構首地址對應的數值減少1

        /* Decrease zipmap length */if (zm[0] < ZIPMAP_BIGLEN) zm[0]--;if (deleted) *deleted = 1;} else {if (deleted) *deleted = 0;}return zm;
}

增加、修改元素

? ? ? ? 如果通過zipmapSet方法傳入的Key在Zipmap中，則是要求修改該Key對應的Value值；如果不在Zipmap中，則是新增元素。

? ? ? ? 首先通過Key和Value的長度計算出存儲該字符串對最少需要多少空間

unsigned char *zipmapSet(unsigned char *zm, unsigned char *key, unsigned int klen, unsigned char *val, unsigned int vlen, int *update) {unsigned int zmlen, offset;unsigned int freelen, reqlen = zipmapRequiredLength(klen,vlen);unsigned int empty, vempty;unsigned char *p;

? ? ? ? 然后判斷Key是否在Zipmap中，并計算出Zipmap的結構總長

    freelen = reqlen;if (update) *update = 0;p = zipmapLookupRaw(zm,key,klen,&zmlen);

? ? ? ? 如果Key不存在，則需要新增元素。這個時候需要重新給Zipmap分配更大的空間，并將需要新增元素的位置指針指向結尾符——即在尾部追加元素。還要讓元素個數自增

    if (p == NULL) {/* Key not found: enlarge */zm = zipmapResize(zm, zmlen+reqlen);p = zm+zmlen-1;zmlen = zmlen+reqlen;/* Increase zipmap length (this is an insert) */if (zm[0] < ZIPMAP_BIGLEN) zm[0]++;}

? ? ? ? 如果Key存在，則是更新Value。這個時候需要計算找到的元素的總長，如果總長比修改后需要的最短長度要短，則需要重新分配Zipmap空間。并把原來元素之后的內容向后平移。

      else {/* Key found. Is there enough space for the new value? *//* Compute the total length: */if (update) *update = 1;freelen = zipmapRawEntryLength(p);if (freelen < reqlen) {/* Store the offset of this key within the current zipmap, so* it can be resized. Then, move the tail backwards so this* pair fits at the current position. */offset = p-zm;zm = zipmapResize(zm, zmlen-freelen+reqlen);p = zm+offset;/* The +1 in the number of bytes to be moved is caused by the* end-of-zipmap byte. Note: the *original* zmlen is used. */memmove(p+reqlen, p+freelen, zmlen-(offset+freelen+1));zmlen = zmlen-freelen+reqlen;freelen = reqlen;}}

? ? ? ? 如果當前元素的總長度比修改后需要的最短長度要長，則說明是Value字符串長度要變短。這個時候要計算空余出來的空間是否大于ZIPMAP_VALUE_MAX_FREE，如果大于則要縮減Zipmap結構空間。

    empty = freelen-reqlen;if (empty >= ZIPMAP_VALUE_MAX_FREE) {/* First, move the tail <empty> bytes to the front, then resize* the zipmap to be <empty> bytes smaller. */offset = p-zm;memmove(p+reqlen, p+freelen, zmlen-(offset+freelen+1));zmlen -= empty;zm = zipmapResize(zm, zmlen);p = zm+offset;vempty = 0;} else {vempty = empty;}

? ? ? ? 如果空余出來的空間很短，就不要做內存重分配的工作了。這樣就是之前結構中Free和FreeData的由來。這也說明FreeData中的數據是不確定的——即它是之前內容的一部分。

? ? ? ? 最后將Key，Value和Free字段放入相應的空間中

    /* Just write the key + value and we are done. *//* Key: */p += zipmapEncodeLength(p,klen);memcpy(p,key,klen);p += klen;/* Value: */p += zipmapEncodeLength(p,vlen);*p++ = vempty;memcpy(p,val,vlen);return zm;
}

總結

以上是生活随笔為你收集整理的Redis源码解析——Zipmap的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。