bitmap原理+性能优化实践

背景

工作中用到了bitmap，结构是Roaring64NavigableMap，源码提供的遍历方式在性能上没达到要求

lou一遍底层源码，浅浅分析一下

总体结构

Roaring64NavigableMap

    存储结构为NavigableMap<Integer, BitmapDataProvider> highToBitmap，NavigableMap底层是红黑树实现

BitmapDataProvider

    实现主要是RoaringBitMap

RoaringBitMap

    底层结构是一个RoaringArray

RoaringArray

    char[] key;

    Container[] values[];

    

从RoaringBitmp说起

RoaringBitmap的原理与应用，看这个就够了_java编程艺术的博客-CSDN博客_roaringbitmap

1、介绍

RoaringBitmap是高效压缩位图，简称RBM

2、基本原理

• 将 32bit int（无符号的）类型数据 划分为 2^16 个桶，即最多可能有216=65536个桶，论文内称为container。用container来存放一个数值的低16位

• 在存储和查询数值时，将数值 k 划分为高 16 位和低 16 位，取高 16 位值找到对应的桶，然后在将低 16 位值存放在相应的 Container 中（存储时如果找不到就会新建一个）

• 比如要将31这个数放进roarigbitmap中，它的16进制为：0000001F,前16位为0000，后16为001F。

所以先需要根据前16位的值：0，找到它对应的通的编号为0，然后根据后16位的值：31，确定这个值应该放到桶中的哪一个位置，如下图所示。

3、桶的类型

在roaringbitmap中共有3种小桶：

• arraycontainer(数组容器)，
• bitmapcontainer(位图容器)，
• runcontainer(行程步长容器)，

类图如下：

3.1arraycontainer

当ArrayContainer（其中每一个元素的类型为 short int 占两个字节,且里面的元素都是按从大到小的顺序排列的）的容量超过4096（即8k）后，会自动转成BitmapContainer(这个所占空间始终都是8k)存储。

1.3.2 bitmapcontainer

这个容器就是位图，只不过这里位图的位数为216（65536）个，也就是2^16个bit, 所占内存就是8kb。然后每一位用0，1表示这个数不存在或者存在

1.3.3 runcontainer

RunContainer中的Run指的是行程长度压缩算法(Run Length Encoding)，对连续数据有比较好的压缩效果。它的原理是，对于连续出现的数字，只记录初始数字和后续数量。

4、对比bitmap

roaringbitmap除了比bitmap占用内存少之外，其并集和交集操作的速度也要比bitmap的快，原因如下：

• 计算上的优化

对于roaringbitmap本质上是将大块的bitmap分成各个小块，其中每个小块在需要存储数据的时候才会存在。所以当进行交集或并集运算的时候，roaringbitmap只需要去计算存在的一些块而不需要像bitmap那样对整个大的块进行计算。

同时在roaringbitmap中32位长的数据，被分割成高 16 位和低 16 位，高 16 位表示块偏移，低16位表示块内位置，单个块可以表达 64k 的位长，也就是 8K 字节。这样可以保证单个块都可以全部放入 L1 Cache，可以显著提升性能

• 程序逻辑上的优化

• roaringbitmap维护了排好序的一级索引以及有序的arraycontainer，当进行交集操作的时候，只需要根据一级索引中对应的值来获取需要合并的容器，而不需要合并的容器则不需要对其进行操作直接过滤掉。

• 当进行合并的arraycontainer中数据个数相差过大的时候采用基于二分查找的方法对arraycontainer求交集,避免不必要的线性合并花费的时间开销。

• roaingbitmap在做并集的时候同样根据一级索引只对相同的索引的容器进行合并操作，而索引不同的直接添加到新的roaringbitmap上即可，不需要遍历容器。

• roaringbitmap在合并容器的时候会先预测结果，生成对应的容器，避免不必要的容器转换操作。

1.6 针对long整数的扩展【64-bit integers (long)】

虽然RoaringBitmap是为32位的情况设计的，但对64位整数进行了扩展。为此提供了两个类:Roaring64NavigableMap和Roaring64Bitmap。

Roaring64NavigableMap依赖于传统的红黑树。键是32位整数，代表元素中最重要的32位，而树的值是32位RoaringBitmap。32位RoaringBitmap表示一组元素的最低有效位。

较新的Roaring64Bitmap方法依赖ART数据结构来保存键/值对。键由元素的最重要的48位组成，而值是16位的Roaring容器

上源码

