此文已由作者刘超授权网易云社区发布。

欢迎访问网易云社区，了解更多网易技术产品运营经验

下面应该对编码区域进行压缩了，在大多数的实现中，压缩代码多少有些晦涩难懂。一般来说，会对每一种b有一个代码实现，在这里我们举例列出的是b=5的代码实现。

整个过程我们可以想象成codeSection是一条条32位大小的袋子，而codes是一系列待压缩的32位的物品，其中货真价实的就5位，其他都是水分(都是0)，下面要做的事情就是把待压缩的物品一件件拿出来，把有水分挤掉，然后往袋子里面装。

装的时候就面临一个问题，32不是5的整数倍，放6个还空着2位，放7个不够空间，这循环怎么写啊？所以只能以最小公倍数32*5=160位为一个处理批次，放在一个循环里面，也即每个循环处理5个袋子，32个物品，使得32个物品正好能放在5个袋子里面。

//bwidth=5
private static int pack(int[] codes, int numOfCodes, int[] codeSection,
int curCode, int bWidth) {
int cur = 0;
// suppose bwidth = 5
// bwidth不一定能被32的整除，所以每32个一组，保证处理完以后，32*bwidth个bit，一定是字对齐的。
while (cur < numOfCodes) {
codeSection[curCode + 0] = 0;
codeSection[curCode + 1] = 0;
codeSection[curCode + 2] = 0;
codeSection[curCode + 3] = 0;
codeSection[curCode + 4] = 0;
//curCode + 0是第一个袋子，先放codes里面从cur+0到cur+5六个物品后，还空着2位，于是把第七个物品前2位截出来，放进去。0x18二进制11000，作用就是最后5位保留前两位，然后右移3位，就把前2位放到了袋子的最后2位。
codeSection[curCode + 0] |= codes[cur + 0] << (32 - 5);
codeSection[curCode + 0] |= codes[cur + 1] << (32 - 10);
codeSection[curCode + 0] |= codes[cur + 2] << (32 - 15);
codeSection[curCode + 0] |= codes[cur + 3] << (32 - 20);
codeSection[curCode + 0] |= codes[cur + 4] << (32 - 25);
codeSection[curCode + 0] |= codes[cur + 5] << (32 - 30);
codeSection[curCode + 0] |= (codes[cur + 6] & 0x18) >> 3;
//curCode+1是第二个袋子。刚才第七个物品前2位被截了放在第一个袋子里，那么首先剩下的3位放在第二个袋子的开头，0x07就是00111，也就是截取后三位。然后再放5个物品，还空着4位，于是第十三个物品截取前四位(0x1E二进制11110)。
codeSection[curCode + 1] |= (codes[cur + 6] & 0x07) << (32 - 3);
codeSection[curCode + 1] |= codes[cur + 7] << (32 - 3 - 5);
codeSection[curCode + 1] |= codes[cur + 8] << (32 - 3 - 10);
codeSection[curCode + 1] |= codes[cur + 9] << (32 - 3 - 15);
codeSection[curCode + 1] |= codes[cur + 10] << (32 - 3 - 20);
codeSection[curCode + 1] |= codes[cur + 11] << (32 - 3 - 25);
codeSection[curCode + 1] |= (codes[cur + 12] & 0x1E) >> 1;
//curCode + 2第三个袋子。先放第十三个物品剩下的1位(0x01二进制00001)，然后再放入6个物品，最后空着1位。将第二十个物品的第1位截取出来(0x10二进制10000)放入。
codeSection[curCode + 2] |= (codes[cur + 12] & 0x01) << (32 - 1);
codeSection[curCode + 2] |= codes[cur + 13] << (32 - 1 - 5);
codeSection[curCode + 2] |= codes[cur + 14] << (32 - 1 - 10);
codeSection[curCode + 2] |= codes[cur + 15] << (32 - 1 - 15);
codeSection[curCode + 2] |= codes[cur + 16] << (32 - 1 - 20);
codeSection[curCode + 2] |= codes[cur + 17] << (32 - 1 - 25);
codeSection[curCode + 2] |= codes[cur + 18] << (32 - 1 - 30);
codeSection[curCode + 2] |= (codes[cur + 19] & 0x10) >> 4;
//curCode + 3第四个袋子。先放第二十个物品剩下的4位(0x0F二进制位01111)。然后放5个物品，最后还空着3位。将第二十六个物品截取3位放入(0x1C二进制11100)。
codeSection[curCode + 3] |= (codes[cur + 19] & 0x0F) << (32 - 4);
codeSection[curCode + 3] |= codes[cur + 20] << (32 - 4 - 5);
codeSection[curCode + 3] |= codes[cur + 21] << (32 - 4 - 10);
codeSection[curCode + 3] |= codes[cur + 22] << (32 - 4 - 15);
codeSection[curCode + 3] |= codes[cur + 23] << (32 - 4 - 20);
codeSection[curCode + 3] |= codes[cur + 24] << (32 - 4 - 25);
codeSection[curCode + 3] |= (codes[cur + 25] & 0x1C) >> 2;
//curCode + 4第五个袋子。先放第二十六个物品剩下的2位。最后这个袋子还剩30位，正好放下6个物品。
codeSection[curCode + 4] |= (codes[cur + 25] & 0x03) << (32 - 2);
codeSection[curCode + 4] |= codes[cur + 26] << (32 - 2 - 5);
codeSection[curCode + 4] |= codes[cur + 27] << (32 - 2 - 10);
codeSection[curCode + 4] |= codes[cur + 28] << (32 - 2 - 15);
codeSection[curCode + 4] |= codes[cur + 29] << (32 - 2 - 20);
codeSection[curCode + 4] |= codes[cur + 30] << (32 - 2 - 25);
codeSection[curCode + 4] |= codes[cur + 31] << (32 - 2 - 30);
//处理下一组
cur += 32;
curCode += 5;
}
int numOfWords = (int) Math.ceil((double) (numOfCodes * 5) / 32.0);
return numOfWords;
}

经过压缩后，整个block的格式如下，整个block被组织成int数组，[]里面的数值为下标：

Header：这部分只占用32位，在这里包含四部分，第一部分5位，表示b=5。第二部分表示Entry部分的长度，占用了3个32位，也即有三个Entry。第三部分表示编码区域的长度，占用了42个

Entry列表：包含3个entry，每个entry占用32位，前7位表示第一个异常位置，后25位表示这个entry在异常区域中的起始位置。

编码区域。总共占用42个int，每5个int可以存放32个压缩后的编码(每个编码占用5位)。第三个entry共占用两个int，保存了11个数值占用55位，另外人为补充9个0.

异常区域。在块中，异常区域是从后向前延伸的。其中从74到60的红色部分属于Entry 1，从59到50的黄色部分属于Entry 2，绿色部分属于Entry 3。

当需要读取这个快的时候，便需要对这个块进行解码。

首先通过解析Header来得到全局信息。

接下来需要读取Entry列表，来解析每个Entry中的信息。

然后对于每个Entry进行解码，代码如下：

//解析entry，得到全局信息
int entrycode = entries[i];
int firstExceptPosition = entrycode >> 25;
int curException = entrycode & 0x1FFFFFF;
//进行解压缩，将编码区域5位一个数值解压为一个int数组。Codes就是解压后的数组，tempCodeSection指向编码区域这个Entry的起始位置，numOfCodes是需要解压的数值的个数，bwidth=5.
Unpack(codes, numOfCodes, tempCodeSection, bwidth);
//第一个循环将异常数值还原
int cur = firstExceptPosition;
while (cur < numOfCodes && curException >= lastException) {
int jump = codes[cur];
codes[cur] = block[curException--];
cur = cur + jump + 1;
}
//第二个循环输出结果并且跳过人为添加的异常数值
for (int j = 0; j < codes.length; j++) {
if (codes[j] > 0) {
output[curOutput++] = codes[j];
}
}

对编码区域的解压方式也正好是压缩方式的逆向过程。是从袋子里面将物品拿出来的时候了。

