Redis实现日榜|直播间榜单|排行榜|Redis实现日榜01

前言

直播间贡献榜是一种常见的直播平台功能，用于展示观众在直播过程中的贡献情况。它可以根据观众的互动行为和贡献值进行排名，并实时更新，以鼓励观众积极参与直播活动。

在直播间贡献榜中，每个观众都有一个对应的贡献值，贡献值用来衡量观众在直播过程中的贡献程度。观众的贡献值可以通过多种途径获得，比如送礼物、打赏主播等。

首先，我们需要创建一个贡献榜单，可以使用Redis的有序集合 （Sorted Set）结构来实现。在有序集合中，每个观众对应一个唯一的ID作为成员，而成员的分数表示观众的贡献值。可以根据观众每次送出礼物增加相应的贡献值。

当有新的观众参与直播并进行互动时，我们可以使用ZADD命令将其用户ID添加到贡献榜单中，并更新相应的贡献值。可以根据贡献值对观众进行排序，从而得到当前排名靠前的观众。

要实时更新贡献榜单，可以使用ZINCRBY命令增加观众的贡献值。当观众进行互动行为时，我们可以调用ZINCRBY命令增加相应观众的贡献值，并确保贡献榜单及时反映观众的最新贡献情况。

Redis实现命令

用户ID为Test1000的得到价值为1314的礼物时，以及获取排行榜时，命令如下。比如

# 增加排行榜用户数据ZINCRBY ROUND_LIST_CACHE_20221222 1314 Test1000​# 展示用户榜单ZRANGE ROUND_LIST_CACHE_20221222 0 -1 WITHSCORES

JAVA简单逻辑代码实现 

1.Spring boot的yml配置文件，配置礼物队列

#yml配置文件配置队列 GiftFlowOutput: content-type: application/json destination: gift_all_flow GiftFlowInput: #礼物队列 content-type: application/json group: GiftAllFlowGroup

2.redis使用lua脚本增加榜单，保证多机并发原子性

//redis lua脚本配置 @Slf4j @Configuration public class RedisConfig { @Autowired private JdkCacheHandler jdkCacheHandler; @Bean("zsetScoreScript") public RedisScript<Long> zsetScoreScript() { DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(); redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/zadd.lua"))); redisScript.setResultType(Long.class); return redisScript; } }

3。LUA脚本具体实现，保留3位有效礼物小数位，后面小数位用于同个时间刷礼物进行排序，目前这里只精确到了秒

local key=KEYS[1] local member=KEYS[2] local newValue=tonumber(string.format("%.16f",ARGV[1])) local oldValue=redis.call('ZSCORE',key,member) if type(oldValue) == 'boolean' then redis.call('ZADD',key,newValue,member) return 1 else redis.call('ZADD',key,tonumber(string.format("%.3f",oldValue))+newValue,member) return 1 end return 0

 4.调用lua脚本，增加排行榜积分

