业务场景中如何巧妙的应用bitmap和zset

1. bitmap和zset数据结构简介

Redis中的bitmap和zset两种数据结构在业务场景中很是有用,巧妙的使用它们每每能将复杂问题完美解决。咱们先来简单介绍一下这两种数据结构。

信息在计算机上存储的基本单位是位。位只能存储0或者1,咱们平时所说的字符串、数字等全部的数据信息在计算机中都是经过多个0和1组合一块儿表示。8个位为1个字节(1B),1024个字节为1KB,1024KB是1M;也就是说1M就有8388608个位,试想咱们用每一个位上的0和1来表示1条信息，那么仅仅1M空间就能存储很是丰富的内容。

编程语言都提供了一套标准的“位”运算操做符来操做计算机底层的位，位运算的执行效率远远高于加、减、乘、除、取模等运算指令，下面是C语言提供的六种运算符:

位运算符	说明
&	按位与
|	按位或
^	按位异或
～	取反
<<	左移
>>	右移

Redis经过bitmap也提供了一系列的“位”操做指令，咱们来简单看几个经常使用的命令:

• SETBIT ：对 key 所储存的字符串值，设置指定偏移量上的位(bit)为0和1。当 key 不存在时，自动生成一个新的字符串值。字符串会进行伸展(grown)以确保它能够将 value 保存在指定的偏移量上。当字符串值进行伸展时，空白位置以 0 填充。offset 参数必须大于或等于 0 ，小于 2^32 (bit 映射被限制在 512 MB 以内)。

语法：SETBIT key offset value,对使用大的 offset 的 SETBIT 操做来讲，内存分配可能形成 Redis 服务器被阻塞。一个好的实践是能够分配一个足够用的连续位空间避免使用SETBIT过程当中频繁的内存分配

• GETBIT ：对 key 所储存的字符串值，获取指定偏移量上的位(bit)。当 offset 比字符串值的长度大，或者 key 不存在时，返回 0 。

GETBIT key offset

• BITCOUNT : 计算给定字符串中，被设置为 1 的比特位的数量。不存在的 key 被当成是空字符串来处理，所以对一个不存在的 key 进行 BITCOUNT 操做，结果为 0 。

语法：BITCOUNT key start

• GET : 获取bitmap的内容，此方法经常用于将bitmap中的内容持久化储存到磁盘数据库中，不过要进行数据结构转换，后边会详细说到。

Redis使用ANSI C语言编写，bitmap中的大部分操做指令也都是基于C对位的基本操做，C语言对二进制位的操做有不少有用的技巧,好比说SETBIT中将指定位设置位0和1,C语言是这样操做:

n | (1 << (m-1));  //从低位到高位，将n的第m位置设置为1
n & ~(1 << (m-1));  //从低位到高位，将n的第m位设置为0
复制代码

咱们之因此要强调从低位到高位，是由于不一样计算机的CPU有不一样的字节序列,也就是大端和小端。

1． Little endian：将低序字节存储在起始地址

2． Big endian：将高序字节存储在起始地址

判断机器使用的是小端模式仍是大端模式有不少方法，下面笔者提供一种(联合体union的存放顺序是全部成员都从低地址开始存放，利用该特性就能够轻松地得到了CPU对内存采用Little-endian仍是Big-endian模式读写。)：

//return 1 : little-endian
//       0 : big-endian
int checkCPUendian()
{
    union {
        unsigned int a;
        unsigned char b; 
    } c;
  
    c.a = 1;
    return (c.b == 1); 
}
复制代码

判断机器使用大端仍是小端很重要，它决定了程序使用GET命令从Redis中获取的bitmap数据后需不须要进行位的从新排序。

咱们再来看看zset，zset称为有序集合。它容许给集合中的每一个元素设置一个score做为权重，这个score值很是重要，笔者研究过Laravel使用redis实现延时任务的源码Lumen框架“异步队列任务”源码剖析，将score值设置为任务要执行的时间戳，一个守护进程经过时间滑动窗口的方式获取任务，而后执行。

