【技术派后端篇】 Redis 实现用户活跃度排行榜

在各类互联网应用中，排行榜是一个常见的功能需求，它能够直观地展示用户的表现或贡献情况，提升用户的参与感和竞争意识。在技术派项目中，也引入了用户活跃度排行榜，该排行榜主要基于 Redis 的 ZSET 数据结构来实现。接下来，将详细介绍如何实现一个生产可用的用户活跃度排行榜。

1 业务场景说明

在技术派这个博客社区中，设计了用户活跃度排行榜，并区分了日榜和月榜。用户活跃度的计算方式如下：

用户每访问一个新的页面，活跃度加 1 分。
对于一篇文章，用户进行点赞、收藏操作，活跃度加 2 分；若取消点赞、取消收藏，则将之前增加的活跃分收回。
用户对文章进行评论，活跃度加 3 分。
用户发布一篇审核通过的文章，活跃度加 10 分。

排行榜将展示活跃度最高的前三十名用户。

2 方案设计

2.1 存储单元设计

排行榜中每一位用户应持有的信息包括：

userId：用于唯一标识具体的用户。
rank：用户在排行榜上的排名。
score：用户的历史最高积分，即排行榜上的积分。

2.2 数据结构选择

最初考虑使用 LinkedList 来实现排行榜，它的优势在于排名变动时不需要进行数组的拷贝。然而，LinkedList 存在一些缺陷：

用户获取自己排名时效率较低，最差情况需要从头到尾扫描。
当多个用户同时更新 score 时，并发更新排名的问题较为突出。

因此，我们最终选择了 Redis 的 ZSET 数据结构。ZSET 是一个带权重的集合，具有以下特性：

set：集合确保元素的唯一性。
权重：可以将用户的 score 作为权重，每个用户对应一个 score。
zset：根据 score 进行排序的集合。

使用 ZSET 时，每个用户的积分作为带权重的元素存入其中，并且已经按照 score 排好序，通过获取元素对应的 index 即可得到用户的排名。

3 排行榜实现

核心包路径： com.github.paicoding.forum.service.rank
核心代码实现：com.github.paicoding.forum.service.rank.service.impl.UserActivityRankServiceImpl

3.1 更新用户活跃积分

3.1.1 参数传递实体

定义 ActivityScoreBo 实体类，用于传递涵盖业务场景的参数。

@Data
@Accessors(chain = true)
public class ActivityScoreBo {/*** 访问页面增加活跃度*/private String path;/*** 目标文章*/private Long articleId;/*** 评论增加活跃度*/private Boolean rate;/*** 点赞增加活跃度*/private Boolean praise;/*** 收藏增加活跃度*/private Boolean collect;/*** 发布文章增加活跃度*/private Boolean publishArticle;/*** 被关注的用户*/private Long followedUserId;/*** 关注增加活跃度*/private Boolean follow;
}

3.1.2 业务流程

计算活跃度

根据业务实体计算需要增加或减少的活跃度。
增加活跃度时：
- 进行幂等判断，防止重复添加。判断之前是否已经添加过相关的活跃度，若已添加则直接返回；否则执行更新，并保存幂等记录。
减少活跃度时：
- 判断之前是否加过活跃度，防止扣减为负数。若之前没有加过，则直接返回；否则执行扣减，并移除幂等判定。

3.1.3 关键要素

幂等策略

为了防止重复添加活跃度，我们将用户的每个加分项记录下来。在执行具体加分时，基于此来做幂等判定。
将用户的每个加分项记录下来，使用 Redis 的 hash数据结构存储用户的活跃更新操作历史记录，每天一个记录。
- key 为 activity_rank_{user_id}_{年月日}
- field 为活跃度更新 key
- value 为添加的活跃度。

榜单评分更新：基于 ZSET 的 incr 操作更新榜单评分，我们扩展了 RedisClient 工具类，增加了 ZSET 的相关操作。
具体代码路径：com.github.paicoding.forum.core.cache.RedisClient#zIncrBy

    /*** 分数更新** @param key* @param value* @param score* @return*/public static Double zIncrBy(String key, String value, Integer score) {return template.execute(new RedisCallback<Double>() {@Overridepublic Double doInRedis(RedisConnection connection) throws DataAccessException {return connection.zIncrBy(keyBytes(key), score, valBytes(value));}});}

3.1.4 具体实现

代码路径：com.github.paicoding.forum.service.rank.service.impl.UserActivityRankServiceImpl#addActivityScore

/*** 添加活跃分** @param userId        用于更新活跃积分的用户* @param activityScore 触发活跃积分的时间类型*/
@Override
public void addActivityScore(Long userId, ActivityScoreBo activityScore) {if (userId == null) {return;}// 1. 计算活跃度(正为加活跃,负为减活跃)String field;int score = 0;if (activityScore.getPath() != null) {field = "path_" + activityScore.getPath();score = 1;} else if (activityScore.getArticleId() != null) {field = activityScore.getArticleId() + "_";if (activityScore.getPraise() != null) {field += "praise";score = BooleanUtils.isTrue(activityScore.getPraise()) ? 2 : -2;} else if (activityScore.getCollect() != null) {field += "collect";score = BooleanUtils.isTrue(activityScore.getCollect()) ? 2 : -2;} else if (activityScore.getRate() != null) {// 评论回复field += "rate";score = BooleanUtils.isTrue(activityScore.getRate()) ? 3 : -3;} else if (BooleanUtils.isTrue(activityScore.getPublishArticle())) {// 发布文章field += "publish";score += 10;}} else if (activityScore.getFollowedUserId() != null) {// 关注添加积分field = activityScore.getFollowedUserId() + "_follow";score = BooleanUtils.isTrue(activityScore.getFollow()) ? 2 : -2;} else {return;}final String todayRankKey = todayRankKey();final String monthRankKey = monthRankKey();// 2. 幂等：判断之前是否有更新过相关的活跃度信息final String userActionKey = ACTIVITY_SCORE_KEY + userId + DateUtil.format(DateTimeFormatter.ofPattern("yyyyMMdd"), System.currentTimeMillis());Integer ans = RedisClient.hGet(userActionKey, field, Integer.class);if (ans == null) {// 2.1 之前没有加分记录，执行具体的加分if (score > 0) {// 记录加分记录RedisClient.hSet(userActionKey, field, score);// 个人用户的操作记录，保存一个月的有效期，方便用户查询自己最近31天的活跃情况RedisClient.expire(userActionKey, 31 * DateUtil.ONE_DAY_SECONDS);// 更新当天和当月的活跃度排行榜Double newAns = RedisClient.zIncrBy(todayRankKey, String.valueOf(userId), score);RedisClient.zIncrBy(monthRankKey, String.valueOf(userId), score);if (log.isDebugEnabled()) {log.info("活跃度更新加分! key#field = {}#{}, add = {}, newScore = {}", todayRankKey, userId, score, newAns);}if (newAns <= score) {// 由于上面只实现了日/月活跃度的增加，但是没有设置对应的有效期；为了避免持久保存导致redis占用较高；因此这里设定了缓存的有效期// 日活跃榜单，保存31天；月活跃榜单，保存1年// 为什么是 newAns <= score 才设置有效期呢？// 因为 newAns 是用户当天的活跃度，如果发现和需要增加的活跃度 scopre 相等，则表明是今天的首次添加记录，此时设置有效期就比较符合预期了// 但是请注意，下面的实现有两个缺陷：//  1. 对于月的有效期，就变成了本月，每天的首次增加活跃度时，都会重新刷一下它的有效期，这样就和预期中的首次添加缓存时，设置有效期不符//  2. 若先增加活跃度1，再减少活跃度1，然后再加活跃度1，同样会导致重新算了有效期// 严谨一些的写法，应该是 先判断 key 的 ttl， 对于没有设置的才进行设置有效期，如下Long ttl = RedisClient.ttl(todayRankKey);if (!NumUtil.upZero(ttl)) {RedisClient.expire(todayRankKey, 31 * DateUtil.ONE_DAY_SECONDS);}ttl = RedisClient.ttl(monthRankKey);if (!NumUtil.upZero(ttl)) {RedisClient.expire(monthRankKey, 12 * DateUtil.ONE_MONTH_SECONDS);}}}} else if (ans > 0) {// 2.2 之前已经加过分，因此这次减分可以执行if (score < 0) {// 移除用户的活跃执行记录 --> 即移除用来做防重复添加活跃度的幂等键Boolean oldHave = RedisClient.hDel(userActionKey, field);if (BooleanUtils.isTrue(oldHave)) {Double newAns = RedisClient.zIncrBy(todayRankKey, String.valueOf(userId), score);RedisClient.zIncrBy(monthRankKey, String.valueOf(userId), score);if (log.isDebugEnabled()) {log.info("活跃度更新减分! key#field = {}#{}, add = {}, newScore = {}", todayRankKey, userId, score, newAns);}}}}
}

事务与并发问题：当前实现存在事务和并发问题。多次的 Redis 操作存在事务问题，未做并发处理导致幂等无法 100% 生效，可能存在重复添加或扣减活跃度的情况。可通过加锁解决并发问题，通过最终一致性保障事务问题。

3.1.5 触发活跃度更新

借助 Event/Listener 方式处理活跃度更新。监听文章/用户的相关操作事件（如发布文章事件）并更新对应的活跃度，同时在 Filter/Interceptor 层实现基于用户浏览行为的活跃度更新。

通过事件监听机制来触发活跃度更新。例如，用户点赞、评论、发布文章等操作都会触发活跃度的更新。代码路径：com.github.paicoding.forum.service.rank.service.listener.UserActivityListener#notifyMsgListener

/*** 用户操作行为，增加对应的积分** @param msgEvent*/
@EventListener(classes = NotifyMsgEvent.class)
@Async
public void notifyMsgListener(NotifyMsgEvent msgEvent) {switch (msgEvent.getNotifyType()) {case COMMENT:case REPLY:CommentDO comment = (CommentDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setRate(true).setArticleId(comment.getArticleId()));break;case COLLECT:UserFootDO foot = (UserFootDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setCollect(true).setArticleId(foot.getDocumentId()));break;case CANCEL_COLLECT:foot = (UserFootDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setCollect(false).setArticleId(foot.getDocumentId()));break;case PRAISE:foot = (UserFootDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setPraise(true).setArticleId(foot.getDocumentId()));break;case CANCEL_PRAISE:foot = (UserFootDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setPraise(false).setArticleId(foot.getDocumentId()));break;case FOLLOW:UserRelationDO relation = (UserRelationDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setFollow(true).setFollowedUserId(relation.getUserId()));break;case CANCEL_FOLLOW:relation = (UserRelationDO) msgEvent.getContent();userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setFollow(false).setFollowedUserId(relation.getUserId()));break;default:}
}

发布文章事件：com.github.paicoding.forum.service.rank.service.listener.UserActivityListener#publishArticleListener

/*** 发布文章，更新对应的积分** @param event*/
@Async
@EventListener(ArticleMsgEvent.class)
public void publishArticleListener(ArticleMsgEvent<ArticleDO> event) {ArticleEventEnum type = event.getType();if (type == ArticleEventEnum.ONLINE) {userActivityRankService.addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setPublishArticle(true).setArticleId(event.getContent().getId()));}
}

基于用户浏览行为的活跃度更新：com.github.paicoding.forum.web.hook.interceptor.GlobalViewInterceptor#preHandler

