当前位置: 代码迷 >> 综合 >> Redis BitMap 使用场景及实现
  详细解决方案

Redis BitMap 使用场景及实现

热度:75   发布时间:2024-02-12 20:20:37

转载自:

https://codingnote.cc/p/157727

https://diuut.com/?p=1055

叙述

前段时间,在网上看到一道面试题:

如何用redis存储统计1亿用户一年的登陆情况,并快速检索任意时间窗口内的活跃用户数量。

觉得很有意思,就仔细想了下 。并做了一系列实验,自己模拟了下 。还是有点收获的,现整理下来。和大家一起分享。

原理

Redis是一个内存数据库,采用单线程和事件驱动的机制来处理网络请求。实际生产的QPS和TPS单台都能达到3,4W,读写性能非常棒。用来存储一些对核心业务弱影响的用户状态信息还是非常不错的。

对于这题,有2个重要的点需要考虑:

1.如何用合适的数据类型来存储1亿用户的数据,用普通的字符串来存储肯定不行。经过查看一个最简单的kv(key为aaa,value为1)的内存占用,发现为48byte。

假设每个用户每天登陆需要占据1对KV的话,那一亿就是(48*100000000)/1024/1024/1024=4.47G。这还是一天的量。

2.如何满足搜索,redis是一个键值对的内存结构,只能根据key来进行定位value值,无法做到像elastic search那样对文档进行倒排索引快速全文检索。

redis其实有这种数据结构的,可以以很少的空间来存储大量的信息。

解决方案

在redis 2.2.0版本之后,新增了一个位图数据,其实它不是一种数据结构。实际上它就是一个一个字符串结构,只不过value是一个二进制数据,每一位只能是0或者1。redis单独对bitmap提供了一套命令。可以对任意一位进行设置和读取。

因为bitmap的每一位只占据1bit的空间 ,所以利用这个特性我们可以把每一天作为key,value为1亿用户的活跃度状态。假设一个用户一天内只要登录了一次就算活跃。活跃我们就记为1,不活跃我们就记为0。把用户Id作为偏移量(offset)。这样我们一个key就可以存储1亿用户的活跃状态。

我们再来算下,这样一个位图结构的值对象占据多少空间。每一个位是1bit,一亿用户就是一亿bit。8bit=1Byte

100000000/8/1024/1024=11.92M

我用测试工程往一个key里通过lua塞进了1亿个bit,然后用rdb tools对内存进行统计,实测如下:

一天1亿用户也就消耗12M的内存空间。这完全符合要求。1年的话也就4个G。

bitmap可以很好的满足一些需要记录大量而简单信息的场景。所占空间十分小。通常来说使用场景分2类:

1.某一业务对象的横向扩展,key为某一个业务对象的id,比如记录某一个终端的功能开关,1代表开,0代表关。基本可以无限扩展,可以记录2^32个位信息。不过这种用法由于key上带有了业务对象的id,导致了key的存储空间大于了value的存储空间,从空间使用角度上来看有一定的优化空间。

2.某一业务的纵向扩展,key为某一个业务,把每一个业务对象的id作为偏移量记录到位上。这道面试题的例子就是用此法来进行解决。十分巧妙的利用了用户的id作为偏移量来找到相对应的值。当业务对象数量超过2^32时(约等于42亿),还可以分片存储。

看起来bitmap完美的解决了存储和统计的问题。

编码实现

基于springboot框架,使用RedisTemplate进行Redis缓存数据处理:

package com.example.demo.uitl;import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import java.util.concurrent.TimeUnit;/*** @Author Diuut* @Date 2020/6/9  11:18*/
@Component
public class BitMapUtil {@Autowiredprivate RedisTemplate redisTemplate;/*** 设置key字段第offset位bit数值** @param key    字段* @param offset 位置* @param value  数值*/public void setBit(String key, long offset, boolean value) {redisTemplate.execute((RedisCallback) con -> con.setBit(key.getBytes(), offset, value));}/*** 判断该key字段offset位否为1** @param key    字段* @param offset 位置* @return 结果*/public boolean getBit(String key, long offset) {return (boolean) redisTemplate.execute((RedisCallback) con -> con.getBit(key.getBytes(), offset));}/*** 统计key字段value为1的总数** @param key 字段* @return 总数*/public Long bitCount(String key) {return (Long) redisTemplate.execute((RedisCallback<Long>) con -> con.bitCount(key.getBytes()));}/*** 统计key字段value为1的总数,从start开始到end结束** @param key   字段* @param start 起始* @param end   结束* @return 总数*/public Long bitCount(String key, Long start, Long end) {return (Long) redisTemplate.execute((RedisCallback) con -> con.bitCount(key.getBytes(), start, end));}/*** 取多个key并集并计算总数** @param key key* @return 总数*/public Long OpOrCount(String... key) {byte[][] keys = new byte[key.length][];for (int i = 0; i < key.length; i++) {keys[i] = key[i].getBytes();}redisTemplate.execute((RedisCallback) con -> con.bitOp(BitOperation.OR, (key[0] + "To" + key[key.length - 1]).getBytes(), keys));redisTemplate.expire(key[0] + "To" + key[key.length - 1], 10, TimeUnit.SECONDS);return this.bitCount(key[0] + "To" + key[key.length - 1]);}/*** 取多个key的交集并计算总数** @param key key* @return 总数*/public Long OpAndCount(String... key) {byte[][] keys = new byte[key.length][];for (int i = 0; i < key.length; i++) {keys[i] = key[i].getBytes();}redisTemplate.execute((RedisCallback) con -> con.bitOp(BitOperation.AND, (key[0] + "To" + key[key.length - 1]).getBytes(), keys));redisTemplate.expire(key[0] + "To" + key[key.length - 1], 10, TimeUnit.SECONDS);return this.bitCount(key[0] + "To" + key[key.length - 1]);}/*** 取多个key的补集并计算总数** @param key key* @return 总数*/public Long OpXorCount(String... key) {byte[][] keys = new byte[key.length][];for (int i = 0; i < key.length; i++) {keys[i] = key[i].getBytes();}redisTemplate.execute((RedisCallback) con -> con.bitOp(BitOperation.XOR, (key[0] + "To" + key[key.length - 1]).getBytes(), keys));redisTemplate.expire(key[0] + "To" + key[key.length - 1], 10, TimeUnit.SECONDS);return this.bitCount(key[0] + "To" + key[key.length - 1]);}/*** 取多个key的否集并计算总数** @param key key* @return 总数*/public Long OpNotCount(String... key) {byte[][] keys = new byte[key.length][];for (int i = 0; i < key.length; i++) {keys[i] = key[i].getBytes();}redisTemplate.execute((RedisCallback) con -> con.bitOp(BitOperation.NOT, (key[0] + "To" + key[key.length - 1]).getBytes(), keys));redisTemplate.expire(key[0] + "To" + key[key.length - 1], 10, TimeUnit.SECONDS);return this.bitCount(key[0] + "To" + key[key.length - 1]);}
}