Roaring64NavigableMap

  /**
   * 
   * @param signedLongs true if longs has to be ordered as plain java longs. False to handle them as
   *        unsigned 64bits long (as RoaringBitmap with unsigned integers)
   * @param cacheCardinalities true if cardinalities have to be cached. It will prevent many
   *        iteration along the NavigableMap
   * @param supplier provide the logic to instantiate new {@link BitmapDataProvider}, typically
   *        instantiated once per high.
   */
  public Roaring64NavigableMap(boolean signedLongs, boolean cacheCardinalities,
      BitmapDataProviderSupplier supplier) {
    this.signedLongs = signedLongs;
    this.supplier = supplier;

    if (signedLongs) {
      highToBitmap = new TreeMap<>();
    } else {
      //默认unsigned 64bits long
      highToBitmap = new TreeMap<>(RoaringIntPacking.unsignedComparator());
    }
    //默认缓存基数
    this.doCacheCardinalities = cacheCardinalities;
    resetPerfHelpers();
  }

  public void addLong(long x) {
    int high = high(x);
    int low = low(x);

    // Copy the reference to prevent race-condition
    Map.Entry<Integer, BitmapDataProvider> local = latestAddedHigh;

    BitmapDataProvider bitmap;
    if (local != null && local.getKey().intValue() == high) {
      bitmap = local.getValue();
    } else {
      bitmap = highToBitmap.get(high);
      if (bitmap == null) {
        //使用 RoaringBitmap 来保存低层数据, 一级存储
        bitmap = newRoaringBitmap();
        //保存高位
        pushBitmapForHigh(high, bitmap);
      }
      latestAddedHigh = new AbstractMap.SimpleImmutableEntry<>(high, bitmap);
    }
    //RoaringBitmap add  存储低位信息
    bitmap.add(low);
    // 扩容处理
    invalidateAboveHigh(high);
  }

  public boolean contains(long x) {
    //高32位
    int high = RoaringIntPacking.high(x);
    BitmapDataProvider lowBitmap = highToBitmap.get(high);
    if (lowBitmap == null) {
      return false;
    }
        //低32位
    int low = RoaringIntPacking.low(x);
    return lowBitmap.contains(low);
  }

  public long getLongCardinality() {
    if (doCacheCardinalities) {
      if (highToBitmap.isEmpty()) {
        return 0L;
      }
      int indexOk = ensureCumulatives(highestHigh());

      // ensureCumulatives may have removed empty bitmaps
      if (highToBitmap.isEmpty()) {
        return 0L;
      }

      return sortedCumulatedCardinality[indexOk - 1];
    } else {
      long cardinality = 0L;
      for (BitmapDataProvider bitmap : highToBitmap.values()) {
        cardinality += bitmap.getLongCardinality();
      }
      return cardinality;
    }
  }

RoaringBitmap

//RoaringBitmap add 
  public void add(final int x) {
    //char 2个字节16bit
    final char hb = Util.highbits(x);
    //获取hb在key数组中的位置（二分查找）
    final int i = highLowContainer.getIndex(hb);
    if (i >= 0) {
      // 此处查找成功
      highLowContainer.setContainerAtIndex(i,
          //RoaringArray return this.values[i];
          highLowContainer.getContainerAtIndex(i).add(Util.lowbits(x)));
    } else {
      final ArrayContainer newac = new ArrayContainer();
      //RoaringArray 在位置i插入key=hb value=插入的newac.add返回的Container
      //ArrayContainer.add
      highLowContainer.insertNewKeyValueAt(-i - 1, hb, newac.add(Util.lowbits(x)));
    }
  }

  public boolean contains(final int x) {
    final char hb = Util.highbits(x);
    final Container c = highLowContainer.getContainer(hb);
    return c != null && c.contains(Util.lowbits(x));
  }

RoaringArray

/**
 * RoaringArray结构
 *         char[] keys = null;
 *      Container[] values = null;
 *        int size = 0;
 */

//没有找到返回-i-1，以便获取index
int getIndex(char x) {
  // before the binary search, we optimize for frequent cases
  if ((size == 0) || (keys[size - 1] == x)) {
    return size - 1;
  }
  // no luck we have to go through the list
  return this.binarySearch(0, size, x);
}

void setContainerAtIndex(int i, Container c) {
  this.values[i] = c;
}

protected Container getContainerAtIndex(int i) {
  return this.values[i];
}

// insert a new key, it is assumed that it does not exist
//arraycopy https://blog.csdn.net/wenzhi20102321/article/details/78444158
void insertNewKeyValueAt(int i, char key, Container value) {
  extendArray(1);
  //i后面的原地移动一位
  System.arraycopy(keys, i, keys, i + 1, size - i);
  keys[i] = key;
  //i后面的原地移动一位
  System.arraycopy(values, i, values, i + 1, size - i);
  values[i] = value;
  size++;
}

三种Container

ArrayContainer

