巧用位运算（三）—— 奇技淫巧 Bitmap

Bitmap

考虑下面一个问题：

如何从40亿个正整数中，判断某个数字是否存在？

第一反应肯定是数组遍历，如果我们考虑用Int类型数组来存放，那么这40亿的数据会占用约12G内存，这显然不太现实。

这里肯定有朋友要杠了：我为啥非要加载到内存中呢，就从文件读取遍历不行吗？
当然可以。但是如果这个需求是频繁出现的呢，每次都去走I/O操作进行文件内容遍历吗？

回归正题，要怎样才能让我们把40亿数据都加载进入内存呢？这个时候我们就可以考虑一种叫做Bitmap的数据结构了。

什么是Bitmap呢，非常简单，我们看下面图1表示一个数组，这个数组长度是10，该数组的初始值均为0，并且数组的值只能是0和1.

图1

现在我们存放三个数字1，2，4，如图2，只需要把数组索引为1，2，4的值置为1。

图2

以此类推，我们这个数组可以存放0~99个数字。反应快的朋友已经发现了，我们这个数组实际上就是一个“bit类型”（bit只能表示为二进制的0和1）的数组。也就是我们通常所说的Bitmap。
刚才我们说到，JAVA里面我们通常用Int存放一个数字，Int占用4个字节，每个字节占用8位bit，即一个Int占用32位bit。也就是说，我们用一个Int类型的变量就可以存放0-31这样最多32个数字。

那我们现在来看看，用Bitmap存放40亿个数字，只需要约40亿/8/1024/1024=476M的内存空间。

解决了内存的问题，我们再来看看查询的问题。

回顾开篇我们提到的问题，如何判断一个数字是否存在。如果用数组存放，我们需要遍历，时间复杂度是O(N)，加上排序和二分查找，可以把时间复杂度降到O(logN)。而Bitmap则可以把查询的时间复杂度降低到O(1)，因为只需要进行一次简单的位运算操作，具体内容后面参照代码再做解释。

Bitmap的用处

Bitmap的最大的好处当然是内存压缩，但它的数据结构也给我们带来了很多额外的好处。

位运算替代SQL

设想以下场景：

一个网站如何统计出所有连续在线一周的用户

当然，在关系型数据库下，我们只需要一条SQL就能轻松搞定这个问题。但如果这个网站有20亿用户呢？你要怎么来设计这条SQL以保证我们查询的性能呢？

无疑Bitmap为我们提供了更好的解决思路。 首先，我们保证用户ID为一串纯数字，以便我们能够把20亿的用户ID存放在Bitmap中。通过上面的学习我们已经了解到Bitmap的本质就是按位存储，这样我们只需要把7天的Bitmap数据进行一个&位运算，最终得出的新的Bitmap就是连续一周在线的用户ID的集合了。

同理，如果我们要找出一周内上过线的所有用户，也只是一个|位运算就能搞定的事情。

布隆过滤器

大名鼎鼎的布隆过滤器实际上底层的数据结构（比特向量）就使用的Bitmap。将数据哈希映射在Bitmap上，然后通过位运算能够快速地判定数据是否已经存在，从而有效地防止缓存穿透问题。

索引

如Oracle等关系型数据库提供了位图索引，其本质就是Bitmap。

海量数据的快速排序、查找、去重

Bitmap的实现

下面我以Go语言为例实现简单的Bitmap数据结构。

package bitmap

const AddressBitsPerWord = 6
const InitCap = 1024
const Bits = 64

type Bitmap struct {
    words     []uint64
    wordInUse uint16
    size      uint16
}

func CreateBitmap() *Bitmap {
    bitmap := &Bitmap{}
    bitmap.words = make([]uint64, InitCap)
    bitmap.wordInUse = 1
    return bitmap
}

func (bm *Bitmap) Set(bitIndex uint16) bool {
    wordIndex := wordIndex(bitIndex)
    bm.expandTo(wordIndex)
    word := bm.words[wordIndex]
    bi := bitIndex & (Bits - 1)
    // 判断是否已经存在
    if (word & (1 << bi)) == (1 << bi) {
        return false
    }
    bm.words[wordIndex] |= 1 << bi
    bm.size += 1
    return true
}

func (bm *Bitmap) Get(bitIndex uint16) bool {
    wordIndex := wordIndex(bitIndex)
    word := bm.words[wordIndex]
    bi := bitIndex & (Bits - 1)
    // 判断是否已经存在
    if (word & (1 << bi)) == (1 << bi) {
        return true
    }
    return false
}

func (bm *Bitmap) List() []uint16 {
    cap := bm.wordInUse << AddressBitsPerWord
    arr := make([]uint16, bm.size)
    index := 0
    for i := uint16(0); i < cap; i++ {
        if bm.Get(i) {
            arr[index] = i
            index++
        }
    }
    return arr
}

func wordIndex(bitIndex uint16) uint16 {
    return bitIndex >> AddressBitsPerWord
}

func (bm *Bitmap) expandTo(index uint16) {
    wordRequired := index + 1
    if bm.wordInUse < wordRequired {
        bm.ensureCapacity(wordRequired)
        bm.wordInUse = wordRequired
    }
}

