演算法篇之BitMap原理與改造,利與弊的取捨

  • 2019 年 11 月 14 日
  • 筆記

記憶體不足,一直都是一件令人頭疼的事情,在有限的資源下,時間與空間的取捨是我們平時開發中思考最多的問題。無論是操作資料庫還是Redis快取,都沒有直接使用記憶體快取速度快,尤其是對大批量數據的處理。為能把記憶體使用率降低,並且提升業務的處理速度,我想了很多辦法,但效果都不顯著,如果數據量再增加,估計也只能加機器。今天分享下,我為適應業務需求對BIMAP演算法的改造思路。

BITMAP實現原理

一個byte類型有8bit,可以存儲8位數,如0,1,2,3,4,5,6,7;byte的0位為1,則說明0這個數存在這個byte中,byte的1位為1,則說明1這個數存在這個byte中。這與數組的原理是一樣的,只是把元素當成數組的下標,然後用0|1表示這個元素是否存在,0為false,1為true。

可是一個byte只能存在0~7的元素,那8以上的數怎麼存儲。這就是需要一個byte數組,byte[0]存儲0~7,byte[1]存儲8~15,以此類推。

比如存儲25這個數,先拿25右移3位(實際是除以8,但使用移位運算速度更快)拿到一個下標,這個下標就表示這個數存在哪個區間(byte數組的下標),再用這個數與8(一個byte=8bit)取模,可以拿到這個元素用byte的哪一個bit來存儲。

舉例:還是以25這個數字為例,將25寫入BitMap中。

假設byte數組的大小為10(不夠擴容,跟List一樣道理)。

計算出25在byte數組的下標:

int index = 25>>3 

等價於 25/8=3(數組下標從0開始,剛好不用減1);

計算出bit在byte中的下標:

int bitIndex = 25%8 = 1(byte元素下標也是從0開始);

所以得出

 int e = byteArray[index];   e |= 0x01<<bitIndex;   byteArray[index]=e;

那麼怎麼判斷一個元素是否存在呢?

byte dest = (0x01<<bitIndex);  return e & dest == dest;

移除元素25:

int e = byteArray[index];  e ^= 0x01<<bitIndex;  byteArray[index]=e;

注意:可根據需求實現是否執行緒安全的。

BITMAP存在的缺點

看起來BITMAP確實很優秀,存儲1~80000的數字,如果以int數組來存儲,需要80000*4(int=4byte)=160000byte的記憶體空間,而使用BitMap只需要80000/4/8=2500byte的記憶體空間,減少了64倍的記憶體空間。而且,它天然的擁有Set集合的去重功能,比如一個數組存在10個1,最終只會存儲一個1。

然而,這些優點都只是理論上的,最優的情況下。因為存儲的是連續的數字,如果數字不是連續的呢,比如我就存儲兩個元素,一個是1,一個是10000,那它依然佔據了1000個元素的存儲空間,是不是利變成了弊。再假如,我存儲的數字是10000~40000,那是不是10000之前的空間就是多餘的,浪費掉了?

所以BitMap的使用尤其要根據業務場景選擇,如果要存儲的整數數據沒有連續性,或者元素個數非常少的情況的,就不要考慮這種方案。可以用在存儲資料庫記錄的自增唯一標識,這種都會是連續性的,而且也會比較集中,不會出現1~100就到3000~4000的這種間隔很大的情況。

而第二個缺點,假設存儲數字是從40000開始的這種情況,可以將存儲的元素減去起始值再存儲。但除非是排序好的數字,否則你還得先提前知道你要存儲的這些數據中的最小值。怎麼解決,假設需要存儲的數據是1000~4000,那起始值就設置為1000,比如存儲元素1003,先再1003-1000得到3,再存儲3。判斷一個元素是否存在或者移除一個元素也是先將目標元素減去起始值再判斷。

改造後的BITMAP

接下來介紹,我為適應業務而改造的BITMAP。就是為了解決無法預估最小值的情況。

先說下原理。改造後的BitMap可以在new的時候指定一個中間值,如果不指定,那就在寫第一個元素的時候,將第一個元素設置為中間值。中間值與起始值的目的等同。為的是實現以中心分散的集中存儲方式,更高效的節省BitMap的記憶體使用。借用負無窮到0,0到正無窮的思路。定義兩個byte數組(不是一個了),小於中間值的存在一個數組,大於等於中間值的存儲在另一個數組。

這就是改造後的BitMap。在存儲元素的時候,實際是先使用中間值減去元素值得出目標值,再存儲這個目標值。看個例子,比如中間值設置為100,現在要存儲的元素是134,存儲過程如下。

計算出134在byte數組的下標:

int index = (100[中間值] - 134)>> 3 

等價於 -34/8=4(數組下標從0開始,剛好不用減1);

