一、使用的场景
日常业务中需要大量存储一些重复的字符串，例如每日签到用户、朋友圈点赞的好友、计算每日登录用户等。字符串无论长短不仅会浪费大量的存储资源，而且读取查询也耗时耗资源，那有没有一种存储方式对这一类场景进行优化呢。
二、什么原理
1、Bitmap如何解决这个问题
拿每日签到用户的场景举例，首先对用户id进行编号(offset)。
图1.png
这样我们只需要每天创建一个只存储位的数组，比如2023-12-8日只有zhangsan和zhaoliu进行了登录，我们只需要一个4位的字节数组来进行存储：
图2.png
这样就达到了减少存储的目的。
2、压缩Bitmap是如何做到压缩的
然而现实的场景中，存储的数据并不是连续的，而是稀疏的，而且创建Bitmap的时候，长度必须是最大的offset的长度，例如我们有64个用户，那么创建的位数组就是64位。
假如我们有1万个用户，但今天只有编号9999的用户登录了系统，那我们还是创建一个1万个Byte的数组，反而增加了内存的消耗。
为了更容易理解这个问题，我们还是以64位Bitmap为例，也就是最大支持存储64个用户，当我们只存储63时：
图3.png
先创建了长度为64位的数组，只在63这个位置上的存储是有意义的，其他位置都是0，那这种情况是不是可以进行优化呢？
我们把数组拆分成2组，1到32的用户存储到第1组，把33到64的用户存储到第二组。这个时候我们发现第一组的数组里全是0，那第一组是不是可以不创建？这样我们就用一个32位大小的数组存储了64位的数据。
接下来更进一步，把64位的数组拆成每16个一组，那么我们只需要创建最后一个分组的数组，也就是16位的数组来达到进一步压缩的目的。
第4组
图4.png
未命名流程图.jpg
三、怎么模拟实现
代码来简单的模拟实现压缩Bitmap
1、先创建一个类累存储单个的Bitmap
public class ArrayContainer {
public char[] values;
public ArrayContainer(int initialCapacity) {
this.values = new char[initialCapacity];
}
}
2、实现压缩Bitmap的逻辑
public class CompressedBitmap {
public static final char Y = 1;
public static final char N = 0;
public static final int ARRAY_CONTAINER_SIZE = 16;
ArrayContainer[] containers;
public CompressedBitmap(long initialCapacity) {
int containersSize = (int)(initialCapacity / ARRAY_CONTAINER_SIZE);
this.containers = new ArrayContainer[containersSize];
}
public void add(long offset){
int shardIndex = getShardIndex(offset);
ArrayContainer container = containers[shardIndex];
if(container == null){
container = new ArrayContainer(ARRAY_CONTAINER_SIZE);
}
int shardOffset = getShardOffset(offset);
container.values[shardOffset] = Y;
containers[shardIndex] = container;
}
public int get(long offset){
int shardIndex = getShardIndex(offset);
ArrayContainer container = containers[shardIndex];
if(container == null){
return N;
}
int shardOffset = getShardOffset(offset);
if(Y == container.values[shardOffset]){
return Y;
}
return N;
}
public int getShardIndex(long offset){
return (int)((offset - 1) / ARRAY_CONTAINER_SIZE);
}
public int getShardOffset(long offset){
return (int)(offset % ARRAY_CONTAINER_SIZE);
}
public static void main(String[] args) {
System.out.println(2 / ARRAY_CONTAINER_SIZE);
}
}
3、测试使用
public static void main(String[] args) {
CompressedBitmap bitmap = new CompressedBitmap(64);
bitmap.add(63);
System.out.println(bitmap.get(63));
System.out.println(bitmap.get(64));
System.out.println(bitmap.get(2));
}
四、Redis实现压缩Bitmap存储
Redis本身是支持Bitmap存储的，但是压缩的Bitmap是不支持的。根据上面的原理，同样可以把一个大的Bitmap转成一个个小的Bitmap来达到压缩的目的。
1、包装实体保存压缩位图的信息
public class CompressedBitInfo implements Serializable {
/**
* 真实offset
*/
private long sourceOffset;
/**
* 分桶的编号
*/
private long bucketIndex;
/**
* 桶内的offset
*/
private long bucketOffset;
public long getSourceOffset() {
return sourceOffset;
}
public void setSourceOffset(long sourceOffset) {
this.sourceOffset = sourceOffset;
}
public long getBucketIndex() {
return bucketIndex;
}
public void setBucketIndex(long bucketIndex) {
this.bucketIndex = bucketIndex;
}
public long getBucketOffset() {
return bucketOffset;
}
public void setBucketOffset(long bucketOffset) {
this.bucketOffset = bucketOffset;
}
@Override
public String toString() {
return "CompressedBitInfo{" + "sourceOffset=" + sourceOffset + ", bucketIndex=" + bucketIndex + ", bucketOffset=" + bucketOffset + '}';
}
}
2、工具类计算每个小桶的Bitmap
public class CompressedBitTookit {
public static final long DEFAULT_BUCKET_SIZE = 65536L;
public static CompressedBitInfo getBitInfo(long sourceOffset, long bucketSize) {
CompressedBitInfo bucketInfo = new CompressedBitInfo();
bucketInfo.setSourceOffset(sourceOffset);
long bucketIndex = sourceOffset / bucketSize;
long bucketOffset = sourceOffset % bucketSize;
bucketInfo.setBucketIndex(bucketIndex);
bucketInfo.setBucketOffset(bucketOffset);
return bucketInfo;
}
public static CompressedBitInfo getBitInfo(long sourceOffset) {
CompressedBitInfo bucketInfo = getBitInfo(sourceOffset, DEFAULT_BUCKET_SIZE);
return bucketInfo;
}
public static long getSourceOffset(long bucketIndex, long bucketSize, long bucketOffset) {
return bucketIndex * bucketSize + bucketOffset;
}
public static long getSourceOffset(long bucketIndex, long bucketOffset) {
return getSourceOffset(bucketIndex, DEFAULT_BUCKET_SIZE, bucketOffset);
}
3、读写Bitmap
public void setCompressedBit(String key, long offset, boolean value, int expireSeconds) {
CompressedBitInfo bitInfo = CompressedBitTookit.getBitInfo(offset);
String bitKey = getBitKey(key, bitInfo.getBucketIndex());
redis.setBit(bitKey, bitInfo.getBucketOffset(),value);
redis.expire(bitKey, expireSeconds);
}
public void setCompressedBit(String key, long offset, boolean value) {
CompressedBitInfo bitInfo = CompressedBitTookit.getBitInfo(offset);
String bitKey = getBitKey(key, bitInfo.getBucketIndex());
redis.setBit(bitKey, bitInfo.getBucketOffset(),value);
}
public void setCompressedBit(String key, long offset) {
this.setCompressedBit(key, offset, true);
}
public void remCompressedBit(String key, long offset) {
this.setCompressedBit(key, offset, false);
}
public Boolean getCompressedBit(String key, long offset) {
CompressedBitInfo bitInfo = CompressedBitTookit.getBitInfo(offset);
String bitKey = getBitKey(key, bitInfo.getBucketIndex());
log.debug("getCompressedBit with key:{}, offset:{}",bitKey, bitInfo.getBucketOffset());
return redis.getBit(bitKey, bitInfo.getBucketOffset());
}
public void deleteAllCompressedBit(String key, long maxOffset) {
CompressedBitInfo bitInfo = CompressedBitTookit.getBitInfo(maxOffset);
for (int i = 0; i < bitInfo.getBucketIndex(); i++) {
String bitKey = getBitKey(key, i);
redis.expire(bitKey, 0);
}
}
private String getBitKey(String key, long bucketIndex) {
return new StringBuffer(key).append("_").append(bucketIndex).toString();
}