/**
 * ArrayContainer 元素有序
       protected int cardinality = 0;
      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) {
      //Arrays.copyOf(this.content, newCapacity)
      //基数越大，扩容倍数越小
      increaseCapacity();
    }
    //x是最大值，在末尾写入
    content[cardinality++] = x;
  } else {
    //x是中间值，插入。相同值被忽略
    int loc = Util.unsignedBinarySearch(content, 0, cardinality, x);
    if (loc < 0) {
      // Transform the ArrayContainer to a BitmapContainer
      // when cardinality = DEFAULT_MAX_SIZE
      if (cardinality >= DEFAULT_MAX_SIZE) {
        return toBitmapContainer().add(x);
      }
      if (cardinality >= this.content.length) {
        increaseCapacity();
      }
      // insertion : shift the elements > x by one position to
      // the right
      // and put x in it's appropriate place
      System.arraycopy(content, -loc - 1, content, -loc, cardinality + loc + 1);
      content[-loc - 1] = x;
      ++cardinality;
    }
  }
  return this;
}

BitmapContainer

/**
 * BitmapContainer
       final long[] bitmap;
       int cardinality;
 */
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;
}

public boolean contains(final char i) {
  return (bitmap[i >>> 6] & (1L << i)) != 0;
}

RunContainer

略

//取int的低16位，直接转换  
protected static char lowbits(int x) {
    return (char) x;
}

工作应用

需求

实现一个人群包发券；人群包数量级900w-10kw，一条数据发券业务处理时间为4ms，每天10h需要发券900w-5kw，且7天内只能发一次，如何发完？

分析

1、一条数据处理时间4ms，10h=10h60min60s*250个/s = 900w，达到最低要求

2、7天只能发一次，前一天发完了900w，第二天还要从头开始遍历，900w之后的完全没法遍历到，第二天发放数量为0，卡死。

3、bitmap这么快的数据结构还发不完，伪需求，打回处理，问题解决。

如何从技术角度分析，怎么提高量级？

能否多线程？

1、Roaring64NavigableMap提供的遍历方法只有iterator迭代器和foreach循环遍历，且是从头往后遍历，没法多线程处理；

map数据结构也是private，无法调用底层map进行多线程处理，卡住~

2、7天只能发一次，能用redis解决，但是依然要遍历之前发过的数据，耗时很大，卡住~

粗暴方案

既然前一天发完了900w，第二天依然顺序遍历，能否记录前一天的900w，第二天直接从900w开始发，节省4ms/个的时间？

回答：可以。用redis或者db均能实现。但是5kw的量级还是远远不够。

优化方案

既然卡点是private方法，能否访问private，然后像调用hashmap一样，然后实现多线程遍历？

回答：可以，使用反射拿到NavigableMap<Integer, BitmapDataProvider> highToBitmap，根据前面的源码分析，可以实现一个多线程遍历map的key，根据key进行遍历。

//获取底层NavigableMap
Method m = Roaring64NavigableMap.class.getDeclaredMethod("getHighToBitmap", null);
m.setAccessible(true);
NavigableMap<Integer, BitmapDataProvider>
                    highToBitmap = (NavigableMap<Integer, BitmapDataProvider>) m.invoke(crowdListBitMap, null);

for (Entry<Integer, BitmapDataProvider> map : highToBitmap.entrySet()) {
  BitmapDataProvider provider = map.getValue();
  try {
    Field f = RoaringBitmap.class.getDeclaredField("highLowContainer");
    f.setAccessible(true);
    RoaringArray highLowContainer = (RoaringArray) f.get(provider);

    Field size = RoaringArray.class.getDeclaredField("size");
    size.setAccessible(true);
    int arraySize = (int) size.get(highLowContainer);
    
    ListeningExecutorService executorService = null;

    //线程池处理
    for (int i = 0; i < arraySize; i++) {
      hiveScanExecutorService.submit(new Runnable() {
        @Override
        public void run() {
          //RoaringArray多线程遍历
          //获取RoaringArray的value[] ContainerPointer pointer = highLowContainer.getContainerPointer(i);
          //遍历次数 = pointer.getContainer().getCardinality()
        }
      });
    }
  } catch (Exception e) {
    System.out.println("1" + e);
  }
}

存在的问题

1、上述代码获取ContainerPointer时，ContainerPointer长度最大为2^16，岂不是要开2^16个线程处理？

2、基于userId遍历存在密集数据的问题，某一个ContainerPointer可能没有数据，很多ContainerPointer的2^16个数可能已经存满了，是否能优化线程的创建？防止CPU100%？

继续优化

1、合理设置线程池，CPU密集型的线程个数设置不能过大，否则存在线程空转

2、动态设置线程的数量，合理配置新建线程的时机。

如优化前每天能发900w，设置10个线程理论一天能发9kw，满足要求。

3、ContainerPointer问题

根据getContainer().getCardinality()实际数据大小，记录每个线程处理map的startIndex和endIndex，使得index内的数据量保持在900w区间，解决线程过多的问题

总结

发现问题的卡点，多想想是否存在解决的办法。

能否有更好更优解决问题的方式。

多读源码。