@Component @Slf4j public class RankScoreUtilManager { private final static DecimalFormat format = new DecimalFormat(ActivityBase.TOTAL_FORMAT); @Autowired private StringRedisTemplate redisTemplate; @Autowired private ActivityTimeCache activityTimeCache; @Resource(name = "zsetScoreScript") private RedisScript<Long> zaddScript; /** * 添加分数到排行榜，可以并发的 */ public void addScoreToRank(String cacheKey, String anchorId, BigDecimal integral, Date eventTime) { try { BigDecimal bigDecimal = dealScore(integral, activityTimeCache.getActivityDTO().getEndTime(), eventTime); String score = format.format(bigDecimal.doubleValue()); Long execute = redisTemplate.execute(zaddScript, Arrays.asList(cacheKey, anchorId), score); log.warn("增加积分到排行榜integral={},anchorId={},score={},execute",integral,anchorId,score,execute); } catch (Exception e) { log.error("增加异常", e); } } private static BigDecimal dealScore(BigDecimal newScore, LocalDateTime activityEndTime, Date eventDate) { DecimalFormat format = new DecimalFormat(ActivityBase.VALID_FORMAT); String formatStr = format.format(EeBigDecimalUtil.scale(newScore, ActivityBase.VALID_SCALE, RoundingMode.DOWN).doubleValue()); StringBuilder sb = new StringBuilder(32); //后面补个0，避免lua进1出错 sb.append(formatStr).append('0'); long n = EeDateUtil.getMilli(activityEndTime) - eventDate.getTime(); String s = Long.toString(Math.abs(n) / 1000); for (int i = s.length(); i < ActivityBase.TIME_SCALE; i++) { sb.append('0'); } sb.append(s); return new BigDecimal(sb.toString()).setScale(ActivityBase.TOTAL_SCALE, RoundingMode.DOWN); } }

5.配置礼物队列名称 

/** * 监听礼物流水队列 */ public interface AllGiftFlowProcessor { String OUTPUT = "GiftFlowOutput"; @Output(OUTPUT) MessageChannel output(); String INPUT = "GiftFlowInput"; @Input(INPUT) SubscribableChannel input(); }

 6.监听礼物队列的listener,前面做了一些活动时间校验的判断，最关键的是最下面roundListBusiness.dealAnchorRoundList(dto);的方法

//监听礼物队列，处理相关业务逻辑，榜单的处理在最下面 @Slf4j @Service public class AllGiftFlowListener { @Autowired private RedisTemplate<String, String> redisTemplate; @Autowired private AnchorLevelBusiness anchorLevelBusiness; private static final String cacheKey = "GIFT:TASK:INTER:EVENT:"; @Autowired private EeEnvironmentHolder eeEnvironmentHolder; @Autowired private ActivityRoundDao activityRoundDao; @Autowired private ActivityTimeCache activityTimeCache; @Autowired private GiftConfigCache giftConfigCache; @Autowired private GiftFlowProcessor giftFlowProcessor; @Autowired private AnchorCache anchorCache; @Autowired private RoundListBusiness roundListBusiness; @Autowired private EeLog eeLog; @StreamListener(AllGiftFlowProcessor.INPUT) public void onReceive(ActivityGiftEventDTO dto) { MqConsumeRunner.run(dto.getEventId().toString(), dto, o -> dealMsgEvent(o), "TaskIntegralProcessor [{}]", dto); } private void dealMsgEvent(ActivityGiftEventDTO dto) { // 过滤非活动时间礼物 ActivityDTO activityDTO = activityTimeCache.getActivityDTO(); if (null == activityDTO) { return; } if (EeDateUtil.toLocalDateTime(dto.getEventDate()).isBefore(activityDTO.getStartTime())) { eeLog.info("礼物时间小于活动开始时间，丢弃礼物"); return; } // 判断活动时间 if (ActivityStatusEnum.NO_START == activityRoundDao.getActivityStatus()) { return; } // 过滤活动礼物 if (giftConfigCache.getData().stream().noneMatch(o -> o.getGiftId().equals(dto.getGiftId()))) { eeLog.info("礼物id:{}不计算", dto.getGiftId()); return; } Integer region = anchorCache.getRegionById(dto.getTarget()); // 是否为签区域主播 if (null == region || !ActivityBase.AnchorRegion.contains(region)) { eeLog.warn("该主播非签约或非参赛区域：{}", dto.getTarget()); return; } // 是否重复消费礼物 Boolean success = redisTemplate.opsForValue().setIfAbsent(cacheKey + dto.getEventId(), "", 15, TimeUnit.DAYS); if (success != null && !success) { eeLog.info("升级事件已处理:" + dto); return; } try { //监听礼物并且处理榜单（最主要的代码就这一句） roundListBusiness.dealAnchorRoundList(dto); } catch (Exception e) { log.error("处理榜单 fail.[" + dto + "]", e); } } }