說明:index為負數=>存儲在第一個byte數組;

index為正數=>存在在第二個byte數組;

計算出bit在byte中的下標:

int bitIndex = |-34|%8 = 2(byte元素下標也是從0開始);

所以得出

int e = leftByteArray[index];  e |= 0x01<<bitIndex;  leftByteArray[index]=e;

獻上完整的工具類程式碼

/**   * @author wujiuye   * @version 1.0 on 2019/9/25 {描述:}   */  public class IntegerBitMap implements BitMap<Integer> {        private int mid;      private byte[] left;      private byte[] right;      private int leftSize = 8;      private int rightSzie = 8;      private int multiplier = 2;      private volatile int realSize;        public IntegerBitMap() {        }        public IntegerBitMap(int mid) {          this.mid = mid;          left = new byte[leftSize];          right = new byte[rightSzie];      }        public IntegerBitMap(int mid, int multiplier) {          this(mid);          this.multiplier = multiplier;      }        /**       * 推薦使用這種方式,使用首個插入的元素作為中間值,這樣分布更均勻       *       * @param obj       */      private synchronized void midByFist(Integer obj) {          if (realSize == 0) {              if (mid == 0) {                  this.mid = obj;              }          }      }        /**       * 數組下標,正數是left的下標,負數則是right的下標       *       * @param obj       * @return       */      private int index(Integer obj) {          if (obj == null) {              throw new NullPointerException("obj is null");          }          int intObj = mid - obj;          return intObj >> 3;      }        private synchronized void ensureSize(int index) {          int size = Math.abs(index);          int newSize;          if (index >= 0 && size >= leftSize) {              newSize = leftSize * multiplier > size ? leftSize * multiplier : (size + 1);              byte[] newLeft = new byte[newSize];              System.arraycopy(left, 0, newLeft, 0, Math.min(leftSize, newLeft.length));              left = newLeft;              leftSize = newLeft.length;          }          if (index < 0 && size >= rightSzie) {              newSize = rightSzie * multiplier > size ? rightSzie * multiplier : (size + 1);              byte[] newRight = new byte[newSize];              System.arraycopy(right, 0, newRight, 0, Math.min(rightSzie, newRight.length));              right = newRight;              rightSzie = newRight.length;          }      }        @Override      public synchronized boolean set(Integer obj) {          midByFist(obj);          int byteIndex = index(obj);          ensureSize(byteIndex);          byte map = byteIndex >= 0 ? left[byteIndex] : right[-byteIndex];          int inByteIndex = Math.abs(mid - obj) % 8;          // 0x0000 0000          byte target = (byte) (0x01 << inByteIndex);          boolean isExist = (map & target) == target;          map |= target;          if (byteIndex >= 0) {              left[byteIndex] = map;          } else {              right[-byteIndex] = map;          }          if (!isExist) {              realSize++;          }          return isExist;      }        /**       * 判斷元素是否存在(時間複雜度比HashMap更優)       *       * @param obj       * @return       */      @Override      public synchronized boolean contains(Integer obj) {          midByFist(obj);          int byteIndex = index(obj);          ensureSize(byteIndex);          byte map = byteIndex >= 0 ? left[byteIndex] : right[-byteIndex];          int inByteIndex = Math.abs(mid - obj) % 8;          // 0x0000 0000          byte target = (byte) (0x01 << inByteIndex);          return (map & target) == target;      }        @Override      public synchronized boolean remove(Integer obj) {          midByFist(obj);          int byteIndex = index(obj);          ensureSize(byteIndex);          byte map = byteIndex >= 0 ? left[byteIndex] : right[-byteIndex];          int inByteIndex = Math.abs(mid - obj) % 8;          // 0x0000 0000          byte target = (byte) (0x01 << inByteIndex);          boolean isExist = (map & target) == target;          if (isExist) {              map ^= target;              if (byteIndex >= 0) {                  left[byteIndex] = map;              } else {                  right[-byteIndex] = map;              }              realSize--;          }          return isExist;      }        @Override      public synchronized void reset() {          leftSize = 8;          rightSzie = 8;          left = new byte[leftSize];          right = new byte[rightSzie];          mid = 0;          realSize = 0;      }        /**       * 將BitMap轉為數組       *       * @return       */      @Override      public Integer[] toArray() {          Integer[] array = new Integer[realSize];          int index = 0;          index = writeToArray(right, array, index);          writeToArray(left, array, index);          return array;      }        private int writeToArray(byte[] source, Integer[] array, int posIndex) {          for (int i = 0; i < source.length; i++) {              for (int j = 0; j < 8; j++) {                  byte r = source[i];                  byte target = (byte) (0x01 << j);                  if ((r & target) == target) {                      array[posIndex++] = (i << 3) + j;                  }              }          }          return posIndex;      }

Tags: