位图Bitmap以及RoaringBitmap

by Posted on

简介

位图（Bitmap）算法，即用位（Bit）来表示数据的状态。如果一个Bit能表示一个数据的两种状态（0/1），那64个bit就能表示64个数据的状态。因此我们能很容易联想到一种场景：如果要表示某一个数字存不存在，那刚好就能用0或1的状态来表示。那具体怎么做呢？举个场景例子，比如某个学校想统计最近3天内有多少同学做了核酸，有一部分同学在这3天有可能被做过好几次核酸（再假设该学校有1万名学生，学号从1-10000）。这时候我们可以用Bitmap来统计，我们给定一个长度为10000的位图，然后用每一位（Bit）代表一名学生，如果该名学生被做过核酸，则该位(Bit)上置为1:

如上图我们标黄的，则是状态置为1的位(Bit)，因此学号1，5，9999的学生都做过核酸。我们只需要统计二进制表示有多少个1就可以了，具体计算方法可以参考下我之前写的文章。此时可能有同学就能想到了，这种场景用HashMap或者HashSet同样能计算出来。确实没错，但是用位图的方式，10000个数字用10000个Bit（约1K空间）即可搞定，如果用Integer数组或者HashMap，则需要差不多40K空间。并且如果这里数字的个数不是1万，而是100万甚至1亿，10亿呢，则空间差距会更加明显。

当然位图不是没有缺点，如果想知道某一个学生具体被做了几次核酸，这是办不到的，因为位图天然有去重的特性；另外假设极端情况，全校只有一个同学被做了核酸，那这1万个Bit中，只有一个Bit被赋值，其余空间全浪费了，因为位图它固定占用了1K的空间。

现在我们尝试来用代码实现这个需求。

BitSet

JDK中有位图的默认实现，就是BitSet：

public class BitSet implements Cloneable, java.io.Serializable {
    // 地址位，一个word单位的地址位
    private final static int ADDRESS_BITS_PER_WORD = 6;
    // 一个word有多少位。这里即 64
    private final static int BITS_PER_WORD = 1 << ADDRESS_BITS_PER_WORD;
    // bit索引位置掩码
    private final static int BIT_INDEX_MASK = BITS_PER_WORD - 1;
​
    private static final long WORD_MASK = 0xffffffffffffffffL;
​
    // 实际存储状态的位，由long数组实现
    private long[] words;
  
    // 通过右移来定位当前”位“在哪个word中。
    private static int wordIndex(int bitIndex) {
        return bitIndex >> ADDRESS_BITS_PER_WORD;
    }
    .......
  }

我们可以看出来，BitSet中以word为单位来存储位信息，一个word即64位，存储64位状态信息，也就是一个long的大小。

我们用BitSet直接来实现：

public static void main(String[] args) {
        BitSet fl = new BitSet();
        fl.set(1);
        fl.set(5);
        fl.set(9999);
        System.out.println("已做人数:" + fl.cardinality());
}

其中的set方法就是将该索引位设置为1，最后的cardinality方法统计位图中1的个数。下面我们看下这两个方法的实现：

public void set(int bitIndex) {
        if (bitIndex < 0)
            throw new IndexOutOfBoundsException("bitIndex < 0: " + bitIndex);
        // 判断当前的bitIndex位于哪个word中
        int wordIndex = wordIndex(bitIndex);
        // 如果当前的位数不够存储，则扩展words数组。
        expandTo(wordIndex);
        // 将对应的”位“状态置为1
        words[wordIndex] |= (1L << bitIndex); // Restores invariants
        checkInvariants();
 }

第一步便是判断当前bitIndex位于哪个word中，其实就是用bitIndex/64来判断。如果实际存储数据的 long[] words长度不够，则扩展数组的长度。因此1和5存储之后，实际words的长度还只是1，但是当9999存储进来之后，就不得不扩展数组长度，因为它的wordIndex是9999/64（即：156），扩展之后整个words的长度为157。最后就是将对应的word中的某一bit置为1（即左移，然后或运算），其中9999超过了long的最大位数，因此实际上它左移的位数为9999%64，即15。最终我们得到words数组存储的具体值如下图：

