转载自：
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]);}
}