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
}
网友评论