Redis使用及源码剖析-18.Redis二进制位数组、慢查询日志和监视器实现-2021-2-3

---

前言

本文对Redis的二进制位数组，慢查询日志和监视器作以简要介绍。

---

一、二进制位数组

1.二进制位数组命令

Redis提供了setbit getbit bitcount和bitop四个命令用于处理二进制位数组，如下所示：

//设置某一位的值
redis>setbit bit 0 1  //0000 0001 
redis>setbit bit 3 1  //0000 1001 
//获取某一位的值
redis>getbit bit 0 
1
redis>getbit bit 1 
0
//统计值为1的位数目
redis>bitcount bit 
2
//bitop可以让多个位数组进行按位与、或、异或等运算
redis>setbit bit2 2 1  //0000 0100 
redis>bitop or or-result bit bit2 //0000 1101 
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.位数组表示

redis使用SDS字符串表示位数组，并使用SDS操作函数来处理位数组，一个1字节长的位数组示意图如下：

需要注意的是，buf数组保存的顺序和日常书写顺序是相反的，比如上图表示的位数组是 0100 1101。之所以这样做是因为一旦需要更长字节的位数组时，SDS会进行动态扩充，如buf长度从1字节变为2字节，此时高位的数字可以直接保存在新扩充的字节里。否则若按照书写顺序保存，则需要先把低位数据迁移到新扩充字节，浪费时间。按照书写顺序扩充的示意图如下：

3.setbit getbit命令实现

setbit和gitbit命令的执行都是类似的，首先要找到对应二进制位，然后设置或者获取值。找到对应二进制位的方法如下：
a.计算 byte = offset/8向下取整 ， byte 值记录了 offset 偏移量指定的二进制位保存在位数组的哪个字节。
b.计算 bit = (offset mod 8) + 1 ，bit 值记录了 offset 偏移量指定的二进制位是 byte 字节的第几个二进制位。
c.根据 byte 值和 bit 值，在位数组找到 offset 偏移量指定的二进制位。
查找offset=10的位的示意图如下：

3.bitcount命令实现

bitcount用于统计二进制位数组中1的位数，若采用遍历方式统计，在位数组特别大时则耗时很长，redis采用的是查表算法和variable-precision SWAR算法，算法比较复杂，在此就不多做介绍了。

4.源码

二进制位数组相关代码位于bitops.c中，setbit和getbit命令实现函数如下所示：

/* SETBIT key offset bitvalue */
void setbitCommand(redisClient *c) {
    robj *o;
    char *err = "bit is not an integer or out of range";
    size_t bitoffset;
    int byte, bit;
    int byteval, bitval;
    long on;

    // 获取 offset 参数
    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
        return;

    // 获取 value 参数
    if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != REDIS_OK)
        return;

    /* Bits can only be set or cleared... */
    // value 参数的值只能是 0 或者 1 ，否则返回错误
    if (on & ~1) {
        addReplyError(c,err);
        return;
    }

    // 查找字符串对象
    o = lookupKeyWrite(c->db,c->argv[1]);
    if (o == NULL) {

        // 对象不存在，创建一个空字符串对象
        o = createObject(REDIS_STRING,sdsempty());

        // 并添加到数据库
        dbAdd(c->db,c->argv[1],o);

    } else {

        // 对象存在，检查类型是否字符串
        if (checkType(c,o,REDIS_STRING)) return;

        o = dbUnshareStringValue(c->db,c->argv[1],o);
    }

    /* Grow sds value to the right length if necessary */
    // 计算容纳 offset 参数所指定的偏移量所需的字节数
    // 如果 o 对象的字节不够长的话，就扩展它
    // 长度的计算公式是 bitoffset >> 3 + 1
    // 比如 30 >> 3 + 1 = 4 ，也即是为了设置 offset 30 ，
    // 我们需要创建一个 4 字节（32 位长的 SDS）
    byte = bitoffset >> 3;
    o->ptr = sdsgrowzero(o->ptr,byte+1);

    /* Get current values */
    // 将指针定位到要设置的位所在的字节上
    byteval = ((uint8_t*)o->ptr)[byte];
    // 定位到要设置的位上面
    bit = 7 - (bitoffset & 0x7);
    // 记录位现在的值
    bitval = byteval & (1 << bit);

    /* Update byte with new bit value and return original value */
    // 更新字节中的位，设置它的值为 on 参数的值
    byteval &= ~(1 << bit);
    byteval |= ((on & 0x1) << bit);
    ((uint8_t*)o->ptr)[byte] = byteval;

    // 发送数据库修改通知
    signalModifiedKey(c->db,c->argv[1]);
    notifyKeyspaceEvent(REDIS_NOTIFY_STRING,"setbit",c->argv[1],c->db->id);
    server.dirty++;

    // 向客户端返回位原来的值
    addReply(c, bitval ? shared.cone : shared.czero);
}

/* GETBIT key offset */
void getbitCommand(redisClient *c) {
    robj *o;
    char llbuf[32];
    size_t bitoffset;
    size_t byte, bit;
    size_t bitval = 0;

    // 读取 offset 参数
    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
        return;

    // 查找对象，并进行类型检查
    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,o,REDIS_STRING)) return;

    // 计算出 offset 所指定的位所在的字节
    byte = bitoffset >> 3;
    // 计算出位所在的位置
    bit = 7 - (bitoffset & 0x7);

    // 取出位
    if (sdsEncodedObject(o)) {
        // 字符串编码，直接取值
        if (byte < sdslen(o->ptr))
            bitval = ((uint8_t*)o->ptr)[byte] & (1 << bit);
    } else {
        // 整数编码，先转换成字符串，再取值
        if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))
            bitval = llbuf[byte] & (1 << bit);
    }

    // 返回位
    addReply(c, bitval ? shared.cone : shared.czero);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

