Redis实现排行榜

1.为什么要做实时排行榜？

活动排行榜是一种有效的营销策略，可以刺激用户参与度。排行榜本身就是一种竞争性的元素，在这种情况下，人们通常会努力争取竞争优势，以获得更好的排名。同时再加以奖励激励，提升用户粘性。实时的榜单相较于定时刷新的榜单，更能刺激用户的参与欲望，提升用户的活跃度。

2.需求

服务化设计，可提供多业务方接入。
按照用户在活动内获得的总代币数量进行排名，总代币数量越多的排名越靠前。
总代币数量相同的情况，获得时间越早的用户排名靠前。
用户获得代币后榜单实时更新
提供查询整个排行排名列表的接口
提供查询用户个人排名的接口
支持子榜单（月榜周榜等）

3.技术选型

Redis zset (Sorted Set) 实现排行榜实时榜单的数据排名。

Redis zset (Sorted Set)

zset底层使用了两种不同的存储结构，分别是 zipList（压缩列表）和 skipList（跳跃列表），大部分场景下都会使用skipList（跳跃列表）的存储结构，所以简单介绍下跳表的优点。跳跃表是一种基于链表实现的数据结构，它利用了链表的有序性和“跳跃”的特性，以实现快速的查找、插入和删除操作。

跳跃表的数据结构示意图：

跳跃表具有以下优点：

快速查找：跳跃表的查找时间复杂度为O(log n)，与二叉搜索树相当，但跳跃表的实现相对简单，且不需要进行平衡操作。
高效插入和删除：跳跃表的插入和删除时间复杂度为O(log n)，与查找相同，且只需要修改少量指针即可完成操作，不需要像平衡树那样进行大量的旋转操作。
支持有序性：跳跃表中的元素按照顺序排列，可以方便地进行有序性操作，如区间查找和范围删除。
可扩展性：跳跃表的高效性和可扩展性使得它成为一种常用的数据结构，许多数据库和分布式系统中都使用跳跃表实现索引和排序等操作。

zset具备高并发场景下的插入和查询，并且有序的特性使其成为非常适合于实现实时排行榜的数据结构。

4.设计要点

创建排行榜
更新排行榜
查询用户个人排名详情
查询topN排名
缓存重构

为了实现缓存重构的功能，需要记录用户的账户（分数）信息，需要创建2张MySQL表，表结构如下：

分数明细表：

分数汇总表：

4.1 排行榜状态

已开启-当第一条数据写入后，排行榜的状态为已开启
计算中-当redis失效后，需要重建排行榜缓存，重建缓存期间排行榜为计算中的状态

4.2 排行榜的配置元素

榜单的配置元素如下，可以用系统配置进行管理，或者使用管理后台的方式管理，本次使用系统配置的方式实现。