因此虽然只需要存储三个值，但是由于9999导致了数组的扩展，从而其余的空间全部浪费。我们可以得出如下结论：对于随机的稀疏数据，BitSet会有较大的空间浪费。

最后就是cardinality方法：

public int cardinality() {
    int sum = 0;
    for (int i = 0; i < wordsInUse; i++)
        sum += Long.bitCount(words[i]);
    return sum;
}

相对就比较简单了，统计位图中1的个数，就是直接统计每个word中1的个数，然后做累加操作即可，这里不再赘述。对于BitSet源码这里也是点到即止，整体的BitSet源码有兴趣同学可以自己查阅。

RoaringBitmap

从上面我们看到了BitSet来实现位图的缺点，对于随机稀疏的数据，空间上有比较大的浪费。因此就有很多压缩算法的Bitmap应运而生，RoaringBitmap就是其中之一。附上Github传送

RoaringBitmap的核心思想便是用一个Container数组来存储状态信息。这个Container与上述的word有点类似。对于一个Integer数字来说，那怎么去定位它处于哪一个Container之中呢？RoaringBitmap的算法思想是利用Interger的高16位来定位它在Container的索引位置，最后将这个数字的低16位存储在Container中。因此这样的话，我们最多有2^16次方，即65536个Container，也就是说RoaringBitmap中实际存储信息的Container数组最大长度为65535，每个Container又可以容纳65536个数据。

我们现在也来尝试用RoaringBitmap来实现上面统计人数的场景：

RoaringBitmap roaringBitmap = new RoaringBitmap();
 roaringBitmap.add(1);
 roaringBitmap.add(5);
 roaringBitmap.add(9999);
 System.out.println("已做人数: " + roaringBitmap.getCardinality());

代码也非常简单，跟BitSet使用非常类似，这里也不再赘述。那它会不会也像BitSet那样浪费空间呢？我们看下它的add实现：

// 存储信息的Container数组
RoaringArray highLowContainer = null;
​
public void add(final int x) {
  // 取高16位信息
  final char hb = Util.highbits(x);
  // 根据高16位信息，定位container索引位置
  final int i = highLowContainer.getIndex(hb);
  // i大于等于0，则直接获取对应的container，将低16位信息保存到container中
  if (i >= 0) {
    highLowContainer.setContainerAtIndex(i,
        highLowContainer.getContainerAtIndex(i).add(Util.lowbits(x)));
  } else {
    // i小于0，说明对应的container没有被创建，则先创建container，再将低16位信息保存到container中。
    final ArrayContainer newac = new ArrayContainer();
    highLowContainer.insertNewKeyValueAt(-i - 1, hb, newac.add(Util.lowbits(x)));
  }
}
​
public final class RoaringArray implements Cloneable, Externalizable, AppendableStorage<Container> {
  ......
  static final int INITIAL_CAPACITY = 4;
  // Container对应的高16位Key数组。注意，这里的keys是排序好的数组。
  char[] keys = null;
  // container数组。
  Container[] values = null;  
  ......
}

注释已经说明了比较详细。其中通过高16位来定位到Container索引位置，数字1、5、9999的高16位都是0，因此也对应了第0个索引位置的Container。如果某几个数字的高16位是1000、2000、10000，那会不会需要扩展数组到10000呢？答案是否定的，Container数组只需要存储实际的key值，然后按照顺序存储，因此如果高16位（Key值）为0、1000、2000、10000这四个数字时，Container数组的Keys存储为: [0,1000,2000,10000]。当有新的key值进入时（比如100），它会通过二分查找来锁定合适的索引位置，即插入之后：[0,100,1000,2000,10000]。

代码中返回的 i 就是对应的Container索引值。区别就是：如果Container已经存在便是正值，如果Container还没创建则是负值。负值情况下则会先创建一个ArrayContainer来存储数据。那问题来了，位图算法为什么要用Array来存储值呢？要回答这个问题，我们需要看下ArrayContainer保存信息的实现方式： 

