30天内资源访问量排名-redis pipeline

待我称王封你为后i 2022-04-05 08:28 75阅读 0赞

## 一、问题描述 ##

实现30天内不同资源的访问量前10名

## 二、问题分析 ##

### 1、存放 ###

将每天的每个资源访问量放入redis,设置有效期为30天，每访问一次就更新一下对应key的value；设计使用Hash,key取值为通常key+日期，并设置有效期为30天，field取资源id,对应value为当日访问量

### 2、读取 ###

#### （1）循环读取30个key ####

循环生成近30天的key,依次读取对应key所有的值放入map,然后相同资源id访问量进行累加

实现简单，但是由于需要连接redis读取30次，会稍微的影响性能

#### （2）使用redis pipeline ####

将近30天的key放入list中，使用管道查出所有值放入map,然后相同资源id访问量进行累加

较（一）进行了优化

## 三、具体实现 ##

### 1、存放 ###

#### 1）redis方法 ####

/** * Hash 操作 * 使用redisTemplate * 也可参考Redis Hincrby 命令 */
            public void hincrBy(String key, String field, Long incr, Long timeout,TimeUnit unit){ 
                key = formatRedisKey(key);
                if (!StringUtils.isEmpty(key) && !StringUtils.isEmpty(field)) { 
                    try { 
                        if(incr == null){ 
                            //此方法会先检查key是否存在，存在+1，不存在先初始化，再+1
                            redisTemplate.opsForHash().increment(key,field,1);
                            redisTemplate.expire(key,timeout, unit);
                        } else { 
                            redisTemplate.opsForHash().increment(key,field,incr);
                            redisTemplate.expire(key,timeout, unit);
                        }
                    } catch (Exception e) { 
                        log.error("RedisUtil:get", e.getMessage());
                    }
                }
            }

#### 2）调用redis方法 ####

/** * 将浏览记录放redis保留30天 */
        	public void saveResViewRecord(String key, Integer resId) { 
                SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");
                String date = df.format(new Date());
                key = key + ":" + date;
                String field = resId + "";
        
                try { 
                    redisUtil.hincrBy(key, field, null, 30L, TimeUnit.DAYS);
                } catch (Exception e) { 
                }
        	}

### 2、读取 ###

#### （1）循环读取 ####

##### 1）redis方法 #####

public Map<Object,Object> hgetAll(String key){ 
                key = formatRedisKey(key);
                return redisTemplate.opsForHash().entries(key);
            }

##### 2)调用redis方法 #####

Map<String, Integer> residMap = new HashMap<>();
        Map<Object, Object> resMap = null;
        try { 
            //循环从redis取前30天记录
            for (int i = 0; i < 30; i++) { 
                key = keyPrefix + DateUtil.formatDate(DateUtil.calcDate(date, Calendar.DATE, -i), "yyyyMMdd");
                //调用redis
                resMap = redisUtil.hgetAll(key);
                //解析返回结果
                for (Map.Entry<Object, Object> entry : resMap.entrySet()) { 
                    if (residMap.containsKey(entry.getKey())) { 
                        residMap.put(entry.getKey().toString(), residMap.get(entry.getKey()) + (entry.getValue() == null ? 0 : (int) entry.getValue()));
                    } else { 
                        residMap.put(entry.getKey().toString(), (entry.getValue() == null ? 0 : (int) entry.getValue()));
                    }
                }
            }
        } catch (Exception e) { 
            return null;
        }

#### (2)redis pipeline ####

##### 1)redis方法 #####

public List<Object> hgetAllPipe(final List<String> keys){ 
                List<Object> result = redisTemplate.executePipelined(new RedisCallback<List<Object>>() { 
                    @Override
                    public List<Object> doInRedis(RedisConnection connection) throws DataAccessException { 
                        for (int i = 0; i < keys.size(); i++) { 
                            byte[] key = redisTemplate.getStringSerializer().serialize(
                                    (formatRedisKey(keys.get(i))));
                            connection.hGetAll(key);
                        }
                        return null;
                    }
                });
                return result;
            }

##### 2)调用redis方法 #####

Map<String, Integer> residMap = new HashMap<>();
        Map<Object, Object> resMap = null;
        //获取近30天的key
        List<String> keys = new ArrayList<>();
        for (int i = 0; i < 1; i++) { 
            key = keyPrefix + DateUtil.formatDate(DateUtil.calcDate(date, Calendar.DATE, -i), "yyyyMMdd");
            keys.add(key);
        }
        try{ 
            //调用redis
            List<Object> objects = redisUtil.hgetAllPipe(keys);
            //解析返回结果
            if(!CollectionUtils.isEmpty(objects)) { 
                objects.forEach(o -> { 
                    Map<Object,Object> map = (Map<Object,Object>)o;
                    if(!CollectionUtils.isEmpty(map)){ 
                        for (Map.Entry<Object, Object> entry : map.entrySet()) { 
                            if (residMap.containsKey(entry.getKey())) { 
                                residMap.put(entry.getKey().toString(), residMap.get(entry.getKey()) + (entry.getValue() == null ? 0 : (int) entry.getValue()));
                            } else { 
                                residMap.put(entry.getKey().toString(), (entry.getValue() == null ? 0 : (int) entry.getValue()));
                            }
                        }
                    }
                });
            }
        } catch (Exception e) { 
            return null;
        }

## 四、分析 ##

doInRedis中的redis操作不会立刻执行，会在connection.closePipeline()之后一并提交到redis并执行，可以提高性能