Redis BitMap 使用场景及实现

痛定思痛。 2022-11-29 11:56 194阅读 0赞

转载自：

[https://codingnote.cc/p/157727][https_codingnote.cc_p_157727]

[https://diuut.com/?p=1055][https_diuut.com_p_1055]

## 叙述 ##

前段时间，在网上看到一道面试题：

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

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

### 原理 ###

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

对于这题，有2个重要的点需要考虑：

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

![20200820142321940.png][]

假设每个用户每天登陆需要占据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亿用户的活跃状态。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpemhlbmd5dTg5MTIzMQ_size_16_color_FFFFFF_t_70][]

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

100000000/8/1024/1024=11.92M

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

![20200820142558478.png][]

一天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 {
     
        @Autowired
        private 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]);
        }
    }

[https_codingnote.cc_p_157727]: https://codingnote.cc/p/157727
[https_diuut.com_p_1055]: https://diuut.com/?p=1055
[20200820142321940.png]: /images/20221124/4787bba48fa943fa8f9601a68bef8f8a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpemhlbmd5dTg5MTIzMQ_size_16_color_FFFFFF_t_70]: /images/20221124/2e14c859dc514d5c9eef624093672937.png
[20200820142558478.png]: /images/20221124/0291ee055a964048a72d6ab5412bcb40.png