【Redis实践】使用zset实现实时排行榜以及一些优化思考

1.概述

我们在做互联网项目的时候会遇到一些排行版的需求，如果排行榜的时效性不高，比如日榜，周榜这种，可以考虑通过定时任务统计、聚合数据并落库，需要查询的时候直接查询这个统计好的数据就好了。但有时候我们遇到的需求时效性会高一点，比如小时榜、分钟榜、甚至实时排行榜，这种情况下再使用定时任务统计的方式就不太合适了。
在Redis中有个叫zset的数据结构，非常适合用来做排名，它的数据结构中有一个score分数，我们可以直接使用Redis的指令，让里面的数据的按分数的大小进行排序。所以zset往往是我们做高时效性排行榜的解决方案。

2.zset的基本概念说明

2.1.数据结构说明

下面列举zset的操作指令，有一定经验的同学看到这些指令就应该知道大概可以如何使用了。

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

下面可以看看zset的使用方法。

2.2.zset做排行榜的指令

以一个例子来说明，假设现在有3个用户和对应的分数分别如下：

user1: 100
user2: 200
user3: 150

现在就通过Redis的指令，来试一下排行榜功能，依次键入以下指令：

zadd leaderboard 100 user1
zadd leaderboard 200 user2
zadd leaderboard 150 user3

zrange leaderboard 0 -1 WITHSCORES
zrevrange leaderboard 0 -1 WITHSCORES

可以看到的是，返回的结果的是一行member，一行分数的结构，按照分数的高低进行排序的。

3. 项目中的实践

下面通过在通过RedisTemplate来封装一下排行榜的demo，然后会列出一些思考，考虑实际存在的问题及其解决方案。

3.1.RedisTemplate实现排行榜

由于Redis在SpringBoot中的配置不是本章的重点，以下忽略配置。提供了几个简单的方法，分别是：

• 添加或替换用户分数
• 添加或更新用户分数
• 获取排行榜前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);
    }
}

方法封装好了之后，通过controller提供一个用户访问入口就可以了。下面讲一讲可能遇到的问题以及处理方案。

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

3.2.1. 限制成员的数量

一个活动如果参与的人数多，就可能出来成员一直不断膨胀的情况，但实际上我们对排行榜的需求往往只是需要前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();
    }
}

这里的每天凌晨两点，可以根据需要调整为每小时清理一次，每10分钟清理一次等等。

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);
    }
}

同样的，历史最高分数的zset也需要考虑限制成员数量的问题。此外，如果要考虑原子性，可以通过将上述的代码封装到lua脚本中执行。

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

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

RedisTemplate中的add()有一个重载方法，可以传入一个set进行批量操作：


这是一个interface，我们可以先实现一下：

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;
    }
}

然后以每50个成员更新一次为例，代码如下：

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();
        }
    }
}

如果要修改阈值为时间，可以维护一个时间窗口，并修改判断条件即可，这里不展开了。

4.总结

本章先讲解了zset的数据结构以及使用方式，然后通过RedisTemplate做了一个Demo，演示如何实现排行榜，并对一些可能遇见的问题做了思考了解决方案。在开发中，可以选择其中的一些方案来解决实际的问题。