func (bm *Bitmap) ensureCapacity(wordRequired uint16) {
    wl := uint16(len(bm.words))
    if wl < wordRequired {
        expand := make([]uint64, wordRequired-wl)
        bm.words = append(bm.words, expand...)
    }
}

func (bm *Bitmap) Size() uint16 {
    return bm.size
}

从上面的Get()方法中我们可以看出，查询数据是否存在只是做了一个取模和位运算。

Bitmap的进化

大家很容易想到Bitmap在应对稀疏数据时不是一个好的选择，例如我们要存以下数据

1，11111，11111111，111111111111

如果使用bitmap进行存储，仅仅4个数字花费的存储空间是不可想象的。这种情况下，我们不得不对Bitmap进行改进。

EWAHCompressedBitmap

谷歌所实现的EWAHCompressedBitmap完全使用了行程压缩算法对Bitmap进行压缩。此处不对EWAHCompressedBitmap做详细介绍，重点介绍一下下面的RoaringBitMap算法。

RoaringBitMap（RBM）

RBM是2016年由S. Chambi、D. Lemire、O. Kaser等人在论文《Better bitmap performance with Roaring bitmaps》与《Consistently faster and smaller compressed bitmaps with Roaring》中提出。

RBM的设计思路是：将32位的无符号整型按高16位进行分组，也就分出了2^16=65536个Container

低16位作为值，放入到对应的Container中， 如下图

image.png

对于Container的实现一般有两种：

• Array
  当Container中的基数不大于4096时，使用ArrayContainer来存放，ArrayContainer本质为uint16的数组，初始长度4，数组会自动扩容。
• Bitmap
  当Container中的基数大于4096时，改用BitmapContainer来存放会更节约空间，BitmapContainer本质为我们上文说到的bitmap。

当然极限情况下也可以考虑用我在前文 《巧用位运算（二）—— 变长整型的实现》提到的变长整形数组来实现。

这里我们可以代入hashmap的设计思路来理解，相当于把uint32分为两部分，高16位作为key，低16位作为value。即有一个最大长度为65535的哈希槽（本质为数组，存放keys）加上最多65535个链表（存放values）。只不过这里的链表实现换成了数组或bitmap的实现（因为只需要存放无符号整数，链表实现浪费空间）。

我们可以体会到，在大数据并且数据稀疏的情况下，使用RoaringBitMap会非常地节省内存空间。大名鼎鼎的ElasticSearch的倒排表索引就用到了RoaringBitMap（实际为Lucene实现）。

下面贴一下我自己基于Go语言实现的RMB

package rbm

import (
    "reflect"
    "roaringbitmap/bitmap"
    "roaringbitmap/container"
    "roaringbitmap/lookup"
    "roaringbitmap/util"
)

type RoaringBitmap struct {
    // 高16位作为key
    keys []uint16
    // 低16位作为value
    values map[uint16]container.Container
    size   uint
}

func (rbm *RoaringBitmap) Values() map[uint16]container.Container {
    return rbm.values
}

func (rbm *RoaringBitmap) Keys() []uint16 {
    return rbm.keys
}

func Init() (rbm *RoaringBitmap) {
    rbm = &RoaringBitmap{}
    rbm.keys = make([]uint16, 0)
    rbm.values = make(map[uint16]container.Container)
    return
}

// 添加整数
func (rbm *RoaringBitmap) Add(val uint32) bool {
    high := uint16(val >> 16)
    low := uint16(val)
    // 添加key
    rbm.addKey(high)
    return rbm.addValue(high, low)
}

func (rbm *RoaringBitmap) addKey(key uint16) {
    // 判断key是否已经存在   使用二分查找
    i, b := lookup.Lookup(rbm.keys, key)
    if b {
        return
    }
    rbm.keys = util.InsertArray(rbm.keys, key, i)
}

func (rbm *RoaringBitmap) addValue(key uint16, value uint16) bool {
    if rbm.values[key] == nil {
        rbm.values[key] = &container.ArrayContainer{
            Values: make([]uint16, 0),
        }
    }
    con := rbm.values[key]
    if (reflect.TypeOf(con).String() == "*container.ArrayContainer") && con.Size() >= 4096 {
        con = rbm.resetContainer(key)
    }
    if !con.Add(value) {
        return false
    }
    rbm.size = rbm.size + 1
    return true
}

func (rbm *RoaringBitmap) List() []uint32 {
    arr := make([]uint32, 0)
    for _, v := range rbm.keys {
        high := v
        container := rbm.values[v]
        lows := container.List()
        for _, low := range lows {
            val := (uint32(high) << 16) | uint32(low)
            arr = append(arr, val)
        }
    }
    return arr
}

func (rbm *RoaringBitmap) Size() uint {
    return rbm.size
}

func (rbm *RoaringBitmap) resetContainer(key uint16) container.Container {
    oldContainer := rbm.values[key]
    newContainer := &container.BitmapContainer{
        Values: bitmap.CreateBitmap(),
    }
    for _, v := range oldContainer.List() {
        newContainer.Add(v)
    }
    rbm.values[key] = newContainer
    return newContainer
}