zset经常使用的命令有不少，下边列举一些最经常使用的简单指令:

• ZADD key score1 member1 [score2 member2]：向有序集合添加一个或多个成员，或者更新已存在成员的分数
• ZCOUNT key min max：计算在有序集合中指定区间分数的成员数
• ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT]：经过分数返回有序集合指定区间内的成员

2. 海量数据统计神器:bitmap

bitmap使用位上的0和1表示信息，由于位是计算机存储计算的基本单位，使用位相比较于其余数据结构能够极大的节省存储空间，同时能够提供很是快的计算效率。由于0和1只能表示非是即否的信息，使用它仅能在一些数据量较大，又只关心是与否的场景下使用使用，即便这样，它的威力也是巨大的。咱们下边列举几个例子，再来讲一下如何将位信息持久化存储到数据库。

2.1 十万客户端心跳状态记录

咱们这个场景是这样的，假设有10万个客户端设备，服务端要对这10万的客户端设备进行心跳检测，方案是：每一个客户端每隔1s向服务端发送ping信息，服务端在收到ping信息以后，记录下来ping的信息，表示客户端有心跳。

一天有86400秒，设备数量又是10万，若是咱们将设备每秒的心跳信息都存储到Mysql的一条记录中，仅一天的数据量Mysql就承受不了了。考虑到心跳信息就是非是即否的属性，又由于时间戳是连续的，因此咱们可使用bitmap中的86400个位来记录一台设备的心跳信息，假设设备的编号位001，咱们能够在redis中这样初始化:

127.0.0.1:6379> setbit 001 86399 0
复制代码

为何是86399呢？由于key的位下标是从0开始的，初始化一个最大的位是为了不每次记录客户端心跳时频繁的内存分配。

以后的工做就特别简单了，客户端上报数据以后，服务端就以客户端设备编号位key,上报时间戳相对于当天凌晨的偏移量位offset，设置心跳便可(下边是Go语言的一个范例):

currentTime := time.Now()
	startTime := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), 0, 0, 0, 0, currentTime.Location())
	offsetSecond := currentTime.Unix() - startTime.Unix()
	redisConn := cache.NewPool().Get()
	defer redisConn.Close()
	result, _ := redisConn.Do("SETBIT", "001", offsetSecond, 1)
	fmt.Println(result)
复制代码

2.2 百万用户在线状态统计

不少APP和网站都称拥有上百万的活跃用户，服务端经常须要统计用户的在线状态，或者查找一些近期活跃用户(好比近一个月连续在线的用户)。由于数据量巨大，怎样使用更快的速度、更小的空间查询出来咱们想要的结果呢?答案就是bitmap。前边咱们说过1M的空间就有8388608个位，咱们能够将用户的id映射为bitmp的偏移量，也就是说使用不到1M的空间，就能够存储百万用户的在线信息了。这时使用BITCOUNT就能够很是方便获得当前在线用户数量了。

2.3 千万消费者数据去重

咱们在生产/消费模式中，数据去重避免重复消费一直都是一个很是头疼的话题。数据量较少的状况下能够存储全量数据，能够经过创建索引或者hash在每次检索数据时检查一下数据是否已经被消费；当数据量较大时，很显然这种方法就很是不适用了，数据查找和存储的代价是巨大的;咱们能够想办法将每一条消费数据都映射为一个惟一的一个int型id上，能够是毫秒时间戳 * 设备id，也能够是其余的组合，只要保证惟一便可。假设咱们天天从消费队列中取出的数据是1千万，咱们可使用不到10M的存储空间记录消费信息了(相应的位设置为1)，另外咱们在判断信息是否被消费时，位运算的代码执行效率远远优于数据库索引和hash。

2.4 bitmap数据如何同步到Mysql数据库

咱们知道bitmap底层实现默认是操做的一个个位，也就是一种010101......这样的字符串表示，但它又不是字符串(存储字符串所使用的空间比二进制位大的多)。一个好的实践方式是将bitmap存储拆分红程序中的一个个int值表示，而后存储到数据库，编程语言中例如PHP,最大的int值也只能存储63个二进制位而已(PHP7的int值使用zend_long结构体，8个字节表示int值,最高位是符号位)，Go使用Uint64最多也只能每次表示64个二进制位而已。咱们先来看看Go从redis中取出的bitmap是什么样子的吧，咱们设置bitmap中的key是testBit，bit的第1位为1:

127.0.0.1:6379> setbit testBit 0 1
(integer) 0
复制代码

Go语言取出testBit代码

redisConn := cache.NewPool().Get()
defer redisConn.Close()

cfg, err := redisConn.Do("get", "testBit")
if err != nil {
	fmt.Println(err)
}
fmt.Println(cfg)
复制代码

结果为：

[128]
复制代码

Go语言从redis中取出的bitmap后会赋值给本身的内部（uint8的slice）变量，每个slice中的值为一个字节表示(连续8个的位),但是我明明设置的是testBit的第1个位呀，为何取出来的值是128而不是1呢？由于笔者的计算机是小端的，因此在取出来值以后，咱们还须要对每一个uint8表示的值进行位的对称转换(第1位和第8位交换，第7位和第2位交换.....)

程序实现上也很是的简单:

/**
 * 反转二进制位
 */
func swapBit(uint8Slice []uint8) {
	var j uint8
	for k, i := range uint8Slice {
		j ^= (j & (1 << 0)) ^ ((i >> 7 & 1) << 0)
		j ^= (j & (1 << 1)) ^ ((i >> 6 & 1) << 1)
		j ^= (j & (1 << 2)) ^ ((i >> 5 & 1) << 2)
		j ^= (j & (1 << 3)) ^ ((i >> 4 & 1) << 3)
		j ^= (j & (1 << 4)) ^ ((i >> 3 & 1) << 4)
		j ^= (j & (1 << 5)) ^ ((i >> 2 & 1) << 5)
		j ^= (j & (1 << 6)) ^ ((i >> 1 & 1) << 6)
		j ^= (j & (1 << 7)) ^ ((i >> 0 & 1) << 7)
		uint8Slice[k] = j
	}
}
复制代码

咱们再设置testBit的第1位为1，使用swapBit函数处理取出来的值。

127.0.0.1:6379> setbit testBit 1 1
(integer) 0
复制代码

理论上咱们获得的应该是3，事实也是如此:

redisConn := cache.NewPool().Get()
	defer redisConn.Close()

	cfg, err := redisConn.Do("get", "testBit")
	if err != nil {
		fmt.Println(err)
	}
	//强制类型转化(interface{} -> uint8)
	int8Slice := cfg.([]uint8)

	swapBit(int8Slice)
	fmt.Println(int8Slice)
复制代码

咱们前边提到，不管哪一种编程语言的int值最多也不过能存储64个连续的二进制位，咱们不妨换一种思路，将连续的60位存储位一个int64变量中(在时间表示中，1分钟等于60秒，1小时等于60分钟，这样一个int值能够表达更具体的含义)，具体的转换过程比较复杂，也就是每7.5个byte转换成一个int64值(可是只使用60个位)，下面是代码示例，感兴趣的读者能够研究一下:

//将uint8转化为uint64,只使用60位存储：1分钟 => 60秒
func fillInt64(int8Slice []uint8, int64Slice []uint64) {
	int64Len := len(int64Slice)
	for i := 0; i < int64Len; i++ {
		odd := i % 2
		if odd == 0 {
			offset := uint64(float64(i) * 7.5)
			int64Slice[i] = uint64(int8Slice[offset]) + uint64(int8Slice[offset + 1]) << 8 + uint64(int8Slice[offset + 2]) << 16 + uint64(int8Slice[offset + 3]) << 24 + uint64(int8Slice[offset + 4]) << 32 + uint64(int8Slice[offset + 5]) << 40 + uint64(int8Slice[offset + 6]) << 48 + uint64(int8Slice[offset + 7] & 31) << 56
		} else {
			offset := uint64(math.Floor(float64(i) * 7.5))
			int64Slice[i] = uint64(int8Slice[offset] & 230) >> 4 + uint64(int8Slice[offset + 1]) << 4 + uint64(int8Slice[offset + 2]) << 12 + uint64(int8Slice[offset + 3]) << 20 + uint64(int8Slice[offset + 4]) << 28 + uint64(int8Slice[offset + 5]) << 36 + uint64(int8Slice[offset + 6]) << 44 + uint64(int8Slice[offset + 7]) << 52
		}
	}
}
复制代码