private static void Unpack(int[] codes, int numberOfCodes, int[] codeSection,
int bwidth) {
int cur = 0;
int curCode = 0;
while (cur < numberOfCodes) {
//从第一个袋子中，拿出6个物品，还剩2位。
codes[cur + 0] = (codeSection[curCode + 0] >> (32 - 5)) & 0x1F;
codes[cur + 1] = (codeSection[curCode + 0] >> (32 - 10)) & 0x1F;
codes[cur + 2] = (codeSection[curCode + 0] >> (32 - 15)) & 0x1F;
codes[cur + 3] = (codeSection[curCode + 0] >> (32 - 20)) & 0x1F;
codes[cur + 4] = (codeSection[curCode + 0] >> (32 - 25)) & 0x1F;
codes[cur + 5] = (codeSection[curCode + 0] >> (32 - 30)) & 0x1F;
codes[cur + 6] = (codeSection[curCode + 0] << 3) & 0x18;
//第一个袋子中的2位和第二个袋子中的前3位组成第7个物品。然后接着从第二个袋子中拿出5个物品，剩下4位。
codes[cur + 6] |= (codeSection[curCode + 1] >> (32 - 3)) & 0x07;
codes[cur + 7] = (codeSection[curCode + 1] >> (32 - 3 - 5)) & 0x1F;
codes[cur + 8] = (codeSection[curCode + 1] >> (32 - 3 - 10)) & 0x1F;
codes[cur + 9] = (codeSection[curCode + 1] >> (32 - 3 - 15)) & 0x1F;
codes[cur + 10] = (codeSection[curCode + 1] >> (32 - 3 - 20)) & 0x1F;
codes[cur + 11] = (codeSection[curCode + 1] >> (32 - 3 - 25)) & 0x1F;
codes[cur + 12] = (codeSection[curCode + 1] << 1) & 0x1E;
//第二个袋子的最后4位和第三个袋子的前1位组成一个物品，然后在第三个袋子里面拿出6个物品，剩下1位。
codes[cur + 12] |= (codeSection[curCode + 2] >> (32 - 1)) & 0x01;
codes[cur + 13] = (codeSection[curCode + 2] >> (32 - 1 - 5)) & 0x1F;
codes[cur + 14] = (codeSection[curCode + 2] >> (32 - 1 - 10)) & 0x1F;
codes[cur + 15] = (codeSection[curCode + 2] >> (32 - 1 - 15)) & 0x1F;
codes[cur + 16] = (codeSection[curCode + 2] >> (32 - 1 - 20)) & 0x1F;
codes[cur + 17] = (codeSection[curCode + 2] >> (32 - 1 - 25)) & 0x1F;
codes[cur + 18] = (codeSection[curCode + 2] >> (32 - 1 - 30)) & 0x1F;
codes[cur + 19] = (codeSection[curCode + 2] << 4) & 0x10;
//第三个袋子的最后1位和第四个袋子的前4位组成一个物品，然后从第四个袋子中拿出5个物品，剩下3位。
codes[cur + 19] |= (codeSection[curCode + 3] >> (32 - 4)) & 0x0F;
codes[cur + 20] = (codeSection[curCode + 3] >> (32 - 4 - 5)) & 0x1F;
codes[cur + 21] = (codeSection[curCode + 3] >> (32 - 4 - 10)) & 0x1F;
codes[cur + 22] = (codeSection[curCode + 3] >> (32 - 4 - 15)) & 0x1F;
codes[cur + 23] = (codeSection[curCode + 3] >> (32 - 4 - 20)) & 0x1F;
codes[cur + 24] = (codeSection[curCode + 3] >> (32 - 4 - 25)) & 0x1F;
codes[cur + 25] = (codeSection[curCode + 3] << 2) & 0x1C;
//第四个袋子剩下的3位和第五个袋子的前2位组成一个物品，然后第五个袋子取出6个物品。
codes[cur + 25] |= (codeSection[curCode + 4] >> (32 - 2)) & 0x03;
codes[cur + 26] = (codeSection[curCode + 4] >> (32 - 2 - 5)) & 0x1F;
codes[cur + 27] = (codeSection[curCode + 4] >> (32 - 2 - 10)) & 0x1F;
codes[cur + 28] = (codeSection[curCode + 4] >> (32 - 2 - 15)) & 0x1F;
codes[cur + 29] = (codeSection[curCode + 4] >> (32 - 2 - 20)) & 0x1F;
codes[cur + 30] = (codeSection[curCode + 4] >> (32 - 2 - 25)) & 0x1F;
codes[cur + 31] = (codeSection[curCode + 4] >> (32 - 2 - 30)) & 0x1F;
cur += 32;
curCode += 5;
}
}

