前提
本文主要内容是分析
JDK
中的
BitMap
實作之
java.util.BitSet
的源碼實作，基于
JDK11
編寫，其他版本的
JDK
不一定合适。
文中的圖比特低位實際應該是在右邊，但是為了提高閱讀體驗，筆者把低位改在左邊了。
什麼是BitMap
BitMap
，直譯為位圖，是一種資料結構，代表了有限域中的稠集（
Dense Set
），每一個元素至少出現一次，沒有其他的資料和元素相關聯。在索引，資料壓縮等方面有廣泛應用（來源于維基百科詞條）。計算機中
1 byte = 8 bit
，一個比特（
bit
，稱為比特或者位）可以表示
1
或者
兩種值，通過一個比特去标記某個元素的值，而
KEY
INDEX
就是該元素，構成一張映射關系圖。因為采用了
Bit
作為底層存儲資料的機關，是以可以極大地節省存儲空間。
在
Java
中，一個
int
類型的整數占
4
位元組，
16
比特，
int
的最大值也就是
20
多億（具體是
2147483647
）。假設現在有一個需求，在
20
億整數中判斷某個整數
m
是否存在，要求使用記憶體必須小于或者等于
4GB
。如果每個整數都使用
int
存儲，那麼存放
20
億個整數，需要
20億 * 4byte /1024/1024/1024
約等于
7.45GB
，顯然無法滿足需求。如果使用
BitMap
，隻需要
20億 bit
記憶體，也就是
20億/8/1024/1024/1024
0.233GB
。在資料量極大的情況下，資料集具備有限狀态，可以考慮使用
BitMap
存儲和進行後續計算等處理。現在假設用
byte
數組去做
BitMap
的底層存儲結構，初始化一個容量為
16
的
BitMap
執行個體，示例如下：
可見目前的
byte
數組有兩個元素
bitmap[0]
（虛拟下标為
[0,7]
）和
bitmap[1]
[8,15]
）。這裡假定使用上面構造的這個
BitMap
執行個體去存儲客戶
ID
和客戶性别關系（比特為
1
代表男性，比特為
代表女性），把
ID
等于
3
的男性客戶和
ID
10
的女性客戶添加到
BitMap
中：
由于
1 byte = 8 bit
，通過客戶
ID
除以
8
就可以定位到需要存放的
byte
數組索引，再通過客戶
ID
基于
8
取模，就可以得到需要存放的
byte
數組中具體的
bit
的索引：
# ID等于3的男性客戶
邏輯索引 = 3
byte數組索引 = 3 / 8 = 0
bit索引 = 3 % 8 = 3
=> 也就是需要存放在byte[0]的下标為3的比特上，該比特設定為1
# ID等于10的女性客戶
邏輯索引 = 10
byte數組索引 = 10 / 8 = 1
bit索引 = 10 % 8 = 2
=> 也就是需要存放在byte[1]的下标為2的比特上，該比特設定為0
然後分别判斷客戶
ID
為
3
10
的客戶性别：
如果此時再添加一個客戶
ID
17
的男性使用者，由于舊的
BitMap
隻能存放
16
個比特，是以需要擴容，判斷
byte
數組中隻需新增一個
byte
元素（
byte[2]
）即可：
原則上，底層的
byte
數組可以不停地擴容，當
byte
數組長度達到
Integer.MAX_VALUE
，
BitMap
的容量達到最大值。
BitSet簡單使用
java.util.BitSet
雖然名字上稱為
Set
，但實際上它就是
JDK
中内置的
BitMap
實作，1這個類算是一個十分古老的類，從注釋上看是
JDK1.0
引入的，不過大部分方法是
JDK1.4
之後新添加或者更新的。以前一小節的例子基于
BitSet
做一個
Demo
：
public class BitSetApp {
public static void main(String[] args) {
BitSet bitmap = new BitSet(16);
bitmap.set(3, Boolean.TRUE);
bitmap.set(11, Boolean.FALSE);
System.out.println("Index 3 of bitmap => " + bitmap.get(3));
System.out.println("Index 11 of bitmap => " + bitmap.get(11));
bitmap.set(17, Boolean.TRUE);
// 這裡不會觸發擴容，因為BitSet中底層存儲數組是long[]
System.out.println("Index 17 of bitmap => " + bitmap.get(17));
}
}
// 輸出結果
Index 3 of bitmap => true
Index 11 of bitmap => false
Index 17 of bitmap => true
API
使用比較簡單，為了滿足其他場景，
BitSet
還提供了幾個實用的靜态工廠方法用于構造執行個體，範圍設定和清除比特值和一些集合運算等，這裡不舉例，後面分析源碼的時候會詳細展開。
BitSet源碼分析
前文提到，
BitMap
如果使用
byte
數組存儲，當新添加元素的邏輯下标超過了初始化的
byte
數組的最大邏輯下标就必須進行擴容。為了盡可能減少擴容的次數，除了需要按實際情況定義初始化的底層存儲結構，還應該選用能夠"承載"更多比特的資料類型數組，是以在
BitSet
中底層的存儲結構選用了
long
數組，一個
long
整數占
64
比特，位長是一個
byte
整數的
8
倍，在需要處理的資料範圍比較大的場景下可以有效減少擴容的次數。後文為了簡化分析過程，在模拟底層
long
數組變化時候會使用盡可能少的元素去模拟。
BitSet
頂部有一些關于其設計上的注釋，這裡簡單羅列概括成幾點：
-
BitSet
-
BitSet
false
-
BitSet
size
length
-
BitSet
按照以往分析源碼的習慣，先看
BitSet
的所有核心成員屬性：
public class BitSet implements Cloneable, java.io.Serializable {
// words是long數組，一個long整數為64bit，2^6 = 64，這裡選取6作為words的尋址參數，可以基于邏輯下标快速定位到具體的words中的元素索引
private static final int ADDRESS_BITS_PER_WORD = 6;
// words中每個元素的比特數，十進制值是64
private static final int BITS_PER_WORD = 1 << ADDRESS_BITS_PER_WORD;
// bit下标掩碼，十進制值是63
private static final int BIT_INDEX_MASK = BITS_PER_WORD - 1;
// 掩碼，十進制值-1，也就是64個比特全是1，用于部分word掩碼的左移或者右移
private static final long WORD_MASK = 0xffffffffffffffffL;
/**
* 序列化相關，略過
*/
private static final ObjectStreamField[] serialPersistentFields = {
new ObjectStreamField("bits", long[].class),
};
/**
* 底層的比特存儲結構，long數組，同時也是序列化字段"bits"的對應值
*/
private long[] words;
/**
* 已經使用的words數組中的元素個數，注釋翻譯：在目前BitSet的邏輯長度中的word（words的元素）個數，瞬時值
*/
private transient int wordsInUse = 0;
/**
* 标記words數組的長度是否使用者
*/
private transient boolean sizeIsSticky = false;
// JDK 1.0.2使用的序列化版本号
private static final long serialVersionUID = 7997698588986878753L;
// 暫時省略其他方法
}
接着看
BitSet
的幾個輔助方法：
// 基于bit的邏輯下标定位words中word元素的索引，直接右移6位
// 舉例：bitIndex = 3，那麼bitIndex >> ADDRESS_BITS_PER_WORD => 0，說明定位到words[0]
// 舉例：bitIndex = 35，那麼bitIndex >> ADDRESS_BITS_PER_WORD => 1，說明定位到words[1]
private static int wordIndex(int bitIndex) {
return bitIndex >> ADDRESS_BITS_PER_WORD;
}
// 每個公共方法都必須保留這些不變量，内部變量的恒等式校驗，字面意思就是每個公共方法必須調用此恒等式校驗
// 第一條恒等式：目前BitSet為空或者最後一個words元素不能為0（其實就是目前BitSet不為空）
// 第二條恒等式：wordsInUse邊界檢查，範圍是[0, words.length]
// 第三條恒等式：wordsInUse或者等于words.length，意味着用到了所有words的元素；或者words[wordsInUse] == 0，意味着words中索引為[wordsInUse, words.length - 1]的元素都沒有被使用
private void checkInvariants() {
assert(wordsInUse == 0 || words[wordsInUse - 1] != 0);
assert(wordsInUse >= 0 && wordsInUse <= words.length);
assert(wordsInUse == words.length || words[wordsInUse] == 0);
}
// 重新計算wordsInUse的值，也就是重新整理已使用的words元素計算值
// 基于目前的wordsInUse - 1向前周遊到i = 0，找到最近一個不為0的words[i]，然後重新指派為i + 1，這裡i是words數組的索引
// wordsInUse其實是words數組最後一個不為0的元素的下标加1，或者說用到的words的元素個數，稱為邏輯容量(logical size)
private void recalculateWordsInUse() {
// Traverse the bitset until a used word is found
int i;
for (i = wordsInUse-1; i >= 0; i--)
if (words[i] != 0)
break;
wordsInUse = i+1; // The new logical size
}
然後看
BitSet
的構造函數和靜态工廠方法：
// 預設的公共構造方法，比特表的邏輯長度為64，words數組長度為2，标記sizeIsSticky為false，也就是比特表的長度不是使用者自定義的
public BitSet() {
initWords(BITS_PER_WORD);
sizeIsSticky = false;
}
// 自定義比特表邏輯長度的構造方法，該長度必須為非負整數，标記sizeIsSticky為true，也就是比特表的長度是由使用者自定義的
public BitSet(int nbits) {
if (nbits < 0)
throw new NegativeArraySizeException("nbits < 0: " + nbits);
initWords(nbits);
sizeIsSticky = true;
}
// 初始化words數組，數組的長度為length = (nbits - 1) / 64 + 1
// 例如nbits = 16，相當于long[] words = new long[(16 - 1) / 64 + 1] => new long[1];
// 例如nbits = 65，相當于long[] words = new long[(65 - 1) / 64 + 1] => new long[2];
// 以此類推
private void initWords(int nbits) {
words = new long[wordIndex(nbits-1) + 1];
}
// 直接自定義底層的words數組構造方法，标記所有words的元素都被使用
private BitSet(long[] words) {
this.words = words;
this.wordsInUse = words.length;
checkInvariants();
}
// 直接自定義底層的words數組構造方法，這個構造方法和上一個方法不一樣，會從入參long數組從後面開始周遊，直到周遊到第一個元素或者不為0的元素，這樣可以盡量截斷無用的高位的0元素
// 簡單來說就是相當于：BitSet.valueOf(new long[]{1L, 0L}) = 移除後面的0元素 => BitSet.valueOf(new long[]{1L})
public static BitSet valueOf(long[] longs) {
int n;
for (n = longs.length; n > 0 && longs[n - 1] == 0; n--)
;
return new BitSet(Arrays.copyOf(longs, n));
}
// 直接自定義底層的words數組構造方法，要求入參為LongBuffer類型，需要把LongBuffer => long[] words，方法和BitSet valueOf(long[] longs)處理邏輯一緻
public static BitSet valueOf(LongBuffer lb) {
lb = lb.slice();
int n;
for (n = lb.remaining(); n > 0 && lb.get(n - 1) == 0; n--)
;
long[] words = new long[n];
lb.get(words);
return new BitSet(words);
}
// 下面兩個構造方法都是基于byte數組從後面開始周遊，直到周遊到第一個元素或者不為0的元素，截斷出一個新的數組，然後轉化為long數組構造BitSet執行個體
public static BitSet valueOf(byte[] bytes) {
return BitSet.valueOf(ByteBuffer.wrap(bytes));
}
public static BitSet valueOf(ByteBuffer bb) {
// 小端位元組排序
bb = bb.slice().order(ByteOrder.LITTLE_ENDIAN);
int n;
// 從後向前周遊獲到第一個元素或者第一個不為0的元素
for (n = bb.remaining(); n > 0 && bb.get(n - 1) == 0; n--)
;
// 這裡需要考慮位元組容器中的位元組元素個數不是8的倍數的情況
long[] words = new long[(n + 7) / 8];
// 截斷後面的0元素
bb.limit(n);
int i = 0;
// 剩餘元素個數大于等于8時候，按照64位去讀取
while (bb.remaining() >= 8)
words[i++] = bb.getLong();
// 剩餘元素個數小于8時候，按照byte讀取，并且通過掩碼計算和左移填充到long數組元素中
for (int remaining = bb.remaining(), j = 0; j < remaining; j++)
words[i] |= (bb.get() & 0xffL) << (8 * j);
return new BitSet(words);
}
這裡構造函數的源碼不是十分複雜，比較繁瑣的是靜态工廠方法
BitSet valueOf(ByteBuffer bb)
，這裡舉例推演一下：
ByteBuffer byteBuffer = ByteBuffer.allocate(10);
byteBuffer.order(ByteOrder.LITTLE_ENDIAN);
byteBuffer.putLong(1L);
byteBuffer.put((byte)3);
byteBuffer.put((byte)1);
byteBuffer.flip();
BitSet bitSet = BitSet.valueOf(byteBuffer);
System.out.println(bitSet.size());
System.out.println(bitSet.length());
// 輸出結果
128
73
過程如下：
接着看正常的
set
、
get
和
clear
方法：
// 設定指定的邏輯索引的比特為true
public void set(int bitIndex) {
// 比特邏輯索引必須大于等于0
if (bitIndex < 0)
throw new IndexOutOfBoundsException("bitIndex < 0: " + bitIndex);
// 計算words數組元素的索引
int wordIndex = wordIndex(bitIndex);
// 判斷是否需要擴容，如果需要則進行words數組擴容
expandTo(wordIndex);
// 相當于words[wordIndex] = words[wordIndex] | (1L << bitIndex)
// 注意long的左移如果超過63位會溢出，也就是1L << 64 => 1L,1L << 65 => 1L << 1,1L << 66 => 1L << 2... 以此類推
// 這裡相當于把左移結果直接和對應的words元素進行或運算，前者因為是基于1進行左移，二進制數一定是隻有一個比特為1，其他比特都是0的64比特二級制序列，或運算會讓對應的words元素與前者對應的比特為1的比特值設定為1，并且重新指派對應的words元素
// 類似于這樣：0000 0000 | 0000 1000 => 0000 1000
words[wordIndex] |= (1L << bitIndex); // Restores invariants
// 不變量恒等式斷言校驗
checkInvariants();
}
// 基于計算的words數組元素下标進行擴容
private void expandTo(int wordIndex) {
// 計算目前的words元素下标需要的words數組長度
int wordsRequired = wordIndex + 1;
// 如果目前的words元素下标需要的words數組長度大于目前已經使用的words數組中的元素個數，則進行擴容（未必一定擴容數組，擴容方法裡面還有一層判斷）
if (wordsInUse < wordsRequired) {
// 基于需要的words數組長度進行擴容
ensureCapacity(wordsRequired);
// 重置目前已經使用的words數組中的元素個數
wordsInUse = wordsRequired;
}
}
// 基于計算的words數組元素下标進行擴容，滿足前提下進行數組拷貝
private void ensureCapacity(int wordsRequired) {
// 目前words數組長度比需要的words數組長度小，則進行擴容
if (words.length < wordsRequired) {
// 配置設定的新數組的長度是舊words數組元素和傳入的需要的words數組長度之間的最大值
int request = Math.max(2 * words.length, wordsRequired);
// 數組擴容
words = Arrays.copyOf(words, request);
// 因為已經進行了擴容，是以要标記比特表的長度不是使用者自定義的
sizeIsSticky = false;
}
}
// 擷取指定的邏輯索引的比特的狀态
public boolean get(int bitIndex) {
// 比特邏輯索引必須大于等于0
if (bitIndex < 0)
throw new IndexOutOfBoundsException("bitIndex < 0: " + bitIndex);
// 不變量恒等式斷言校驗
checkInvariants();
// 計算words數組元素的索引
int wordIndex = wordIndex(bitIndex);
// words數組元素的索引必須小于正在使用的words元素個數，并且把1L左移bitIndex結果直接和對應的words元素進行與運算的結果不是全部比特為0則傳回true，否則傳回false
// 類似于這樣（傳回true的場景）：0000 1010 & 0000 1000 => 0000 1000 => 說明定位到的words中的元素曾經通過set方法設定過對應1L << bitIndex的比特為1
// 類似于這樣（傳回false的場景）：0000 0110 & 0000 1000 => 0000 0000 => 說明定位到的words中的元素未曾通過set方法設定過對應1L << bitIndex的比特為1，對應比特使用的是預設值0
return (wordIndex < wordsInUse) && ((words[wordIndex] & (1L << bitIndex)) != 0);
}
// 設定指定的邏輯索引的比特為false
public void clear(int bitIndex) {
// 比特邏輯索引必須大于等于0
if (bitIndex < 0)
throw new IndexOutOfBoundsException("bitIndex < 0: " + bitIndex);
// 計算words數組元素的索引
int wordIndex = wordIndex(bitIndex);
// 如果words數組元素的索引大于等于正在使用的words元素個數，說明該邏輯下标的比特處于初始化狀态還未被使用，不用處理
if (wordIndex >= wordsInUse)
return;
// 相當于words[wordIndex] = words[wordIndex] & (~(1L << bitIndex))
// 把左移結果各比特取反然後和對應的words元素進行與運算，再重新指派到對應的words元素
// 類似于：0000 1100 & (~(0000 1000)) => 0000 1100 & 1111 0111 => 0000 0100
words[wordIndex] &= ~(1L << bitIndex);
// 重新計算wordsInUse的值，也就是重新整理已使用的words元素計算值
recalculateWordsInUse();
// 不變量恒等式斷言校驗
checkInvariants();
}
這裡模拟一下
set
方法的過程：
接着看集合運算相關的方法：
// 判斷兩個BitSet是否存在交集，這是一個判斷方法，不會修改目前BitSet的結構
public boolean intersects(BitSet set) {
// 對比目前BitSet執行個體和入參BitSet執行個體使用的words元素數量，取較小值作為周遊基準
for (int i = Math.min(wordsInUse, set.wordsInUse) - 1; i >= 0; i--)
// 周遊和對比每一個words數組的元素，隻要滿足與運算結果不為0就傳回true（這個條件是很寬松的，隻要底層邏輯比特表剛好兩個BitSet執行個體在同一邏輯索引的比特都為1即可滿足）
if ((words[i] & set.words[i]) != 0)
return true;
return false;
}
// AND運算，底層是兩個BitSet執行個體對應索引的words數組元素進行與運算，直覺來看就是計算兩個BitSet執行個體的交集，存放在本BitSet執行個體中
public void and(BitSet set) {
// 入參為本BitSet執行個體，不進行處理
if (this == set)
return;
// 如果目前BitSet執行個體已經使用的words數組元素數量比傳入的多，那麼目前BitSet執行個體把多出來那部分words數組元素置為0
while (wordsInUse > set.wordsInUse)
words[--wordsInUse] = 0;
// 周遊目前的BitSet執行個體的words數組已使用的元素，與傳入的BitSet執行個體的相同索引元素進行與運算和重新指派
for (int i = 0; i < wordsInUse; i++)
words[i] &= set.words[i];
// 重新計算wordsInUse的值，也就是重新整理已使用的words元素計算值
recalculateWordsInUse();
// 不變量恒等式斷言校驗
checkInvariants();
}
// OR運算，底層是兩個BitSet執行個體對應索引的words數組元素進行或運算，直覺來看就是計算兩個BitSet執行個體的并集，存放在本BitSet執行個體中
public void or(BitSet set) {
// 入參為本BitSet執行個體，不進行處理
if (this == set)
return;
// 計算兩個BitSet中words數組已使用元素的公共部分，其實也就是較小的wordsInUse
int wordsInCommon = Math.min(wordsInUse, set.wordsInUse);
// 目前的BitSet執行個體的words數組已使用的元素數量比傳入的BitSet執行個體小，以傳入的執行個體為準進行擴容，并且拷貝其wordsInUse值
if (wordsInUse < set.wordsInUse) {
ensureCapacity(set.wordsInUse);
wordsInUse = set.wordsInUse;
}
// 兩個BitSet執行個體words數組已使用元素的公共部分分别按索引進行或運算，指派在目前BitSet執行個體對應索引的words元素
for (int i = 0; i < wordsInCommon; i++)
words[i] |= set.words[i];
// 如果傳入的BitSet執行個體還有超出words數組已使用元素的公共部分，這部分words數組元素也拷貝到目前的BitSet執行個體中，因為前面有擴容判斷，走到這裡目前BitSet執行個體的wordsInUse大于等于傳入的BitSet執行個體的wordsInUse
if (wordsInCommon < set.wordsInUse)
System.arraycopy(set.words, wordsInCommon,
words, wordsInCommon,
wordsInUse - wordsInCommon);
// 不變量恒等式斷言校驗
checkInvariants();
}
// XOR運算，底層是兩個BitSet執行個體對應索引的words數組元素進行異或運算，實作和OR基本相似，完成處理後需要重新計算目前BitSet執行個體的wordsInUse值
public void xor(BitSet set) {
int wordsInCommon = Math.min(wordsInUse, set.wordsInUse);
if (wordsInUse < set.wordsInUse) {
ensureCapacity(set.wordsInUse);
wordsInUse = set.wordsInUse;
}
// Perform logical XOR on words in common
for (int i = 0; i < wordsInCommon; i++)
words[i] ^= set.words[i];
// Copy any remaining words
if (wordsInCommon < set.wordsInUse)
System.arraycopy(set.words, wordsInCommon,
words, wordsInCommon,
set.wordsInUse - wordsInCommon);
recalculateWordsInUse();
checkInvariants();
}
// AND NOT運算，底層是兩個BitSet執行個體對應索引的words數組元素進行與運算之前傳入BitSet執行個體對應索引的words數組元素先做非運算，過程和AND運算類似
public void andNot(BitSet set) {
// Perform logical (a & !b) on words in common
for (int i = Math.min(wordsInUse, set.wordsInUse) - 1; i >= 0; i--)
words[i] &= ~set.words[i];
recalculateWordsInUse();
checkInvariants();
}
and
接着看搜尋相關方法（
nextXXXBit
previousYYYBit
），這裡以
nextSetBit(int fromIndex)
方法為例子：
// 以比特邏輯索引fromIndex為起點，向後搜尋并且傳回第一個狀态為true的比特的邏輯索引，搜尋失敗則傳回-1
public int nextSetBit(int fromIndex) {
// 起始的比特邏輯索引必須大于等于0
if (fromIndex < 0)
throw new IndexOutOfBoundsException("fromIndex < 0: " + fromIndex);
// 不變量恒等式斷言校驗
checkInvariants();
// 基于起始的比特邏輯索引計算words數組元素的索引
int u = wordIndex(fromIndex);
// words數組元素的索引超過已經使用的words數組元素數量，說明數組越界，直接傳回-1
if (u >= wordsInUse)
return -1;
// 這裡和之前的set方法的左移類似，不過使用了-1L進行左移，例如-1L << 2 => 1111 1111 << 2 => 1111 1100（這裡假設限制長度8，溢出的高位舍棄）
// 舉例：0000 1010 & (1111 1111 << 2) => 0000 1010 & 1111 1100 => 0000 1000（索引值為4，目前這裡不一定得到存在比特為1的結果）
long word = words[u] & (WORD_MASK << fromIndex);
// 基于得到的word進行周遊
while (true) {
// 說明word中存在比特為1，計算和傳回該比特的邏輯索引
if (word != 0)
// 基于起始的比特邏輯索引計算words數組元素的索引 * 64 + word中低位連續為0的比特數量
return (u * BITS_PER_WORD) + Long.numberOfTrailingZeros(word);
// 說明word中全部比特為0，則u需要向後遞增，等于wordsInUse說明越界傳回-1，這個if存在指派和判斷兩個邏輯
if (++u == wordsInUse)
return -1;
word = words[u];
}
}
nextSetBit(int fromIndex)
方法先查找
fromIndex
所在的
words
數組元素，不滿足後再向後進行檢索，該方法注釋中還給出了一個經典的使用例子，這裡摘抄一下：
BitSet bitmap = new BitSet();
// add element to bitmap
// ... bitmap.set(1993);
for (int i = bitmap.nextSetBit(0); i >= 0; i = bitmap.nextSetBit(i + 1)) {
// operate on index i here
if (i == Integer.MAX_VALUE) {
break; // or (i+1) would overflow
}
}
最後看規格屬性相關的一些
Getter
// 擷取BitSet中的比特總數量
public int size() {
// words數組長度 * 64
return words.length * BITS_PER_WORD;
}
// 擷取BitSet中的比特值為1的總數量
public int cardinality() {
int sum = 0;
// 擷取每一個已使用的words數組元素的比特為1的數量然後累加
for (int i = 0; i < wordsInUse; i++)
sum += Long.bitCount(words[i]);
return sum;
}
// 擷取BitSet的邏輯大小（不計算隻是初始化但是未使用的高位比特），簡單來說就是words[wordsInUse - 1]中去掉高位比特連續0的第一個比特值為1的邏輯索引，例如0001 1010，高位連續3個0，邏輯索引為4
public int length() {
if (wordsInUse == 0)
return 0;
return BITS_PER_WORD * (wordsInUse - 1) +
(BITS_PER_WORD - Long.numberOfLeadingZeros(words[wordsInUse - 1]));
}
其他按範圍設定或者清除值如
set(int fromIndex, int toIndex)
clear(int fromIndex, int toIndex)
等方法限于篇幅不進行詳細分析，套路大緻是相似的。
BitSet沒有解決的問題
ADDRESS_BITS_PER_WORD
的字段注釋中也提到過：
The choice of word size is determined purely by performance concerns
（
'word'
大小的選擇純粹是由性能考慮決定的）。這裡的尋址基礎值
6
的選擇是因為底層選用了
long
數組，盡可能減少底層數組的擴容次數。但是這裡存在一個比較沖突的問題，似乎在
JDK
中沒有辦法找到位寬比
long
大而且占用記憶體空間跟
long
等效的資料類型，像
byte
String
（底層是
char
數組）等等都會在擴容次數方面遠超
long
類型。因為底層是數組存儲結構，并且沒有限定數組中元素的下邊界和上邊界，在一些特定的場景下會浪費過多的無用記憶體空間。以前文提到過的例子做改造，如果要把
10
億到
20
億之間的整數全部加裝到
BitSet
執行個體中（這些值在
BitSet
的對應的邏輯比特索引的值都标記為
1
），那麼在
BitSet
執行個體的底層比特表中，邏輯索引
[0,10億)
的值都會是初始化值
，也就是約一半的
words[N]
的值都是
，這部分記憶體空間是完全被浪費掉的。在實際的業務場景中，很多時候業務的主鍵并不是使用資料庫的自增主鍵，而是使用通過
SnowFlake
這類算法生成的帶自增趨勢的數值主鍵，如果算法定義的基準時間戳比較大，那麼生成出來的值會遠超
int
類型的上限（使用
long
類型承載）。也就是
BitSet
沒有解決的問題如下：
- 問題一：比特表的邏輯索引值上限是
Integer.MAX_VALUE
Long.MAX_VALUE
JDK
length
int
java.lang.reflect.Array
- 問題二：
BitSet
[0,下邊界)
對于問題一，可以考慮做一個簡單的映射。假設現在需要存儲
[Integer.MAX_VALUE + 1,Integer.MAX_VALUE + 3]
到
BitSet
執行個體中，可以實際存儲
[1,3]
，處理完畢後，通過
long realIndex = (long) bitIndex + Integer.MAX_VALUE
恢複實際的索引值，這樣就可以邏輯上擴大
BitSet
的存儲範圍，猜測可以滿足
99%
以上的場景：
public class LargeBitSetApp {
public static void main(String[] args) {
long indexOne = Integer.MAX_VALUE + 1L;
long indexTwo = Integer.MAX_VALUE + 2L;
long indexThree = Integer.MAX_VALUE + 3L;
BitSet bitmap = new BitSet(8);
// set(int fromIndex, int toIndex) => [fromIndex,toIndex)
bitmap.set((int) (indexOne - Integer.MIN_VALUE), (int) (indexThree - Integer.MIN_VALUE) + 1);
System.out.printf("Index = %d,value = %s\n", indexOne, bitmap.get((int) (indexOne - Integer.MIN_VALUE)));
System.out.printf("Index = %d,value = %s\n", indexTwo, bitmap.get((int) (indexTwo - Integer.MIN_VALUE)));
System.out.printf("Index = %d,value = %s\n", indexThree, bitmap.get((int) (indexThree - Integer.MIN_VALUE)));
}
}
// 輸出結果
Index = 2147483648,value = true
Index = 2147483649,value = true
Index = 2147483650,value = true
對于問題二，已經有現成的實作，就是類庫
RoaringBitmap
，倉庫位址是：https://github.com/RoaringBitmap/RoaringBitmap。該類庫被
Apache
的多個大資料元件使用，經得起生産環境的考驗。引入依賴如下：
<dependency>
<groupId>org.roaringbitmap</groupId>
<artifactId>RoaringBitmap</artifactId>
<version>0.9.23</version>
</dependency>
簡單使用：
public class RoaringBitmapApp {
public static void main(String[] args) {
RoaringBitmap bitmap = RoaringBitmap.bitmapOf(1, 2, 3, Integer.MAX_VALUE, Integer.MAX_VALUE - 1);
System.out.println(bitmap.contains(Integer.MAX_VALUE));
System.out.println(bitmap.contains(1));
}
}
// 輸出結果
true
true
RoaringBitmap
區分了不同場景去建立底層的存儲容器：
- 稀疏場景：使用
ArrayContainer
4096
- 密集場景：使用
BitmapContainer
java.util.BitSet
4096
- 聚集場景（了解為前兩種場景的混合）：使用
RunContainer
小結
學習和分析
BitSet
的源碼是為了更好地在實際場景中處理有限狀态的大批量資料集合之間的運算，剛好在筆者的業務中有比對的場景，後面有機會的話在另一篇實戰文章中再詳細展開。
參考資料：
-
JDK11
- RoaringBitmap Performance Tricks
（本文完 c-4-d e-a-20220103）