 7.榜单的具体实现逻辑

@Component @Slf4j public class RoundListBusiness { //平台主播榜单 private final static String CHRISTMAS_ROUND_ANCHOR_LIST = "CHRISTMAS:ROUND:ANCHOR:LIST"; private final static String CHRISTMAS_ROUND_LIST_LOCK = "CHRISTMAS:ROUND:LIST:LOCK"; @Autowired private RankScoreUtilManager rankScoreUtilManager; @Autowired private ActivityTimeCache activityTimeCache; @Autowired RedisTemplate<String, String> redisTemplate; @Autowired private AllGiftFlowProcessor allGiftFlowProcessor; /** * 处理榜单加分逻辑 */ public void dealAnchorRoundList(ActivityGiftEventDTO dto) { ActivityDTO activityDTO = activityTimeCache.getActivityDTO(); if (EeDateUtil.toLocalDateTime(dto.getEventDate()).isBefore(activityDTO.getStartTime())) { return; } if (!EeDateUtil.toLocalDateTime(dto.getEventDate()).isBefore(activityDTO.getEndTime())) { return; } //记录总的榜单流水 try { //插入总的流水 allGiftFlowProcessor.output().send(MessageBuilder.withPayload(dto).build()); } catch (Exception e) { log.error("插入总的礼物流水异常dto={}", dto, e); } LocalDateTime now = LocalDateTime.now(); if (!now.isBefore(activityDTO.getEndTime())) { //2.判断是否符合处理上一轮榜单的逻辑 if (isThrowAwayBeforeGift(dto.getEventId(), now, activityDTO.getEndTime())) { log.warn("这里跳出了dto={},now={}", dto, EeDateUtil.format(now)); return; } } dealRoundList(dto, dto.getTotalStarAmount()); } /** * 处理主播榜单加分逻辑 */ private void dealRoundList(ActivityGiftEventDTO dto, BigDecimal value) { //增加平台主播榜单 incrAnchorListValue(CHRISTMAS_ROUND_ANCHOR_LIST, dto.getTarget(), value, dto.getEventDate()); } /** * 具体加分方法 */ public void incrAnchorListValue(String listCacheKey, String userId, BigDecimal value, Date eventTime) { if (EeStringUtil.isNotEmpty(listCacheKey)) { //增加榜单分数 rankScoreUtilManager.addScoreToRank(listCacheKey, userId, value, eventTime); } } /** * 判断是否已经超过结算时间 */ private boolean isThrowAwayBeforeGift(String eventId, LocalDateTime now, LocalDateTime endTime) { //如果当前时间超过了结算时间，直接丢弃礼物 if (!now.isBefore(endTime.plusSeconds(ActivityBase.PROCESS_TS))) { log.error("主播榜单-当前时间超过了结算时间，直接丢弃礼物: {}", eventId); return true; } //如果上一轮的榜单已经锁定，丢弃礼物 if (checkBlockRankList(CHRISTMAS_ROUND_ANCHOR_LIST)) { log.error("主播榜单-榜单被锁定后丢弃礼物: {}, {}", eventId, EeDateUtil.format(LocalDateTime.now())); return true; } return false; } /** * 判断结算时榜单是否已经被锁定 */ public boolean checkBlockRankList(String listCacheKey) { Boolean cache = redisTemplate.opsForHash().hasKey(CHRISTMAS_ROUND_LIST_LOCK, listCacheKey); return null != cache && cache; } /** * 锁定榜单,把锁定的榜单都放入一个hash中 */ public void setBlockRankList(String cacheKey) { redisTemplate.opsForHash().put(CHRISTMAS_ROUND_LIST_LOCK, cacheKey, EeDateUtil.format(LocalDateTime.now())); } }

总结：目前这段代码只是实现了简单的日榜逻辑，还有一段结算的代码我没有复制出来，结算榜单无非就是在每天0点的时候结算前一天的榜单，对榜单前几名的主播进行礼物发放，后续将会更新几种复杂榜单的实现方式，包括：晋级榜单，积分晋级榜单，滚动日榜，滚动周榜，滚动月榜的一些实现方式