1.概述

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

2.zset的基本概念说明

2.1.数据结构说明

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

指令	详细指令	说明
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只会返回成员的名称，而不包括其对应的分数。

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

2.2.zset做排行榜的指令

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

user1: 100
user2: 200
user3: 150

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

zadd leaderboard 100 user1
zadd leaderboard 200 user2
zadd leaderboard 150 user3zrange 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 {@Autowiredprivate 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 {@Autowiredprivate 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;@Overridepublic T getValue() {return value;}public void setValue(T value) {this.value = value;}@Overridepublic Double getScore() {return score;}public void setScore(Double score) {this.score = score;}@Overridepublic 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，演示如何实现排行榜，并对一些可能遇见的问题做了思考了解决方案。在开发中，可以选择其中的一些方案来解决实际的问题。