这样咱们就能够将bitmap中的值转换成int64值类型的slice了，可使用','分割存储到数据库中，Mysql查询统计的时候只须要根据','分割加载到内部数据int类型中，根据偏移量计算便可。

3. zset: 时间滑动窗口,计划任务的首选

前边咱们说过，将zset中的score设置为时间戳能够造成一个时间滑动窗口，利用这个特性能够在业务逻辑中实现很是丰富的功能，咱们来简单看一下:

3.1 延时任务定时器

说到定时器，开发者每每会想到javascript中的timer，其实服务端也有定时器，用于控制程序在指定的时间执行任务。php中经过pcntl库来实现定时器，咱们来看php官网给出的例子:

<?php
pcntl_signal(SIGALRM, function () {
    echo 'Received an alarm signal !' . PHP_EOL;
}, false);

pcntl_alarm(5);

while (true) {
    pcntl_signal_dispatch();
    sleep(1);
}
复制代码

例子中，设置了一个5s后执行的任务，经过分发信号的机制实现，在生产环境中不多这样来实现定时任务，laravel程序中能够经过以下命令来设置一个任务延时执行:

$job = (new ExampleJob())->delay(Carbon::now()->addMinute(1));
dispatch($job);
复制代码

其原理就是将任务打包成payload，组成一个消息体添加到redis中，将其score设置为任务要执行的时间戳，经过一个守护进程隔必定时间(例如3s扫描一下zset)取出要执行的任务执行。

Go语言对timer的支持比较友好，感兴趣的读者能够研究一下，go1.12版本使用的是4叉堆,到1.13版本使用了64叉堆，能够在上万高并发下保持毫秒级的偏差，在生产环境中有普遍的应用。

3.2 计划任务规划单

兵法云:“兵马未到，粮草先行”，在不少微服务中，服务端每每将大量要下发到服务端的资源和任务先整理好，使用cron脚本或者守护进程，在指定的时间将任务下发下去。这个时候zset结构又派上用场了，能够将任务要执行的时间戳存储为score，这样就造成了一个计划任务规划单，原理和前边讲的延时任务差很少。

3.3 如何实现十万客户端在线状态实时监测

前边咱们讲了如何使用bitmap来存储客户端的心跳信息，如今咱们再来看一下使用zset如何实现客户端状态的实时监控。咱们只关心客户端最后的状态，咱们设置设备的编号为key,设备上报心跳的最后时间戳为score,客户端每次上报心跳信息的时候，咱们都更新设备的score便可。

经过一个守护进程，使用redis的ZRANGEBYSCORE命令取出前3秒内设备的心跳信息，便可判断哪些设备在这段时间没有了心跳(离线)。下面演示了获取1575880725->1575880728有心跳的设备:

func main() {
	redisConn := cache.NewPool().Get()
	defer redisConn.Close()

	result, err := redis.Strings(redisConn.Do("ZRANGEBYSCORE", "client_heart_bit", 1575880725, 1575880728, "WITHSCORES"))
	if err != nil {
		fmt.Println(err)
	}

	fmt.Println(result)
}
复制代码

4.结束语

本文并未涉及太多的代码逻辑和架构设计，而是从业务的角度，讲解了如何在合适的场景使用redis的bitmap和zset来解决问题。同时，笔者延伸了不少拓展内容，也只是抛砖引玉罢了，读者感兴趣能够深刻探索研究，简单总结是以下两点:

• 1.在数据信息较大，表达的信息非是即否的时候，若是能将信息映射为一个int类型的值，能够考虑使用bitmap
• 2.zset结构在服务端实现延时任务、计划任务和简单监控统计业务上很是有用