【Redis实践】zset排行榜

1.背景

排行榜的时效性不高，比如日榜，周榜这种，可以考虑通过定时任务统计、聚合数据并落库，需要查询的时候直接查询这个统计好的数据就好了。但有时候我们遇到的需求时效性会高一点，比如小时榜、分钟榜、甚至实时排行榜，这种情况下再使用定时任务统计的方式就不太合适

2.zset简介

2.1 操作指令

指令	详细指令	说明
zadd	zadd key score member	添加成员和分数，也可以替换成员分数
zincrby	zincrby key score member	为某个成员累加分数，如果成员不存在则创建成员
zrem	zrem key member	删除某个成员
zscore	zscore key member	返回某个成员的分数
zrange	zrange key 0 -1 withscores	按分值从小到大排
zrevrange	zrevrange key 0 -1 withscores	按分值从大到小排

两个range方法，0 -1 是零和负一，中间用空格隔开，意思是获取所有的分数，如果是想获取指定数量的分数，例如top10，这里可以使用 0 9，最后一个withscores的意思的Redis会返回每个成员的分数。在没有这个选项的情况下，ZRANGE只会返回成员的名称，而不包括其对应的分数

2.2 使用示例

• 当前存在已下数据

  user1: 100
  user2: 200
  user3: 150

• 通过指令实现排行榜

  zadd leaderboard 100 user1
  zadd leaderboard 200 user2
  zadd leaderboard 150 user3
  
  zrange leaderboard 0 -1 WITHSCORES
  zrevrange leaderboard 0 -1 WITHSCORES

  3.实战

3.1 RedisTemplate实现排行榜

• 添加或替换用户分数
• 添加或更新用户分数
• 获取排行榜前N名
• 获取某个用户的排名
• 删除指定用户

  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.data.redis.core.RedisTemplate;
  import org.springframework.data.redis.core.ZSetOperations;
  import org.springframework.stereotype.Service;
  
  import java.util.Set;
  
  @Service
  public class LeaderboardService {
  
      @Autowired
      private RedisTemplate<String, String> redisTemplate;
  
      private static final String LEADERBOARD_KEY = "leaderboard";
  
      /**
       * 添加或替换用户分数
       */
      public void addOrReplaceScore(String userId, double score) {
          ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
          zSetOps.add(LEADERBOARD_KEY, userId, score);
      }
  
      /**
       * 添加或更新用户分数
       */
      public void addOrUpdateScore(String userId, double score) {
          ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
          zSetOps.incrementScore(LEADERBOARD_KEY, userId, score);
      }
  
      /**
       * 获取排行榜前N名
       */
      public Set<ZSetOperations.TypedTuple<String>> getTopRanks(int topN) {
          ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
          return zSetOps.reverseRangeWithScores(LEADERBOARD_KEY, 0, topN - 1);
      }
  
      /**
       * 获取用户排名
       */
      public Long getUserRank(String userId) {
          ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
          Long rank = zSetOps.reverseRank(LEADERBOARD_KEY, userId);
          // 排名从1开始
          return rank != null ? rank + 1 : null; 
      }
  
      /**
       * 删除指定用户
       */
      public void removeUser(String userId) {
          redisTemplate.opsForZSet().remove(LEADERBOARD_KEY, userId);
      }
  }

3.2.可能存在的问题及解决方案

一个活动如果参与的人数多，就可能出来成员一直不断膨胀的情况，但实际上我们对排行榜的需求往往只是需要前xx名的数据，例如前10名、前100名、前10000名等等。根据实际的需求，我们可以限制zset中的数量。假如现在保留一万名，就可以提供一个方法，清理排名一万以后的数据

// 限制排行榜最大长度
private static final int MAX_RANKING_SIZE = 10000; 

/**
 * 清理低活跃数据
 */
public void cleanUpInactiveUsers() {
    ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
    Long memberCount = Optional.ofNullable(zSetOps.zCard(LEADERBOARD_KEY)).orElse(0L);
    if (memberCount > MAX_RANKING_SIZE) {
        zSetOps.removeRange(LEADERBOARD_KEY, 0, -MAX_RANKING_SIZE - 1);
    }
}

这个方法可以在插入新的成员时调用，但是由于会多次操作Redis，其实是不建议在保存排行榜分数的时候执行的，可以考虑通过定时任务来处理，

@Component
public class ScheduledTasks {

    @Autowired
    private LeaderboardService leaderboardService;

    // 每天凌晨2点清理
    @Scheduled(cron = "0 0 2 * * ?")
    public void cleanInactiveUsersTask() {
        leaderboardService.cleanUpInactiveUsers();
    }
}

3.2.2.保留当前分数与最高分数

zset中针对同一个用户只能保存一个分数，如果要实现保存当前分数和最高分数，可考虑用两个zset来处理，处理方式也比较简单，按照：获取当前分数、比较分数、更新历史最高分数的顺序做就好了

public void updateScore(String userId, double newScore) {
    // 1. 获取当前分数
    Double currentScore = redisTemplate.opsForZSet().score("currentLeaderboard", userId);

    // 2. 更新当前分数
    redisTemplate.opsForZSet().add("currentLeaderboard", userId, newScore);

    // 3. 更新历史最高分数
    if (currentScore == null || newScore > currentScore) {
        redisTemplate.opsForZSet().add("highestLeaderboard", userId, newScore);
    }
}

3.2.3.批量操作成员分数，减少并发

在并发较高的情况下，如果想减少Redis插入请求，我们可以在内存中先保存一部分的请求，等达到某个阈值的时候，再做Redis的插入操作。这里阈值可以是积累了多少个成员做批量更新，也可以是积累到了一定的时间，例如积累了一分钟的数据。

public class MemberValue<T> implements ZSetOperations.TypedTuple<T> {
    private T value;
    private Double score;

    @Override
    public T getValue() {
        return value;
    }

    public void setValue(T value) {
        this.value = value;
    }

    @Override
    public Double getScore() {
        return score;
    }

    public void setScore(Double score) {
        this.score = score;
    }

    @Override
    public int compareTo(ZSetOperations.TypedTuple<T> o) {
        return 0;
    }
}

private Set<ZSetOperations.TypedTuple<String>> memberSet = new HashSet<>();

@Async
public void asyncBatchSetScore(String userId, double score) {
    MemberValue<String> memberValue = new MemberValue<>();
    memberValue.setScore(score);
    memberValue.setValue(userId);
    synchronized (LeaderboardService.class) {
        memberSet.add(memberValue);
        if (memberSet.size() >= 50) {
            ZSetOperations<String, String> zSetOps = redisTemplate.opsForZSet();
            zSetOps.add(LEADERBOARD_KEY, memberSet);
            memberSet.clear();
        }
    }
}
//如果要修改阈值为时间，可以维护一个时间窗口，并修改判断条件即可