springboot 整合redis 实现zset分页查询功能

功能背景:
需要展示用户的访问记录,并且可以用时间筛选
起初是将用户的访问记录存储在mysql里,数据量在100W的时候 经过添加索引,能做到秒查询.但是在表数据到达600W的时候性能就很低下了.
查询的sql语句如下

   select id,create_time,url,`type`,username,use_bill,account_id from proxy_visit
        <where>
            create_time between #{startTime} and #{endTime} and account_id = #{userId}
            <if test="type != '' and type != null">
                and `type` =#{type}
            </if>
        </where>
        order by create_time desc limit #{pageNum},#{limit}

这里使用的是mysql的分页 没有使用pagehelp插件,用插件会有一个count(*)的操作 会慢上加慢
这里用分页查询10页,每页100条的时候 响应的很快 可是越到后面响应的越慢
这里还用到了mysql的limit分页公式

pageNum = (pageNum-1)*limit;

表添加了复合索引 account_id,create_time NORMAL类型 BTREE

后面了解技术选型可能错了,在数据量大的情况下不应该用mysql 因为本身sql调优就不熟练的我 在网上查找mysql调优的博客然后再一个一个去试太烦了
然后就搜索了一下使用redis能不能实现这个功能
挨个看一下Redis所支持的数据类型：

1、String: 主要用于存储字符串，显然不支持分页和排序。
2、Hash: 主要用于存储key-value型数据，
3、List: 主要用于存储一个列表，列表中的每一个元素按元素的插入时的顺序进行保存，如果我们将评论模型按createDate排好序后再插入List中，似乎就能做到排序了，而且再利用List中的LRANGE key start stop指令还能做到分页。嗯，到这里List似乎满足了我们分页和排序的要求，但是评论还会被删除，就需要更新Redis中的数据，如果每次删除评论后都将Redis中的数据全部重新写入一次，显然不够优雅，效率也会大打折扣，如果能删除指定的数据无疑会更好，而List中涉及到删除数据的就只有LPOP和RPOP这两条指令，但LPOP和RPOP只能删除列表头和列表尾的数据，不能删除指定位置的数据，所以List也不太适合(转载的时候看了下，是有 LREM命令可以做到删除，但是LRANGE 似乎是一个耗时命令 O(N) )。
4、Set: 主要存储无序集合，无序！排除。
5、SortedSet: 主要存储有序集合，SortedSet的添加元素指令ZADD key score member [[score,member]…]会给每个添加的元素member绑定一个用于排序的值score，SortedSet就会根据score值的大小对元素进行排序，在这里就可以将createDate当作score用于排序，SortedSet中的指令ZREVRANGE key start stop又可以返回指定区间内的成员，可以用来做分页，SortedSet的指令ZREM key member可以根据key移除指定的成员，能满足删评论的要求，所以，SortedSet在这里是最适合的（时间复杂度O(log(N))）。

这里可以使用sortedset类型来存储用户的访问记录数据 score用来存时间戳,就可以实现根据时间去排序
存储的数据格式如下
为了防止value相同的情况下 存入会发生覆盖,所以在value拼接了时间戳
**> key格式为visit_userId:type 前缀+用户id+类型

value为 url^用户名@流量:13位时间戳
score为时间戳**

存数据就用zadd命令
取数据就用zrevrangebyscore命令
直接上代码
先把这两个redis命令封装一下,一会直接在接口中调用

  /**
     * Redis Zadd 命令用于将一个或多个成员元素及其分数值加入到有序集当中。
     如果某个成员已经是有序集的成员，那么更新这个成员的分数值，并通过重新插入这个成员元素，来保证该成员在正确的位置上。
     分数值可以是整数值或双精度浮点数。
     如果有序集合 key 不存在，则创建一个空的有序集并执行 ZADD 操作。
     当 key 存在但不是有序集类型时，返回一个错误。
     * @param string
     * @param i
     * @param string2
     * @return 被成功添加的新成员的数量，不包括那些被更新的、已经存在的成员。
     */
    public static Long zadd(String key, double score, String member) {
        Jedis jedis = null;
        Long result = null;
        try {
            jedis = jedisPool.getResource();
            result = jedis.zadd(key, score, member);
        } catch (Exception e) {
            LogUtils.error("错误日志：" + e.getMessage());
        } finally {
            jedis.close();
        }
        return result;
    }

    /**
     * Redis Zrevrangebyscore 返回有序集中指定分数区间内的所有的成员。有序集成员按分数值递减(从大到小)的次序排列。
     具有相同分数值的成员按字典序的逆序(reverse lexicographical order )排列。
     除了成员按分数值递减的次序排列这一点外， ZREVRANGEBYSCORE 命令的其他方面和 ZRANGEBYSCORE 命令一样。
     * @param key
     * @param max
     * @param min
     * @param offset
     * @param count
     * @return 指定区间内，带有分数值(可选)的有序集成员的列表。
     */
    public static LinkedHashSet<String> zrevrangebyscore(String key, String max, String min, int offset, int count) {
        Jedis jedis = null;
        LinkedHashSet<String> result = null;
        try {
            jedis = jedisPool.getResource();
            result = (LinkedHashSet<String>) jedis.zrevrangeByScore(key, max, min, offset, count);
        } catch (Exception e) {
            LogUtils.error("错误日志：" + e.getMessage());
        } finally {
            jedis.close();
        }
        return result;
    }

项目中我不需要自己存储数据,代理端会帮我将数据按照我指定的格式存入redis,但是我在测试阶段需要我自己去完成添加数据的操作

随机生成一些数据然后存入redis里 注意key拼接用户名+类型最好加一个前缀,以防到时候业务扩展 ,可能有其他的功能也需要用到相同的userid+type 的key格式

public JSONObject zAdd(Integer count,String username,String userId){
        List<String>urlList = new ArrayList<>();
        urlList.add("www.baidu.com");
        urlList.add("www.google.com");
        urlList.add("www.qq.com");
        urlList.add("www.weixin.com");
        urlList.add("www.sougou.com");
        for (int i = 0; i < count; i++) {
            String url = urlList.get(RandomDataUtils.getRandomData(0, 5));
            Integer type = RandomDataUtils.getRandomData(0, 4);
            long ctime = System.currentTimeMillis();//获取当前时间戳
            String key = "visit_"+userId+":"+type;
            BigDecimal bigDecimal = new BigDecimal(RandomDataUtils.getRandomDataDouble(0, 10));
            BigDecimal usebill = bigDecimal.setScale(2, BigDecimal.ROUND_UP);//随机生成的double小数点后面太长了,转成bigdecimal的时候截取了一下小数点后2位
            String value = url+"^"+username+"@"+usebill+":"+ctime;
            JedisUtil.zadd(key,ctime,value);
        }
    return success1("ok");
    }

经过了数小时的数据生成,redis里终于有600W条数据了.然后开始吧这些数据都取出来
这里LinkedHashSet set = JedisUtil.zrevrangebyscore(key, max, min, pageNum, limit);
max 和min两个参数主要是用于查询时间区间 redis会把score在这个两个值区间的数据查出来
max和min的顺序不能搞错了,max是最大值所以需要是结束时间的时间戳,min是最小值所以需要是开始时间的时间戳
把数据从redis里取出来之后,再用set的迭代器,遍历将value字符串截取从而存入到对象list里返给前端,让前端直接接收对象list进行数据渲染. 这里选择使用arraylist 因为他可以保证排序不被打乱

@GetMapping("/adminGetNewVisit")
    @Role(value = "admin")
    public JSONObject adminGetNewVisit(@NotNull(message = "分页不可为空")Integer pageNum, @NotNull(message = "分页不可为空")Integer limit, String startTime, String endTime, @NotNull(message = "类型不可为空") Integer type,String userId){
        List<VisitVO> list = new ArrayList<>();
        if(StrUtil.isBlank(startTime) || StrUtil.isBlank(endTime)){
            startTime= DateUtil.today()+" 00:00:00";
            endTime= DateUtil.today()+" 23:59:59";
        }
        //将时间转换为时间戳
        String min = DateUtils.dateToStamp(startTime);//开始日期  作为最小值
        String max = DateUtils.dateToStamp(endTime);//结束日期  作为最大值
        pageNum = pageNum * limit -limit;//当前页数*展示多少条-展示多少条=从第几条返回  分页公式
        String key = "visit_"+userId+":"+type;
        LinkedHashSet<String> set = JedisUtil.zrevrangebyscore(key, max, min, pageNum, limit);
        Iterator<String> iterator = set.iterator();
        //遍历set 将时间戳转换为日期
        while (iterator.hasNext()){
            String value = iterator.next();
            String url = org.apache.commons.lang3.StringUtils.substringBeforeLast(value, "^");//截取最前面的url
            String ctime = org.apache.commons.lang3.StringUtils.substringAfterLast(value, ":");//截取最后面的时间戳
            String isBill = org.apache.commons.lang3.StringUtils.substringAfterLast(value, "@");//截取@符号后面的内容
            String useBill = org.apache.commons.lang3.StringUtils.substringBeforeLast(isBill, ":");//截取后的 流量:时间戳   截取:前面的流量
            String isUsername= org.apache.commons.lang3.StringUtils.substringBeforeLast(value, "@");//截取@符号前面面的内容
            String username = StringUtils.substringAfterLast(isUsername, "^");//截取后的 url^username  截取^后面的url
            Date date = DateUtil.date(Long.parseLong(ctime));
            String time = DateUtil.format(date, "yyyy-MM-dd HH:mm:ss");
            VisitVO visitVO = new VisitVO();
            visitVO.setUrl(url);
            visitVO.setUsername(username);
            visitVO.setUseBill(useBill);
            visitVO.setCreatedTime(time);
            list.add(visitVO);
        }
        return success1(list);
    }

JedisUtil.zrevrangebyscore(key, max, min, pageNum, limit);
这里的分页参数 pageNum和limit 需要redis能接收的
经过实验得出分页公式
用户输入的页数*显示条数-显示条数 =redis的pageNum

再次查询数据,发现不管是分多少页 都能做到毫秒响应,redis真是的太赞了!!!