二、慢查询日志

1.慢查询日志简介

Redis 的慢查询日志功能用于记录执行时间超过给定时长的命令请求， 用户可以通过这个功能产生的日志来监视和优化查询速度。服务器配置有两个和慢查询日志相关的选项：
a.slowlog-log-slower-than 选项指定执行时间超过多少微秒（1 秒等于 1,000,000 微秒）的命令请求会被记录到日志上。
b.slowlog-max-len 选项指定服务器最多保存多少条慢查询日志。服务器使用先进先出的方式保存多条慢查询日志： 当服务器储存的慢查询日志数量等于 slowlog-max-len 选项的值时， 服务器在添加一条新的慢查询日志之前， 会先将最旧的一条慢查询日志删除。

2.慢查询日志保存

服务器状态RedisServer中包含了几个和慢查询日志功能有关的属性，如下所示：

struct redisServer {

    // ...

    // 下一条慢查询日志的 ID
    long long slowlog_entry_id;

    // 保存了所有慢查询日志的链表
    list *slowlog;

    // 服务器配置 slowlog-log-slower-than 选项的值
    long long slowlog_log_slower_than;

    // 服务器配置 slowlog-max-len 选项的值
    unsigned long slowlog_max_len;

    // ...

};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

slowlog_entry_id 属性的初始值为 0 ， 每当创建一条新的慢查询日志时， 这个属性的值就会用作新日志的 id 值， 之后程序会对这个属性的值增一。slowlog 链表保存了服务器中的所有慢查询日志， 链表中的每个节点都保存了一个 slowlogEntry 结构， 每个 slowlogEntry 结构代表一条慢查询日志：

typedef struct slowlogEntry {

    // 唯一标识符
    long long id;

    // 命令执行时的时间，格式为 UNIX 时间戳
    time_t time;

    // 执行命令消耗的时间，以微秒为单位
    long long duration;

    // 命令与命令参数
    robj **argv;

    // 命令与命令参数的数量
    int argc;

} slowlogEntry
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

showlog链表和slowlogEntry示意图如下所示：

3.慢查询日志添加

在每次执行命令的之前和之后，程序都会记录微秒格式的当前 UNIX 时间戳， 这两个时间戳之间的差就是服务器执行命令所耗费时长，服务器会将这个时长作为参数之一传给 slowlogPushEntryIfNeeded 函数，而 slowlogPushEntryIfNeeded 函数则负责检查是否需要为这次执行的命令创建慢查询日志，函数具体作用如下：
a.检查命令的执行时长是否超过 slowlog-log-slower-than 选项所设置的时间， 如果是的话， 就为命令创建一个新的日志， 并将新日志添加到 slowlog 链表的表头
b.检查慢查询日志的长度是否超过 slowlog-max-len 选项所设置的长度， 如果是的话， 那么将多出来的日志从 slowlog 链表中删除掉。

三、监视器

1.监视器简介

通过执行 MONITOR 命令， 客户端可以将自己变为一个监视器， 实时地接收并打印出服务器当前处理的命令请求的相关信息：

redis> MONITOR
OK
1378822099.421623 [0 127.0.0.1:56604] "PING"
1378822105.089572 [0 127.0.0.1:56604] "SET" "msg" "hello world"
1378822109.036925 [0 127.0.0.1:56604] "SET" "number" "123"
1378822140.649496 [0 127.0.0.1:56604] "SADD" "fruits" "Apple" "Banana" "Cherry"
1378822154.117160 [0 127.0.0.1:56604] "EXPIRE" "msg" "10086"
1378822257.329412 [0 127.0.0.1:56604] "KEYS" "*"
1378822258.690131 [0 127.0.0.1:56604] "DBSIZE"
1
2
3
4
5
6
7
8
9

每当一个客户端向服务器发送一条命令请求时， 服务器除了会处理这条命令请求之外， 还会将关于这条命令请求的信息发送给所有监视器， 如下图所示：

2.称为监视器

发送 MONITOR 命令可以让一个普通客户端变为一个监视器， 该命令的实现原理可以用以下伪代码来实现：

def MONITOR():

    # 打开客户端的监视器标志
    client.flags |= REDIS_MONITOR

    # 将客户端添加到服务器状态的 monitors 链表的末尾
    server.monitors.append(client)

    # 向客户端返回 OK
    send_reply("OK")
1
2
3
4
5
6
7
8
9
10

可以看到，客户端要成为监视器时会打开REDIS_MONITOR标志，并且将客户端对象添加到服务端链表的末尾。服务端监视器链表定义如下：

struct redisServer {
    // 链表，保存了所有监视器
    list *monitors;    /* List of slaves and MONITORs */
};
1
2
3
4

3.向监视器发送消息

服务器在每次处理命令请求之前， 都会调用 replicationFeedMonitors 函数， 由这个函数将被处理命令请求的相关信息发送给各个监视器。示意图如下所示：

---

总结

本文对Redis的二进制位数组、慢查询日志和监视器做了简要介绍，如有不当，请多多指正。