public class RankConfig {

/**

* 排行榜榜单大小

private Integer rankSize;

/**

* 榜单编码

private String rankCode

/**

* 分数类型

private String scoreType;

/**

* 榜单开始时间

private String rankStartTime;

/**

* 榜单截止时间

private String rankEndTime;

/**

* 榜单zset缓存保留时间（天）

private Long rankZsetCacheDay;

}

4.3 接入方的数据上报

通过接口调用的方式上报数据
通过消息队列的方式上报数据

4.4 zset的value和score结构

上面需求有提到相同的分数，分数获取时间早的排名靠前，所以需要把分数的获取时间也计入到score的维度里，value为userId,socre为用户的分数 + 秒级时间戳拼接值，由于double类型在zset内部会有精度问题，所以将score转为整数类型来处理，避免因为精度问题导致的数据异常，score由2部分组成：

橙色部分，score的后7位（不足7位在前面补0），是排行榜的结束时间与用户获取分数的时间的秒差值。
绿色部分，score的除后7位外的部分，是用户的分数。

4.5 更新排行榜

用户获取分数的时间必须要在榜单规定的时间区间内，否则视为无效的数据。

为了防止接入方重复推送数据，规定每一条明细数据必须带有唯一标识，在分数明细表定义唯一索引判断非重复数据后再进行处理。

第一次写榜单数据时，需要写一个redis标记，标识榜单已开始写入数据，用于判断榜单是未开启还是缓存失效需要重构缓存。

为了节省zset占用的内存空间，无需将所有的用户数据都存入zset,只需要在满足2个条件的情况下才将用户的数据更新到zset内。

zset内成员数没有达到排行榜大小

用户代币数大于zset内最小的代币数

写入zset的流程

判断整数位的分数大于zset内的整数位分数或zset内没有数据，取当前的总分数作为score的分数位
计算lastIncomTime和榜单截止时间得到的秒级差值并补齐7位后拼接分数位
更新score

4.6 子榜单的更新

子榜单也视为一个独立的榜单，相较于总榜单，使用的用户分数明细数据是相同的一份数据，只不过是统计的时间维度不同，所以根据分数的类型去找关联的榜单列表时，需要将时间因素加进去，对找出来的所有榜单同时去写排名相关的数据，就可以实现子榜单的玩法。

4.7 查询榜单

在异常情况下会出现redis失效的情况，此时需要对榜单进行缓存重构，缓存重构期间为了友好体验，返回一个排行榜正在计算中的状态。

4.8 榜单缓存重建

由于存在写入场景比较频繁的情况，在扫描分数表的过程中，在用户A的数据已经被扫描并记录到内存后，如果用户A继续获得分数，会导致出现最终结果与用户A最新的分数不一致的情况，所以在缓存重构期间实时榜单队列仍然需要记录最新的数据，最后与扫描分数表的结果队列进行合并，如果有重叠用户则取实时榜单队列内的分数为准。

时序示意图

5.代码实现

5.1 zset的score处理

/**

* 将zset的score的小数部分还原为分数获取的真实时间

* @param rankCode

* @param zsetScore

* @return

public String getScoreTime(String rankCode,double zsetScore){

String[] arr = String.valueOf(zsetScore).split("\\.");

long timeOffSet = Long.valueOf(arr[1]);

long timestamp = getRankStartTimestamp(getRankZsetKey(rankCode)) + timeOffSet;

return SfDateUtil.formatDateTime(SfDateUtil.parseDateTime(timestamp * 1000));

}

/**

* 将zset的score的整数部分还原为真实的分数

* @param zsetScore

* @return

public Long getScore(double zsetScore){

return new Double(Math.floor(zsetScore)).longValue();

}

/**

* 构造zset的score

* @param rankCode

* @param sumScore

* @param scoreTime

* @return

public double genZsetScore(String rankCode,Long sumScore, Date scoreTime){

// 与榜单开启时间的差值得出小数位

long timeOffset = (scoreTime.getTime()/1000) - getRankStartTimestamp(rankCode);

// 分数作为整数位与小数位拼接

String scoreStr = sumScore + "." + timeOffset;

// 转换为double

return Double.valueOf(scoreStr);

}

5.2更新排行榜数据

榜单的数据来源可以有多种形式，我使用消息队列的形式来消费分数数据，定义分数的消息数据类型如下:

public class RankScoreMessageDTO {

/**

* 排行分数类型

private String scoreType;

/**

* 分数变更的时间

private String scoreTime;

/**

* 用户id

private String userId;

/**

* 变更的分数大于0累加小于0扣减

private Long score;

/**

* 分数流水（不允许重复）

private String scoreFlow;

}

更新排行榜的主要逻辑

校验必要参数
判断非重复消费后写分数明细表
根据分数类型找出有哪些榜单引用该种分数类型
给所有榜单批量写入分数汇总表
判断符合写zset后写zset

public void consumeScore(RankScoreMessageDTO rankScoreMessageDTO){

// 判断是否缺少必要字段

if(isRequiredParamMissing(rankScoreMessageDTO)){

log.warn("RankManager comsumeScore fail,参数缺失: {}",rankScoreMessageDTO);

return;

}

try{

// 写入分数明细表根据唯一键约束校验是否重复消费

RankScoreDetailPO rankScoreDetailPO = new RankScoreDetailPO();

BeanUtils.copyProperties(rankScoreMessageDTO,rankScoreDetailPO);

rankScoreDetailMapper.insert(rankScoreDetailPO);

}catch (DuplicateKeyException e){

log.warn("RankManager comsumeScore fail,重复消费: {}",rankScoreMessageDTO);

return;

}

// 查询分数类型对应的榜单编码根据分数类型和时间找出榜单的rankCode

List<String> rankCodes = rankUtils.getRankCodeByScoreType(rankScoreMessageDTO.getScoreType(),rankScoreMessageDTO.getScoreTime());

if(SfCollectionUtil.isNotEmpty(rankCodes)){

// 写入分数汇总表

writeScoreSum(rankCodes,rankScoreMessageDTO);

// 循环处理每个榜单

for(String rankCode: rankCodes){

// 榜单配置

RankConfig rankConfig = rankUtils.getRankConfigByRankCode(rankCode);

// 本次更新后的最新总分

RankScoreSumPO rankScoreSumPO = getUserRankScoreSum(rankCode, rankScoreMessageDTO.getUserId());

// 是否符合入队资格

if(canWriteZset(rankScoreSumPO)){

// 写入zset

writeZset(rankScoreSumPO,rankConfig);

}

写入分数汇总表使用了INSERT ON DUPLICATE KEY UPDATE的操作，第一次写入是插入操作，后续同一个用户在同一个榜单内的数据都是update sum_score字段的操作，并判断是否更新last_incom_time字段。

判断是否能写入zset的实现：

private boolean canWriteZset(RankScoreSumPO rankScoreSumPO){

RankConfig rankConfig = rankUtils.getRankConfigByRankCode(rankScoreSumPO.getRankCode());

if(rankConfig == null){

return false;

}

// 队列是否已满

String rankZsetKey = rankUtils.getRankZsetKey(rankScoreSumPO.getRankCode());

long zsetSize = sfRedisUtil.zSetSize(rankZsetKey);

if(rankConfig.getRankSize() > zsetSize){

return true;

}

// zset按score升序排序集合内第一个值分数最低

String minScoreValue = sfRedisUtil.zSetRange(rankZsetKey,0,0).stream().findFirst().orElse(null);

// 是否大于排行榜分数最小的分数

double minZsetScore = StringUtils.isEmpty(minScoreValue) ? 0 : sfRedisUtil.zSetScore(rankZsetKey,minScoreValue);

// 是否大于排行榜分数最小的分数

if(rankScoreSumPO.getSumScore() > rankUtils.getScore(minZsetScore)){

return true;

}

// 2个条件都不满足不用写入zset

return false;

}

5.3 写入zset的实现

/**

* 将用户分数更新至zset榜单

* @param rankScoreSumPO

private void writeZset(RankScoreSumPO rankScoreSumPO,RankConfig rankConfig){

// zset开始写数据的flag

zsetStartWriteFlag(rankConfig);

// 先更新整数位的分数值

String key = rankUtils.getRankZsetKey(rankConfig.getRankCode());

// 待写入zset的score的整数位下面去拼接小数部分

double zsetScore = rankScoreSumPO.getSumScore();

// zset内的score 与rankScoreSumPO参数做对比决定本次更新的zsetScore

Double lastScore = sfRedisUtil.zSetScore(key,rankScoreSumPO.getRankUser());

// lastScore的时间戳

long lastScoreTimestamp = lastScore == null ? 0 : rankUtils.getScoreTimestamp(rankConfig.getRankCode(), lastScore);

// 本次获取分数的时间戳

long scoreTimestamp = rankScoreSumPO.getLastIncomTime().getTime()/1000;

// 取较大的时间戳

scoreTimestamp = scoreTimestamp > lastScoreTimestamp ? scoreTimestamp : lastScoreTimestamp;

// 构造写入zset内的score

zsetScore = rankUtils.genZsetScore(rankConfig.getRankCode(), rankScoreSumPO.getSumScore(), scoreTimestamp);

// 待写入zset的score大于zset里面的score 或者zset里面的score为空才做写入操作

if(lastScore == null || zsetScore > lastScore){

sfRedisUtil.zSetAdd(key,rankScoreSumPO.getRankUser(), zsetScore);

}

5.4 查询个人排名

使用zset的zrevrank命令，可以直接返回用户的排名数。

/**

* 查询用户个人排名

* @param rankCode 榜单code 排行榜的唯一标识

* @param userId 用户id

* @return

public PersonalRankRespDTO getPersonalRank(String rankCode, String userId){

// 判断榜单状态

int rankStatus = rankUtils.getRankStatus(rankCode);

// 榜单计算中直接返回计算中状态

if(rankStatus == RankStatusEnum.CALCULATING.getStatus()){

return PersonalRankRespDTO.builder().rankStatus(rankStatus).build();

}

// zset的key

String key = rankUtils.getRankZsetKey(rankCode);

// 个人排名

Long redisRankNo = sfRedisUtil.zSetReverseRank(key,userId);

// null值表示用户不在榜单上否则+1（zset的排名从0开始）返回

RankListItemDTO personalRank = new RankListItemDTO();

if(redisRankNo == null){

personalRank.setRankNo(-1L);

} else {

double zSetScore = sfRedisUtil.zSetScore(key,userId);

personalRank.setValue(userId);

personalRank.setRankNo(redisRankNo + 1);

personalRank.setScoreTime(rankUtils.getScoreTime(rankCode,zSetScore));

personalRank.setScore(rankUtils.getScore(zSetScore));

}

return PersonalRankRespDTO.builder().rankStatus(rankStatus).personalRank(personalRank).build();

}

5.5 查询榜单区间排名列表

startNum,endNum为了方便理解，对外从1开始，查询时需要减1。使用zset的zSetReverseRangeByScores命令，直接返回排序好的元素，非常方便。

/**

* 查询排行榜前N名列表

* @param rankCode 榜单code 排行榜的唯一标识

* @param startNum 需要查询的排行榜起始排名区间对外从1开始，查询时需要减1

* @param endNum 需要查询的排行榜截止排名区间

* @return

public RankListRespDTO getRankList(String rankCode, long startNum, long endNum){

// 判断榜单状态

int rankStatus = rankUtils.getRankStatus(rankCode);

// 榜单计算中直接返回计算中状态

if(rankStatus == RankStatusEnum.CALCULATING.getStatus()){

return RankListRespDTO.builder().rankStatus(rankStatus).build();

}

// zset的key

String key = rankUtils.getRankZsetKey(rankCode);

// 按区间范围取数

Set<ZSetOperations.TypedTuple<String>> typedTuples = sfRedisUtil.zSetReverseRangeByScores(key,startNum,endNum);

// 构造返回结果

List<RankListItemDTO> rankList = Lists.newArrayList();

if(SfCollectionUtil.isNotEmpty(typedTuples)){

long rankNo = 1;

for(ZSetOperations.TypedTuple<String> typedTuple: typedTuples){

RankListItemDTO rankListItemDTO = new RankListItemDTO();

rankListItemDTO.setRankNo(rankNo++);

rankListItemDTO.setValue(typedTuple.getValue());

rankListItemDTO.setScore(rankUtils.getScore(typedTuple.getScore()));

rankListItemDTO.setScoreTime(rankUtils.getScoreTime(rankCode,typedTuple.getScore()));

rankList.add(rankListItemDTO);

}

} else {

// 取数出来为空可能是榜单失效判断是否需要重构缓存

if(needRebuildZset(rankCode)){

// 将榜单状态设置为计算中

rankUtils.setRankStatus(RankStatusEnum.CALCULATING);

// 抢到锁的处理重构任务

if(rankUtils.tryGetRebuildZsetLock(rankCode)){

RankConfig rankConfig = rankUtils.getRankConfigByRankCode(rankCode);

// 异步处理重建任务

COMMON_THREAD_POOL_TASK_EXECUTOR.execute(() -> {

rebuildZset(rankConfig);

});

}

// 返回榜单计算中的状态

rankStatus = RankStatusEnum.CALCULATING.getStatus();

}

return RankListRespDTO.builder().rankStatus(rankStatus).rankList(rankList).build();

}

5.6 重构zset

扫描分数总表，取出topN的数据。
将从DB取出来的数据写入zset，与重构期间的实时数据合并。
设置榜单状态为已开启
释放重构zset的分布式锁

private void rebuildZset(RankConfig rankConfig){

try{

log.info("rankCode:{} rebuildZset start! ",rankConfig.getRankCode());

// 任务计时开始

long rebuildZsetStartTime = System.currentTimeMillis();

// 扫描分数汇总表时已排序完毕的列表

List<RankScoreSumPO> scanDBScoreSumList = getRebuildZsetScanDBList(rankConfig);

// 跑重建任务时榜单的zset仍然在实时写数据把扫描出来的数据写入实时榜单的zset做合并

for(RankScoreSumPO rankScoreSumPO: scanDBScoreSumList){

writeZset(rankScoreSumPO,rankConfig);

}

// 将榜单状态设置为已开启

rankUtils.setRankStatus(RankStatusEnum.OPENED);

log.info("rankCode:{} rebuildZset finished! cost:{} mill",rankConfig.getRankCode(),System.currentTimeMillis() - rebuildZsetStartTime);

}catch (Exception e){

log.error("rankCode:{} rebuildZset error!",e);

}finally {

// 释放重构zset的分布式锁

rankUtils.releaseRebuildZsetLock(rankConfig.getRankCode());

}

扫描分数表

遍历128个分片表，每次取topN且分数大于前面分片表得出的topN最小值

/**

* 获取扫描DB排序完成的榜单列表

* @return

private List<RankScoreSumPO> getRebuildZsetScanDBList(RankConfig rankConfig){

List<RankScoreSumPO> rankedScoreSumList = new ArrayList<>();

for (int i = 1; i <= 128; i++) {

// 遍历分片表依次合并topN

try{

log.info("rankCode:{} rebuildZset dataNode:{} start",i);

long dataNodeStartTime = System.currentTimeMillis();

// 分片名

String dataNode = "sstuinfo" + i;

// 每个分片去取topN的时候不用都取N条数据，只用取分数大于已排序好的队列内最小分数的数据

long miniSumScore = SfCollectionUtil.isNotEmpty(rankedScoreSumList)

? rankedScoreSumList.get(rankedScoreSumList.size() - 1).getSumScore() : 0;

// MySQL取数

List<RankScoreSumPO> unRankScoreSumList = rankScoreSumMapper.queryTopScoreSumList(rankConfig.getRankCode(),rankConfig.getRankSize(),miniSumScore,dataNode);

// 和历史已排名的前N进行排名得出新一轮的前N列表

rankedScoreSumList = mergeRankScoreSumList(rankedScoreSumList,unRankScoreSumList,rankConfig.getRankSize());

log.info("rankCode:{} rebuildZset dataNode:{} finished cost:{} mill",i,System.currentTimeMillis() - dataNodeStartTime);

}catch (Exception e){

log.error("rankCode:{} rebuildZset Error, dataNode:{},e:", i, e);

}

return rankedScoreSumList;

}

取数sql

扫描分数表期间的集合排序合并

/**

* 将已排序好的列表和刚从新的分片读出来的数据进行合并，排序后保留N条数据

* @param rankedScoreSumList

* @param unRankScoreSumList

* @param topN

* @return

private List<RankScoreSumPO> mergeRankScoreSumList(List<RankScoreSumPO> rankedScoreSumList,List<RankScoreSumPO> unRankScoreSumList,int topN){

if(SfCollectionUtil.isNotEmpty(unRankScoreSumList)){

// 合并2个集合

rankedScoreSumList.addAll(unRankScoreSumList);

// 按总分和时间排序

rankedScoreSumList = rankedScoreSumList.stream().sorted(Comparator.comparing(RankScoreSumPO::getSumScore,Comparator.reverseOrder()).thenComparing(RankScoreSumPO::getLastIncomTime)).collect(Collectors.toList());

// 保留前N条数据

if(rankedScoreSumList.size() > topN){

rankedScoreSumList = rankedScoreSumList.subList(0,topN);

}

return rankedScoreSumList;

}