再谈如何优雅地使用Redis之位图操作

前言

在之前的文章《如何优雅地使用Redis之位图操作》里为大家介绍了Redis位图操作常见的应用场景，今天继续聊聊Redis位图的其他应用。

首先我们还是从之前的例子入手。在之前的文章中，我们用Redis位图存储了每个用户注册后每天的登录情况，具体的存储方案是以用户id为key，然后以注册天数为offset，bit值为1表示当天登录过，bit值为0表示当天未登录。现在假设我们有另一个需求，就是统计出用户注册后第3天、第5天、第10天、第20天、第30天的登录情况，注意这里要统计的是具体的登录情况，而不是登录的总天数。一种最简单的方案就是通过循环调用getbit命令，查询出每一天的登录状态。由于getbit命令一次只能查询一个offset的bit值，这就意味着，使用这种方式的话，你需要统计多少天的登录情况，就需要调用多少次getbit命令，而每调用一次getbit命令，都需要一次网络请求（因为一般来讲，Redis服务跟应用服务器是不在同一台机器上的），所以当你需要统计的天数比较多时，这种方式的性能是比较差的。

既要实现这个需求，又要兼顾性能，有2个思路可以借鉴。一个思路是使用Redis的管道操作；另一个思路就是《如何优雅地使用Redis之位图操作》这篇文章提到的，通过解析字节数组的方式来获取对应比特位的bit值。

Redis管道操作

先说说什么是Redis的管道操作。Redis官方对管道操作的介绍是：一次请求/响应服务器能实现处理新的请求即使旧的请求还未被响应。这样就可以将多个命令发送到服务器，而不用等待回复，最后在一个步骤中读取该答复。

简而言之，管道操作类似批量操作，可以将多个Redis操作批量发送给Redis，然后一次性地读取操作结果。接下来我们使用一个简单的例子来看看如何用管道操作来实现上述的功能。

   @Test
    public void testPinelined(){
        Jedis client = new Jedis(host, port);
        String key="user_111";
        client.setbit(key,5,true);
        client.setbit(key,20,true);

        long[] offsets=new long[5];
        offsets[0]=3;
        offsets[1]=5;
        offsets[2]=10;
        offsets[3]=20;
        offsets[4]=30;
        Pipeline pipelined = client.pipelined();
        for(long offset:offsets){
            pipelined.getbit(key,offset);
        }
        List<Object> result = pipelined.syncAndReturnAll();
        System.out.println(result);
    }

我们先设置用户注册第5天跟第20天为已登录状态，然后使用管道操作批量读取用户注册后第3天、第5天、第10天、第20天、第30天的登录情况，因此正确的输出应该是除了第5天跟第20天为已登录外，其他都为未登录状态，运行下程序看看结果：

可以看到，确实除了第5天跟第20天为已登录之外，其他天都为未登录。

前面说了使用管道操作的好处就是可以将多个操作批量发送给Redis，然后一次性读取所有命令的结果，因此可以减少网络请求的次数，在命令比较多的情况下可以大大提升性能。然而上述的例子使用的Redis是单节点的，单节点的Redis对管道操作支持比较好，如果是Redis集群，则有些客户端没有提供相关的管道操作，如常用的Jedis客户端就没有提供Redis集群模式下的管道操作。因此如果你使用的是Redis集群，可能无法直接使用管道操作实现上述功能。

基于字节数组解析的getbits

我们还可以使用解析字节数组的方式来一次性获取多个bit值，我将其命名为getbits，顾名思义，就是可以一次性获取多个bit值。最简单的思路就是先获取该key值对应的字节数组，这可以通过get命令来实现。然后再计算出对应的offset在字节数组中的索引，以及在某个字节中的比特位索引，接下来就可以统计出该比特位的bit值了。

我们以offset为30为例，只需要将30除以8（一个字节有8比特），再向下取整，就可以计算出offset为30的比特位在字节数组中的下标了，在这里30除以8向下取整是3，即offset为30的比特位在字节数组中的下标为3（下标是从0开始的）。要计算offset在对应的字节中的比特位下标也很简单，只需要将offset对8取模就行了，比如30对8取模的值为6，说明offset为30的比特位在对应的字节中的比特位下标为6（这里的下标也是从0开始的）。找到了某个offset在字节数组中的下标以及在字节中的比特位下标，就可通过右移的方式计算出该比特位的值了，计算方法在《如何优雅地使用Redis之位图操作》中已经介绍过，不再赘述。接下来看代码：

private static final int BIT_AMOUNT_IN_ONE_BYTE =8;

 public boolean[] getBits(String key, long[] offsets) {
  int offsetLen = offsets.length;
  boolean[] result=new boolean[offsetLen];
  byte[] bytes = this.get(key.getBytes());
  for(int i = 0; i< offsetLen; i++){
   long offset=offsets[i];
   int byteIndexOfTheBytes = getByteIndexInTheBytes(offset);
   int bitIndexOfTheByte = getBitIndexInTheByte(offset);
   byte b = bytes[byteIndexOfTheBytes];
   int shiftCount = getRightShiftStep(bitIndexOfTheByte);
   result[i]=((b>>shiftCount)&1)==1;
  }
  return result;
 }

 private int getByteIndexInTheBytes(long offset){
  return (int) offset/ BIT_AMOUNT_IN_ONE_BYTE;
 }

 private int getBitIndexInTheByte(long offset){
  return (int) offset%BIT_AMOUNT_IN_ONE_BYTE;
 }

 private static int getRightShiftStep(int bitIndexOfTheByte){
  return BIT_AMOUNT_IN_ONE_BYTE -bitIndexOfTheByte-1;
 }

代码就不解释了，思路跟上面讲解的是一致的。

当然这种方式也是存在隐患的。因为我们测试的数据的offset都比较小，就拿我们的例子来说，最大的offset也才到30，因此通过get命令返回的字节数组比较小，没什么大问题。如果我们的offset比较大，比如是百万级别甚至千万级别，这种方式就会有问题了，因为这个时候字节数组会非常大，可能达到几十兆甚至几百兆，这么大的数据通过网络传输需要非常久的时间，也可能造成服务器内存溢出。所以这种方式还有改进的余地，至于如何改进，留给读者去思考，也欢迎在留言区留言。

https://github.com/hzjjames/sedis。（PS：记得给个star哈）