? ? ? ? 本文介紹的是Redis中Zipmap的原理和實(shí)現(xiàn)。（轉(zhuǎn)載請指明出于breaksoftware的csdn博客）

基礎(chǔ)結(jié)構(gòu)

? ? ? ? Zipmap是為了實(shí)現(xiàn)保存Pair(String,String)數(shù)據(jù)的結(jié)構(gòu)，該結(jié)構(gòu)包含一個頭信息、一系列字符串對（之后把一個“字符串對”稱為一個“元素”（ELE））和一個尾標(biāo)記。用圖形表示該結(jié)構(gòu)就是：

? ? ? ? Redis源碼中并沒有使用結(jié)構(gòu)體來表達(dá)該結(jié)構(gòu)。因為這個結(jié)構(gòu)在內(nèi)存中是連續(xù)的，而除了HEAD和紅色背景的尾標(biāo)記END（恒定是0xFF）是固定的8位，其他部分都是不定長的。

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

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

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

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

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

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

#define ZIPMAP_VALUE_MAX_FREE 4

? ? ? ? 明白上述結(jié)構(gòu)后，閱讀后面的代碼就變得很簡單了。

?

創(chuàng)建Zipmap

? ? ? ? Redis提供了下面方法創(chuàng)建一個空的Zipmap結(jié)構(gòu)。

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

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

長度信息編碼

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

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);}}
}

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

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

長度信息解碼

? ? ? ? 長度信息解碼是對編碼做的逆向操作。它判斷傳入的長度信息起始地址的內(nèi)容是否小于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內(nèi)容的整體長度

? ? ? ? 對應(yīng)于上圖就是計算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內(nèi)容的整體長度

? ? ? ? 對應(yīng)于上圖就是計算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參數(shù)第一次賦值時，它代表ValueLen Struct的長度。于是p[used]取出的就是Free的內(nèi)容，即FreeData的長度。計算和的操作中再加上1，是Free字段的長度——sizeof(char)。

計算元素長度

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

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

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

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

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

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

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

查找元素、計算Zipmap總長

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

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對應(yīng)的元素首地址。如果totlen不為NULL，則會順帶計算出Zipmap結(jié)構(gòu)的總長。一開始時，p在Zipmap首地址上前進(jìn)8位是為了過掉HEAD結(jié)構(gòu)，直接指向元素首地址。相似的方法還有

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

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

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

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

        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相關(guān)信息的長度。然后跳到下個元素的首地址。

        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 */}

? ? ? ? 最后根據(jù)totlen是否為NULL，計算Zipmap總長。計算時加1是為了把之前過掉HEAD結(jié)構(gòu)的長度給補(bǔ)上。

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

? ? ? ? Zipmap還提供了下面的方法計算結(jié)構(gòu)總長，當(dāng)然它只是對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;
}

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

遍歷Zipmap

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

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操作實(shí)則是過掉Free占用的一個字節(jié)。

? ? ? ? 調(diào)用上面函數(shù)遍歷的方法是：

  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);}

獲取元素個數(shù)

? ? ? ? 之前我們介紹基礎(chǔ)結(jié)構(gòu)時說過。如果元素個數(shù)少于0xFE，則結(jié)構(gòu)首地址保存的數(shù)值就是元素個數(shù)值。如果大于等于0xFE，則要遍歷整個結(jié)構(gòu)

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空間

? ? ? ? 重分配操作除了重新分配空間外，還要將結(jié)尾符設(shè)置。

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);

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

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

? ? ? ? 接著重分配Zipmap結(jié)構(gòu)的空間，以節(jié)約空間。zipmapResize方法還輔助性的把結(jié)尾符給設(shè)置上了。

        zm = zipmapResize(zm, zmlen-freelen);

? ? ? ? 接著判斷元素個數(shù)是否在0xFE之內(nèi)。如果在，則讓Zipmap結(jié)構(gòu)首地址對應(yīng)的數(shù)值減少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對應(yīng)的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的結(jié)構(gòu)總長

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

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

    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空間。并把原來元素之后的內(nèi)容向后平移。

      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;}}

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

    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;}

? ? ? ? 如果空余出來的空間很短，就不要做內(nèi)存重分配的工作了。這樣就是之前結(jié)構(gòu)中Free和FreeData的由來。這也說明FreeData中的數(shù)據(jù)是不確定的——即它是之前內(nèi)容的一部分。

? ? ? ? 最后將Key，Value和Free字段放入相應(yīng)的空間中

    /* 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;
}

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。