11. Simple Family

另一种多个数值打包在一起，并且是字对齐的编码方式，就是下面我们要介绍的Simple家族。

对于32位机器，一个字是32个bit，我们希望这32个bit里面可以放多个数值。比如1位的放32个，2位的放16个，3位放10个，不过浪费2位，4位放8个，5位放6个，6位放5个，7位和8位都是4个算一样，9位，10位都是放3个，11位，12位一直到16位都是放2个，32位放1个，共10种方法。那么来了32位，我们怎么区分里面是哪种方法呢？在放置真正的数据之前，需要放置一个选择器(selector)，来表示我们是怎么存储的，10种方法4位就够了。

如果这4位另外存储，就不是字对齐的了，所以还是将这4位放在32位里面，这样数据位就剩了28位了。那这28位怎么安排呢？1位的28个，2位的14个，3位的9个，4位的7个，5位的5个，6位和7位的4个，8位和9位的3个，10位到14位的都是2个，15位到28位的都是1个，共9种。如果同样存储2个，当然选最长的位数14，所以形成以下的表格：

由于一共9种情况，所以这种编码方式称为Simple-9。

4位selector来表示9，太浪费了，浪费了差不多一半的编码(24=16)，如果少一种情况，8种的话，就少一位做selector，多一位保存数值了，比如去掉selector=e这种情况，所有5位的都保存成7位，这样保存数据就是29位了，很遗憾，29是个质数，除了1和本身不能被任何数整除，多出的一位也往往浪费掉。

于是我们再进一步，用30位来保存数值，这样会有10种情况，而且浪费的情况也减少了。如下面的表格所示：

 

虽然看起来30比28要好一些，然而剩下的两位如何表示10种情况呢？我们假设文档编号之间的差值不会剧烈抖动，一会儿大一会儿小，而是维持在一个稳定的水平，就算发生突然的变化，变化完毕后也能保持较稳定的水平。所以我们可以采取这样的策略，首先对于第一个32位，假设selector是某一个数r，比如r=f，6位保存一个数值，那么下一个32位开始的两位表示四种情况：

1) 如果用的位数少，比如3位就够，则selector=r-1=e，也即用5位保存

2) 如果用的位数不变，还是用6位保存，则selector=r

3) 如果用的位数多 ，selector=r+1=g，也即用7位保存

4) 如果r+1=7位放不下，比如需要10位，说明了变化比较突然，于是r取最大值j，用30位来保存。

一旦出现了突然的较大变化，则会使用最大的selector j，然后随着突然变化后的慢慢平滑，selector还会降到一个文档的值。当然，如果事先经过统计，发现最大的文档间隔也不过需要15位，则对于第四种情况，可以使得r=i。

这种用相对的方式来表示10种情况，成为Relative-10。

既然selector只需要2位，那么上面的表格中selector=d和selector=g的没有用的两位，不是也可以成为下一个32位的selector么？这样下一个整个32位都可以来保存数据了。

另外上面表格中每个数值的编码长度一栏为什么会从7跳到10呢？是因为8位，9位，10位都只能保存3个，当然用最大的10位了。然而如果没有用的2位可以留给下一个32位做selector，那就有必要做区分了，比如一个值只需要9位，咱们就用9位，三九二十七，剩下的三位中的两位作为下一个32位的selector。另外14位和15位也是这个情况，如果两个14位，就可以剩下两位作为下一个的selector，如果是两个15位就没有给下一个剩下什么。

这样selector由10种情况变成了12种情况，如图下面的表格所示：

 

如果上一个32位剩下2位作为selector，则当前的32位则可以全部用于保存数据。当然也会有剩余，可留给后来人。那么32位全部用来保存数据的情况下，selector应该如下面表格所示：

 