@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {if (handler instanceof HandlerMethod) {HandlerMethod handlerMethod = (HandlerMethod) handler;Permission permission = handlerMethod.getMethod().getAnnotation(Permission.class);if (permission == null) {permission = handlerMethod.getBeanType().getAnnotation(Permission.class);}if (permission == null || permission.role() == UserRole.ALL) {if (ReqInfoContext.getReqInfo() != null) {// 用户活跃度更新SpringUtil.getBean(UserActivityRankService.class).addActivityScore(ReqInfoContext.getReqInfo().getUserId(), new ActivityScoreBo().setPath(ReqInfoContext.getReqInfo().getPath()));}return true;}if (ReqInfoContext.getReqInfo() == null || ReqInfoContext.getReqInfo().getUserId() == null) {if (handlerMethod.getMethod().getAnnotation(ResponseBody.class) != null|| handlerMethod.getMethod().getDeclaringClass().getAnnotation(RestController.class) != null) {// 访问需要登录的rest接口response.setContentType(MediaType.APPLICATION_JSON_UTF8_VALUE);response.getWriter().println(JsonUtil.toStr(ResVo.fail(StatusEnum.FORBID_NOTLOGIN)));response.getWriter().flush();return false;} else if (request.getRequestURI().startsWith("/api/admin/") || request.getRequestURI().startsWith("/admin/")) {response.sendRedirect("/admin");} else {// 访问需要登录的页面时，直接跳转到登录界面response.sendRedirect("/");}return false;}if (permission.role() == UserRole.ADMIN && !UserRole.ADMIN.name().equalsIgnoreCase(ReqInfoContext.getReqInfo().getUser().getRole())) {// 设置为无权限response.setStatus(HttpStatus.FORBIDDEN.value());return false;}}return true;
}

通过上述实现，我们能够高效地更新用户活跃积分，并确保幂等性和并发安全。

3.2 排行榜查询

在前面的实现中，我们已经将用户活跃度的数据存储在Redis中，形成了一个完整的排行榜。接下来，我们需要将这个排行榜展示给用户。以下是排行榜查询的基本流程和具体实现。

3.2.1 业务流程

从Redis中获取topN的用户和评分：
- 使用Redis的zRangeWithScores命令获取排名靠前的N个用户及其对应的评分。
查询用户的基本信息：
- 根据获取的用户ID，查询用户的基本信息。
根据用户评分进行排序：
- 将用户信息和评分进行整合，并根据评分进行排序。
更新每个用户的排名：
- 为每个用户设置排名。

在这里插入图片描述

3.2.2 具体实现

代码路径：com.github.paicoding.forum.service.rank.service.impl.UserActivityRankServiceImpl#queryRankList

 @Overridepublic List<RankItemDTO> queryRankList(ActivityRankTimeEnum time, int size) {String rankKey = time == ActivityRankTimeEnum.DAY ? todayRankKey() : monthRankKey();// 1. 获取topN的活跃用户List<ImmutablePair<String, Double>> rankList = RedisClient.zTopNScore(rankKey, size);if (CollectionUtils.isEmpty(rankList)) {return Collections.emptyList();}// 2. 查询用户对应的基本信息// 构建userId -> 活跃评分的map映射，用于补齐用户信息Map<Long, Integer> userScoreMap = rankList.stream().collect(Collectors.toMap(s -> Long.valueOf(s.getLeft()), s -> s.getRight().intValue()));List<SimpleUserInfoDTO> users = userService.batchQuerySimpleUserInfo(userScoreMap.keySet());// 3. 根据评分进行排序List<RankItemDTO> rank = users.stream().map(user -> new RankItemDTO().setUser(user).setScore(userScoreMap.getOrDefault(user.getUserId(), 0))).sorted((o1, o2) -> Integer.compare(o2.getScore(), o1.getScore())).collect(Collectors.toList());// 4. 补齐每个用户的排名IntStream.range(0, rank.size()).forEach(i -> rank.get(i).setRank(i + 1));return rank;}

代码逻辑：
1. 获取topN的活跃用户：
  - 使用RedisClient.zTopNScore方法从Redis中获取排名靠前的N个用户及其评分。
  - 如果返回的列表为空，则直接返回空列表。
2. 查询用户的基本信息：
  - 将获取的用户ID和评分映射到一个Map<Long, Integer>中。
  - 调用userService.batchQuerySimpleUserInfo方法批量查询用户的基本信息。
3. 根据评分进行排序：
  - 将用户信息和评分整合到RankItemDTO对象中。
  - 使用Stream API对RankItemDTO列表进行排序，按评分从高到低排序。
4. 更新每个用户的排名：
  - 使用IntStream.range方法为每个用户设置排名。
    通过上述步骤，我们能够高效地从Redis中获取排行榜数据，并将其展示给用户。

3.2.3 Redis实现

核心的Redis实现如下，直接基于zRangeWithScores获取指定排名的用户和对应分数，其中topN的写法如下：com.github.paicoding.forum.core.cache.RedisClient#zTopNScore

 /*** 找出排名靠前的n个** @param key* @param n* @return*/public static List<ImmutablePair<String, Double>> zTopNScore(String key, int n) {return template.execute(new RedisCallback<List<ImmutablePair<String, Double>>>() {@Overridepublic List<ImmutablePair<String, Double>> doInRedis(RedisConnection connection) throws DataAccessException {Set<RedisZSetCommands.Tuple> set = connection.zRangeWithScores(keyBytes(key), -n, -1);if (set == null) {return Collections.emptyList();}return set.stream().map(tuple -> ImmutablePair.of(toObj(tuple.getValue(), String.class), tuple.getScore())).sorted((o1, o2) -> Double.compare(o2.getRight(), o1.getRight())).collect(Collectors.toList());}});}