面试杀手锏：Redis源码之BitMap

2022-03-31 09:46 浏览：247

在上文《面试杀手锏：Redis源码之SDS》中我们深入分析了 SDS 的实现，本次介绍的位图(BitMap)就是借助 SDS 实现的。

本文在最后讲解了BitMap对腾讯面试题的解决方案，并基于BitMap实现了仿GitHub提交次数的日历图，希望各位看官看的开心。

1.位图简介

如果我们需要记录某一用户在一年中每天是否有登录我们的系统这一需求该如何完成呢?如果使用KV存储，每个用户需要记录365个，当用户量上亿时，这所需要的存储空间是惊人的。

Redis 为我们提供了位图这一数据结构，每个用户每天的登录记录只占据一位，365天就是365位，仅仅需要46字节就可存储，极大地节约了存储空间。

位图数据结构其实并不是一个全新的玩意，我们可以简单的认为就是个数组，只是里面的内容只能为0或1而已(二进制位数组)。

2.命令实战

Redis提供了SETBIT、GETBIT、BITCOUNT、BITOP四个常用命令用于处理二进制位数组。

SETBIT：为位数组指定偏移量上的二进制位设置值，偏移量从0开始计数，二进制位的值只能为0或1。返回原位置值。
GETBIT：获取指定偏移量上二进制位的值。
BITCOUNT：统计位数组中值为1的二进制位数量。
BITOP：对多个位数组进行按位与、或、异或运算。

127.0.0.1:6379> SETBIT first 0 1    # 0000 0001
(integer) 0
127.0.0.1:6379> SETBIT first 3 1    # 0000 1001
(integer) 0
127.0.0.1:6379> SETBIT first 0 0    # 0000 1000
(integer) 1
127.0.0.1:6379> GETBIT first 0
(integer) 0
127.0.0.1:6379> GETBIT first 3
(integer) 1
127.0.0.1:6379> BITCOUNT first      # 0000 1000
(integer) 1
127.0.0.1:6379> SETBIT first 0 1    # 0000 1001
(integer) 0
127.0.0.1:6379> BITCOUNT first      # 0000 1001
(integer) 2
127.0.0.1:6379> SETBIT first 1 1    # 0000 1011
(integer) 0
127.0.0.1:6379> BITCOUNT first      # 0000 1011
(integer) 3
127.0.0.1:6379> SETBIT x 3 1        
(integer) 0
127.0.0.1:6379> SETBIT x 1 1        
(integer) 0
127.0.0.1:6379> SETBIT x 0 1        # 0000 1011
(integer) 0
127.0.0.1:6379> SETBIT y 2 1        
(integer) 0
127.0.0.1:6379> SETBIT y 1 1        # 0000 0110
(integer) 0
127.0.0.1:6379> SETBIT z 2 1        
(integer) 0
127.0.0.1:6379> SETBIT z 0 1        # 0000 0101
(integer) 0
127.0.0.1:6379> BITOP AND andRes x y z    #0000 0000
(integer) 1
127.0.0.1:6379> BITOP OR orRes x y z      #0000 1111
(integer) 1
127.0.0.1:6379> BITOP XOR x y z           #0000 1000
(integer) 1
# 对给定的位数组进行按位取反
127.0.0.1:6379> SETBIT value 0 1
(integer) 0
127.0.0.1:6379> SETBIT value 3 1            #0000 1001
(integer) 0
127.0.0.1:6379> BITOP NOT notValue value    #1111 0110
(integer) 1

3.BitMap源码分析

3.1 数据结构

如下展示了一个用 SDS 表示的一字节(8位)长的位图：

扩展：Redis 中的每个对象都是有一个 redisObject 结构表示的。

typedef struct redisObject {
 // 类型
 unsigned type:4;
 // 编码
 unsigned encoding:4;
 unsigned lru:REDIS_LRU_BITS; 
 // 引用计数
 int refcount;
 // 执行底层实现的数据结构的指针
 void *ptr;
} robj;

type 的值为 REDIS_STRING表示这是一个字符串对象
sdshdr.len 的值为1表示这个SDS保存了一个1字节大小的位数组
buf数组中的buf[0]实际保存了位数组
buf数组中的buf[1]为自动追加的0字符

为了便于我们观察，buf数组的每个字节都用一行来表示，buf[i]表示这是buf数组的第i个字节，buf[i]之后的8个格子表示这个字节上的8位。

为再次拉齐各位思想，如下展示了另外一个位数组：

位数组由buf[0]、buf[1]和buf[2]三个字节保存，真实数据为1111 0000 1100 0011 1010 0101

3.2 GETBIT

GETBIT用于返回位数组在偏移量上的二进制位的值。值得我们注意的是，GETBIT的时间复杂度是O(1)。

GETBIT命令的执行过程如下：

GETBIT命令源码如下所示：

void getbitCommand(client *c) {
    robj *o;
    char llbuf[32];
    uint64_t bitoffset;
    size_t byte, bit;
    size_t bitval = 0;
    // 获取offset
    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)
        return;
    // 查找对应的位图对象
    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,o,OBJ_STRING)) return;
  // 计算offset位于位数组的哪一行
    byte = bitoffset >> 3;
    // 计算offset在一行中的第几位，等同于取模
    bit = 7 - (bitoffset & 0x7);
    // #define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)
    if (sdsEncodedObject(o)) {
        // SDS 是RAW 或者 EMBSTR类型
        if (byte < sdslen(o->ptr))
            // 获取指定位置的值
            // 注意它不是真正的一个二维数组不能用((uint8_t*)o->ptr)[byte][bit]去获取呀~
            bitval = ((uint8_t*)o->ptr)[byte] & (1 << bit);
    } else {
        //  SDS 是 REDIS_ENCODING_INT 类型的整数，先转为String
        if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))
            bitval = llbuf[byte] & (1 << bit);
    }
    addReply(c, bitval ? shared.cone : shared.czero);
}

举个栗子

以GETBIT array 3为例，array表示上图中三个字节的位数组。

因为 GETBIT 命令执行的所有操作都可以在常数时间内完成，所以该命令的算法复杂度为O(1)。

3.3 SETBIT

SETBIT用于将位数组在偏移量的二进制位的值设为value，并向客户端返回旧值。

SITBIT命令的执行过程如下：

因为SETBIT命令执行的所有操作都可以在常数时间内完成，所以该命令的算法复杂度为O(1)。

SETBIT命令源码如下所示：

void setbitCommand(client *c) {
    robj *o;
    char *err = "bit is not an integer or out of range";
    uint64_t bitoffset;
    ssize_t byte, bit;
    int byteval, bitval;
    long on;
    // 获取offset
    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)
        return;
    // 获取我们需要设置的值
    if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != C_OK)
        return;
    
    if (on & ~1) {
        // 设置了0和1之外的值，直接报错
        addReplyError(c,err);
        return;
    }
    // 根据key查询SDS对象（会自动扩容）
    if ((o = lookupStringForBitCommand(c,bitoffset)) == NULL) return;
    
    byte = bitoffset >> 3;
    byteval = ((uint8_t*)o->ptr)[byte];
    bit = 7 - (bitoffset & 0x7);
    bitval = byteval & (1 << bit);
    
    byteval &= ~(1 << bit);
    byteval |= ((on & 0x1) << bit);
    ((uint8_t*)o->ptr)[byte] = byteval;
    // 发送数据修改通知
    signalModifiedKey(c,c->db,c->argv[1]);
    notifyKeyspaceEvent(NOTIFY_STRING,"setbit",c->argv[1],c->db->id);
    server.dirty++;
    addReply(c, bitval ? shared.cone : shared.czero);
}