这样每个32位都分为两种情况，一种是上一个留下了2位做selector，本身32位都保存数据，另一种是上一个没有留下什么，本身2位做selector，30位做数据。

这种上一个32位为下一个留下遗产，共12种情况的，成为Carryover-12编码。

下面举一个具体的例子，比如我们想编码如下的输入：

int[] input = {5, 30, 120, 60, 140, 160, 120, 240, 300, 200, 500, 800, 300, 900};

首先定义上面的两个编码表：

class Item {
public Item(int lengthOfCode, int numOfCodes, boolean leftForNext) {
this.lengthOfCode = lengthOfCode;
this.numOfCodes = numOfCodes;
this.leftForNext = leftForNext;
}
int lengthOfCode;
int numOfCodes;
boolean leftForNext;
}
static Item[] noPreviousSelector = new Item [12];
static {
noPreviousSelector[0] = new Item(1, 30, false);
noPreviousSelector[1] = new Item(2, 15, false);
noPreviousSelector[2] = new Item(3, 10, false);
noPreviousSelector[3] = new Item(4, 7, true);
noPreviousSelector[4] = new Item(5, 6, false);
noPreviousSelector[5] = new Item(6, 5, false);
noPreviousSelector[6] = new Item(7, 4, true);
noPreviousSelector[7] = new Item(9, 3, true);
noPreviousSelector[8] = new Item(10, 3, false);
noPreviousSelector[9] = new Item(14, 2, true);
noPreviousSelector[10] = new Item(15, 2, false);
noPreviousSelector[11] = new Item(28, 1, true);
}
static Item[] hasPreviousSelector = new Item [12];
static {
hasPreviousSelector[0] = new Item(1, 32, false);
hasPreviousSelector[1] = new Item(2, 16, false);
hasPreviousSelector[2] = new Item(3, 10, true);
hasPreviousSelector[3] = new Item(4, 8, false);
hasPreviousSelector[4] = new Item(5, 6, true);
hasPreviousSelector[5] = new Item(6, 5, true);
hasPreviousSelector[6] = new Item(7, 4, true);
hasPreviousSelector[7] = new Item(8, 4, false);
hasPreviousSelector[8] = new Item(10, 3, true);
hasPreviousSelector[9] = new Item(15, 2, true);
hasPreviousSelector[10] = new Item(16, 2, false);
hasPreviousSelector[11] = new Item(28, 1, true);
}

形成的编码格式如图

 

假设约定的起始selector为6，也即表3-10的g行。对于第一个32位，是不会有前面遗传下来的selector的，所以前两位表示selector=1，也即就用原始值6，第g行，接下来应该是4个7位的数值，最后剩余两位作为下一个32位的selector。Selector=2，所以用6+1=7，也即表3-11的h行，下面的整个32位都是数值，保存了4个8位的，没有遗留下什么。接下来的32位中，头两位是selector=1，还是第7行，也即第h行，只不过是表3-10中的，所以接下来应该是3个9位，遗留下最后两位selector=2，也即是7+1=8行，也即表3-11的第i行，接下来应该是3个10位的。

整个解码的过程如下：