public final class ArrayContainer extends Container implements Cloneable {
  // 数组默认初始化大小
  private static final int DEFAULT_INIT_SIZE = 4;
  private static final int ARRAY_LAZY_LOWERBOUND = 1024;
  static final int DEFAULT_MAX_SIZE = 4096;// containers with DEFAULT_MAX_SZE or less integers
                                           // should be ArrayContainers
  // 数组中实际存储数据个数的一个统计值
  protected int cardinality = 0;
  // 存储低16位数据的数组
  char[] content;
  
  public Container add(final char x) {
    if (cardinality == 0 || (cardinality > 0
            && (x) > (content[cardinality - 1]))) {
      // 数据按顺序存储，如果当前存储数据超过4096，则转换为BitmapContainer存储数据。
      if (cardinality >= DEFAULT_MAX_SIZE) {
        return toBitmapContainer().add(x);
      }
      if (cardinality >= this.content.length) {
        // 扩展数组容量，来支持新的插入
        increaseCapacity();
      }
      content[cardinality++] = x;
    } else {
      // 由于数据介于content数组的中最大值和最小值之间，因此通过二分查找来找到合适的索引位置
      int loc = Util.unsignedBinarySearch(content, 0, cardinality, x);
      // loc大于0，则数组中已经存在该值，不做处理，如果不存在则loc小于0，于是插入该值。
      if (loc < 0) {
        // 
        if (cardinality >= DEFAULT_MAX_SIZE) {
          return toBitmapContainer().add(x);
        }
        if (cardinality >= this.content.length) {
          increaseCapacity();
        }
        // 数组中大于x的值往右挪动，给x腾出位置来。
        System.arraycopy(content, -loc - 1, content, -loc, cardinality + loc + 1);
        content[-loc - 1] = x;
        ++cardinality;
      }
    }
    return this;
  }
}

我们看到，ArrayContainer完全就是用数组来存储信息，然后通过数组拷贝，二分查找等操作来维持这个数组有序。

我们这里能得出一个结论：对于高16位和低16位都是顺序存储，然后通过二分查找来定位索引位置。于是乎，对于一个新加入的数据，能够快速的定位它存不存在，以及需要被存储在什么位置。

有一个点需要注意，当数组存储的数据超过4096之后，则转换为BitmapContainer存储数据。这里虽然终于看到了Bitmap，但为什么是4096呢？我们假设ArrayContainer刚好存满了4096个数据，那此时它大约占用多少空间呢？char占用2个字节，那总共占用空间是4096*2=8K。如果我们用位图，存满所有65536个数据，需要1024个long数组，大约占用1024*8=8K空间。所以，当数据超过4096时，转换成位图存储，这样就能避免空间的浪费。数组可以用来存储稀疏数据，当数据量超过阈值时，用位图来存储稠密数据。

我们把BitmapContainer的add方法也放出来，如下所示：

final long[] bitmap = new long[1024]; 
@Override
public Container add(final char i) {
  final long previous = bitmap[i >>> 6];
  long newval = previous | (1L << i);
  bitmap[i >>> 6] = newval;
  if (USE_BRANCHLESS) {
    cardinality += (int)((previous ^ newval) >>> i);
  } else if (previous != newval) {
    ++cardinality;
  }
  return this;
}

这段代码是比较好理解的，跟BitSet非常相似。只不过用了前值和后值的对比，来判断需不需要对cardinality值加1。previous ^ newval异或之后是0，说明该值之前已经存在，否则 (previous ^ newval) >>> i 就是1。

RoaringBitmap中我们看到了两个关键的Container：ArrayContainer和BitmapContainer。在数据存储时，会做相应的转换。当然，RoaringBitmap还有一个Container：RunContainer。它采用游程编码方式，来压缩连续出现的数字。如果需要保存的数据是比较连续的数据，则可以考虑使用该Container，它能带来极大的压缩效果，节省存储空间。如果数据随机无规则，最好不要用该Container，因为可能会给存储带来适得其反的效果。

总结

本文对位图算法，以及RoaringBitmap做了简单介绍，当然它们还有很多用法。当然位图最酷的用法，还是位图之间的And，Or，Xor，Andnot操作。如果把两个位图理解成两个不同的集合，则两个不同的位图做And运算，能得出两个位图中都存在的数据，即取交集操作；做Or即相当于做并集；做Andnot相当于取差集；做Xor相当于两个差集的并集。

Published by Binson