//block是编码好的块，defaultSelector是默认的selector
private static int[] decompress(int[] block, int defaultSelector, int numOfCodes) {
//当前处理到编码块的哪一个int
int curInBlock = 0;
//解码结果
int[] output = new int[numOfCodes];
int curInOutput = 0;
//前一个selector，用于根据相对值计算当前selector的值
int prevSelector = defaultSelector;
int curSelector = 0;
//最初的编码表用当然是没有遗留的
Item[] curSelectorTable = noPreviousSelector;
//尚未处理的bit数
int bitsRemaining = 0;
//当前int中每个编码的bit数
int bitsPerCode = curSelectorTable[curSelector].lengthOfCode;
//当前要解码的32位编码
int cur = 0;
//一个循环，当curInBlock > block.length的时候，编码块已经处理完毕，但是还需要等待最后一个32位处理完毕，当bitsRemaining大于等于bitsPerCode的时候，说明还有没处理完的编码
while (curInBlock < block.length || bitsRemaining >= bitsPerCode) {
//当bitsRemaining不足一个编码的时候，说明当前的32位处理完毕，应该读入下一个32位了。
if(bitsRemaining < bitsPerCode){
//当bitsRemaining小于2，说明当前32位剩下的不足2位，不够给下一个做selector的，因而下一个selector在下一个32位的头两位。
if(bitsRemaining < 2){
//取下一个32位
cur = block[curInBlock++];
//前两位为selector
int selector = (cur >> 30) & 0x03;
//根据selector的相对值计算当前的selector
if(selector == 0){
curSelector = prevSelector - 1;
} else if (selector == 1){
curSelector = prevSelector;
} else if (selector == 2) {
curSelector = prevSelector + 1;
} else {
curSelector = curSelectorTable.length - 1;
}
prevSelector = curSelector;
//当前32位中数据仅仅占30位
bitsRemaining = 30;
//使用编码表3-10
curSelectorTable = noPreviousSelector;
}
//如果bitRemaining大于等于2，足够给下一个做selector，则解析最后两位为selector。
else {
int selector = cur & 0x03;
if(selector == 0){
curSelector = prevSelector - 1;
} else if (selector == 1){
curSelector = prevSelector;
} else if (selector == 2) {
curSelector = prevSelector + 1;
} else {
curSelector = curSelectorTable.length - 1;
}
prevSelector = curSelector;
//取下一个32位，全部用于保存数值
cur = block[curInBlock++];
bitsRemaining = 32;
//使用编码表3-11
curSelectorTable = hasPreviousSelector;
}
bitsPerCode = curSelectorTable[curSelector].lengthOfCode;
}
//在bitRemaing中截取一个编码，解码到输出，更新bitsRemaining
int mask = (1 << bitsPerCode) - 1;
output[curInOutput++] = (cur >> (bitsRemaining - bitsPerCode)) & mask;
bitsRemaining = bitsRemaining - bitsPerCode;
}
return output;
}

12. 跳跃表

上面说了很多的编码方式，能够让倒排表又小又快的存储和解码。

但是对于倒排表的访问除了顺序读取，还有随机访问的问题，比如我想得到第31篇文档的ID。

第一点，差值编码使得每个文档ID都很小，是个不错的选择。第二点，上面所述的很多编码方式都是变长的，一个挨着一个存储的。这两点使得随机访问成为一个问题，首先因为第二点我们根本不知道第30篇文档在文件中的什么位置，其次就算是找到了，因为第二点，找到的也是差值，需要知道上一个才能知道自己，然而上一个也是差值，难不成每访问一篇文档整个倒排表都解压缩一遍？

看过前面前端编码的同学可能想起了，差值和前缀这不差不多的概念么？弄几个排头兵不就行啦。

如图，上面的倒排表被用差值编码表示成了下面一行的数值。我们将倒排表分成组，比如3个数值一组，每组有个排头兵，排头兵不是差值编码的，这样如果你找第31篇文档，那它肯定在第10组里面的第二个，只需要解压一个组的就可以了。

 

有了跳跃表，根据文档ID查找位置也容易了，比如要在文档57，在排头兵中可以直接跳过17,34,45，肯定在52的组里，发现52+5=57，一进组就找到了。

当然排头兵如果比较大，也可以用差值编码，是基于前一个排头兵的差值，由于排头兵比较少，反正查找的时候要一个个看，都解压了也没问题。

如果链表比较长，导致排头兵比较多，没问题还可以有多级跳跃表，排头兵还有排头兵，排长上面是连长。这样查找的时候，先查找连长，找到所在的连再查找排长，然后再进组查找。

上面提到的PForDelta的编码方式可以和跳跃表进行整合，由于PForDelta的编码区域是定长的，基本可以满足随机访问，然而对于差值问题，可以再entry中添加排头兵的值的信息，使得128个数值成为一组。

我们姑且称这种方式为跳跃表规则。

11.1—11.15云计算基础服务全场5折起

免费体验云安全(易盾)内容安全、验证码等服务

更多网易技术、产品、运营经验分享请点击。