评论分数排序定时任务优化
This commit is contained in:
parent
312c80ca2a
commit
0735f8d0ca
@ -4,6 +4,7 @@ import com.red.circle.common.business.core.ReplaceString;
|
|||||||
import com.red.circle.common.business.core.SensitiveWordFilter;
|
import com.red.circle.common.business.core.SensitiveWordFilter;
|
||||||
import com.red.circle.common.business.enums.AccountStatusEnum;
|
import com.red.circle.common.business.enums.AccountStatusEnum;
|
||||||
import com.red.circle.common.business.enums.DynamicMessageEnum;
|
import com.red.circle.common.business.enums.DynamicMessageEnum;
|
||||||
|
import com.red.circle.component.redis.service.RedisService;
|
||||||
import com.red.circle.framework.core.asserts.ResponseAssert;
|
import com.red.circle.framework.core.asserts.ResponseAssert;
|
||||||
import com.red.circle.other.app.common.room.DynamicCommon;
|
import com.red.circle.other.app.common.room.DynamicCommon;
|
||||||
import com.red.circle.other.app.dto.clientobject.dynamic.DynamicCommentListCO;
|
import com.red.circle.other.app.dto.clientobject.dynamic.DynamicCommentListCO;
|
||||||
@ -59,6 +60,7 @@ public class DynamicCommentCmdExe {
|
|||||||
private final DynamicPopularService dynamicPopularService;
|
private final DynamicPopularService dynamicPopularService;
|
||||||
private final UserAccountCommon userAccountCommon;
|
private final UserAccountCommon userAccountCommon;
|
||||||
private final DynamicPictureService dynamicPictureService;
|
private final DynamicPictureService dynamicPictureService;
|
||||||
|
private final RedisService redisService;
|
||||||
|
|
||||||
public DynamicCommentListCO execute(DynamicCommentCmd cmd) {
|
public DynamicCommentListCO execute(DynamicCommentCmd cmd) {
|
||||||
ResponseAssert.isTrue(DynamicErrorCode.CURRENT_LEVEL_IS_TOO_LOW, commentLimit(cmd));
|
ResponseAssert.isTrue(DynamicErrorCode.CURRENT_LEVEL_IS_TOO_LOW, commentLimit(cmd));
|
||||||
@ -201,9 +203,29 @@ public class DynamicCommentCmdExe {
|
|||||||
comment.setReplyCommentId(cmd.getCommentId());
|
comment.setReplyCommentId(cmd.getCommentId());
|
||||||
dynamicCommentService.save(comment);
|
dynamicCommentService.save(comment);
|
||||||
|
|
||||||
|
// 标记根评论需要更新热度(新增子评论时)
|
||||||
|
if (Objects.nonNull(cmd.getRootCommentId())) {
|
||||||
|
markCommentForHotScoreUpdate(cmd.getRootCommentId());
|
||||||
|
}
|
||||||
|
|
||||||
return comment;
|
return comment;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 标记根评论需要更新热度分数
|
||||||
|
*/
|
||||||
|
private void markCommentForHotScoreUpdate(Long rootCommentId) {
|
||||||
|
if (rootCommentId == null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
redisService.setAdd("comment:hot_score:pending", rootCommentId.toString());
|
||||||
|
} catch (Exception e) {
|
||||||
|
// 标记失败不影响主流程,只记录日志
|
||||||
|
// 定时任务会兜底更新
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private boolean isNotBlank(DynamicCommentCmd cmd) {
|
private boolean isNotBlank(DynamicCommentCmd cmd) {
|
||||||
return StringUtils.isNotBlank(cmd.getContent().replaceAll(" ", ""));
|
return StringUtils.isNotBlank(cmd.getContent().replaceAll(" ", ""));
|
||||||
}
|
}
|
||||||
|
|||||||
@ -41,6 +41,7 @@ public class DynamicCommentLikeCmdExe {
|
|||||||
private final DynamicCommentLikeService dynamicCommentLikeService;
|
private final DynamicCommentLikeService dynamicCommentLikeService;
|
||||||
private final DynamicMessageService dynamicMessageService;
|
private final DynamicMessageService dynamicMessageService;
|
||||||
private final DynamicPictureService dynamicPictureService;
|
private final DynamicPictureService dynamicPictureService;
|
||||||
|
private final com.red.circle.component.redis.service.RedisService redisService;
|
||||||
|
|
||||||
public Long execute(DynamicLikeCmd cmd) {
|
public Long execute(DynamicLikeCmd cmd) {
|
||||||
|
|
||||||
@ -52,6 +53,9 @@ public class DynamicCommentLikeCmdExe {
|
|||||||
ResponseAssert.isFalse(DynamicErrorCode.LIKED, getExist(cmd));
|
ResponseAssert.isFalse(DynamicErrorCode.LIKED, getExist(cmd));
|
||||||
save(cmd, id);
|
save(cmd, id);
|
||||||
|
|
||||||
|
// 标记根评论需要更新热度
|
||||||
|
markCommentForHotScoreUpdate(cmd.getRootCommentId());
|
||||||
|
|
||||||
if (Objects.equals(cmd.getReqUserId(), cmd.getToUserId())) {
|
if (Objects.equals(cmd.getReqUserId(), cmd.getToUserId())) {
|
||||||
return getIncrQuantity(cmd);
|
return getIncrQuantity(cmd);
|
||||||
}
|
}
|
||||||
@ -68,6 +72,9 @@ public class DynamicCommentLikeCmdExe {
|
|||||||
|
|
||||||
deleteCommentLike(cmd);
|
deleteCommentLike(cmd);
|
||||||
|
|
||||||
|
// 标记根评论需要更新热度
|
||||||
|
markCommentForHotScoreUpdate(cmd.getRootCommentId());
|
||||||
|
|
||||||
if (Objects.equals(cmd.getReqUserId(), cmd.getToUserId())) {
|
if (Objects.equals(cmd.getReqUserId(), cmd.getToUserId())) {
|
||||||
return getDecrQuantity(cmd);
|
return getDecrQuantity(cmd);
|
||||||
}
|
}
|
||||||
@ -140,4 +147,19 @@ public class DynamicCommentLikeCmdExe {
|
|||||||
dynamicMessageService.save(message);
|
dynamicMessageService.save(message);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 标记根评论需要更新热度分数
|
||||||
|
*/
|
||||||
|
private void markCommentForHotScoreUpdate(Long rootCommentId) {
|
||||||
|
if (rootCommentId == null) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
redisService.setAdd("comment:hot_score:pending", rootCommentId.toString());
|
||||||
|
} catch (Exception e) {
|
||||||
|
// 标记失败不影响主流程,只记录日志
|
||||||
|
// 定时任务会兜底更新
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -1,28 +1,38 @@
|
|||||||
package com.red.circle.other.app.scheduler;
|
package com.red.circle.other.app.scheduler;
|
||||||
|
|
||||||
import com.red.circle.component.redis.annotation.TaskCacheLock;
|
import com.red.circle.component.redis.annotation.TaskCacheLock;
|
||||||
|
import com.red.circle.component.redis.service.RedisService;
|
||||||
import com.red.circle.framework.mybatis.constant.PageConstant;
|
import com.red.circle.framework.mybatis.constant.PageConstant;
|
||||||
import com.red.circle.other.infra.database.cache.service.other.DynamicCacheService;
|
import com.red.circle.other.infra.database.cache.service.other.DynamicCacheService;
|
||||||
import com.red.circle.other.infra.database.rds.entity.dynamic.DynamicComment;
|
import com.red.circle.other.infra.database.rds.entity.dynamic.DynamicComment;
|
||||||
import com.red.circle.other.infra.database.rds.service.dynamic.DynamicCommentService;
|
import com.red.circle.other.infra.database.rds.service.dynamic.DynamicCommentService;
|
||||||
import com.red.circle.tool.core.collection.CollectionUtils;
|
import com.red.circle.tool.core.collection.CollectionUtils;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.HashSet;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Objects;
|
|
||||||
import java.util.Optional;
|
import java.util.Optional;
|
||||||
|
import java.util.Set;
|
||||||
|
import java.util.stream.Collectors;
|
||||||
|
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.scheduling.annotation.Scheduled;
|
import org.springframework.scheduling.annotation.Scheduled;
|
||||||
import org.springframework.stereotype.Component;
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 评论热度分数更新定时任务.
|
* 评论热度分数更新定时任务(增量更新).
|
||||||
* 每10分钟执行一次,更新最近7天的根评论热度分数
|
* 每10分钟执行一次,优先更新有变化的评论
|
||||||
*/
|
*/
|
||||||
@Slf4j
|
@Slf4j
|
||||||
@Component
|
@Component
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
public class DynamicCommentHotScoreTask {
|
public class DynamicCommentHotScoreTask {
|
||||||
|
|
||||||
|
private static final String PENDING_SET_KEY = "comment:hot_score:pending";
|
||||||
|
|
||||||
|
private final RedisService redisService;
|
||||||
private final DynamicCommentService dynamicCommentService;
|
private final DynamicCommentService dynamicCommentService;
|
||||||
private final DynamicCacheService dynamicCacheService;
|
private final DynamicCacheService dynamicCacheService;
|
||||||
|
|
||||||
@ -36,34 +46,29 @@ public class DynamicCommentHotScoreTask {
|
|||||||
log.info("评论热度分数更新任务 - 开始执行");
|
log.info("评论热度分数更新任务 - 开始执行");
|
||||||
long startTime = System.currentTimeMillis();
|
long startTime = System.currentTimeMillis();
|
||||||
|
|
||||||
// 查询最近7天的根评论(分批处理)
|
int incrementalUpdated = 0;
|
||||||
long sevenDaysAgo = System.currentTimeMillis() - (7 * 24 * 60 * 60 * 1000L);
|
int fallbackUpdated = 0;
|
||||||
int totalUpdated = 0;
|
|
||||||
Long lastId = null;
|
|
||||||
boolean hasMore = true;
|
|
||||||
|
|
||||||
while (hasMore) {
|
// 1. 优先更新有变化的评论(增量更新)
|
||||||
// 分批查询,每次1000条
|
Set<Long> pendingIds = getPendingCommentIds();
|
||||||
List<DynamicComment> rootComments = queryRecentRootComments(sevenDaysAgo, lastId, 1000);
|
if (CollectionUtils.isNotEmpty(pendingIds)) {
|
||||||
|
incrementalUpdated = updatePendingComments(pendingIds);
|
||||||
if (CollectionUtils.isEmpty(rootComments)) {
|
}
|
||||||
hasMore = false;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
// 批量更新热度分数
|
// 2. 兜底更新最近24小时的评论(每小时整点执行)
|
||||||
for (DynamicComment rootComment : rootComments) {
|
int currentMinute = LocalDateTime.now().getMinute();
|
||||||
updateCommentHotScore(rootComment);
|
if (currentMinute == 0) {
|
||||||
totalUpdated++;
|
fallbackUpdated = updateRecentComments(pendingIds);
|
||||||
}
|
|
||||||
|
|
||||||
// 准备下一批
|
|
||||||
lastId = rootComments.get(rootComments.size() - 1).getId();
|
|
||||||
hasMore = rootComments.size() >= 1000;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
long costTime = System.currentTimeMillis() - startTime;
|
long costTime = System.currentTimeMillis() - startTime;
|
||||||
log.info("评论热度分数更新任务 - 执行完成,更新{}条评论,耗时{}ms", totalUpdated, costTime);
|
if (fallbackUpdated > 0) {
|
||||||
|
log.info("评论热度分数更新任务 - 执行完成,增量更新{}条,兜底更新{}条,总计{}条,耗时{}ms",
|
||||||
|
incrementalUpdated, fallbackUpdated, incrementalUpdated + fallbackUpdated, costTime);
|
||||||
|
} else {
|
||||||
|
log.info("评论热度分数更新任务 - 执行完成,增量更新{}条,耗时{}ms",
|
||||||
|
incrementalUpdated, costTime);
|
||||||
|
}
|
||||||
|
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
log.error("评论热度分数更新任务 - 执行失败", e);
|
log.error("评论热度分数更新任务 - 执行失败", e);
|
||||||
@ -71,16 +76,118 @@ public class DynamicCommentHotScoreTask {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 查询最近的根评论(root_comment_id 为 null)
|
* 获取待更新的评论ID集合
|
||||||
*/
|
*/
|
||||||
private List<DynamicComment> queryRecentRootComments(long since, Long lastId, int limit) {
|
private Set<Long> getPendingCommentIds() {
|
||||||
return dynamicCommentService.query()
|
try {
|
||||||
.isNull(DynamicComment::getRootCommentId)
|
Set<String> pendingStrSet = redisService.setGet(PENDING_SET_KEY).stream().map(String::valueOf).collect(Collectors.toSet());
|
||||||
.ge(DynamicComment::getCreateTime, new java.sql.Timestamp(since))
|
if (CollectionUtils.isEmpty(pendingStrSet)) {
|
||||||
.gt(lastId != null, DynamicComment::getId, lastId)
|
return new HashSet<>();
|
||||||
.orderByAsc(DynamicComment::getId)
|
}
|
||||||
.last(PageConstant.formatLimit(limit))
|
|
||||||
.list();
|
Set<Long> pendingIds = new HashSet<>();
|
||||||
|
for (String idStr : pendingStrSet) {
|
||||||
|
try {
|
||||||
|
pendingIds.add(Long.parseLong(idStr));
|
||||||
|
} catch (NumberFormatException e) {
|
||||||
|
log.warn("无效的评论ID: {}", idStr);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return pendingIds;
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("获取待更新评论ID集合失败", e);
|
||||||
|
return new HashSet<>();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 更新待处理的评论
|
||||||
|
*/
|
||||||
|
private int updatePendingComments(Set<Long> pendingIds) {
|
||||||
|
int updated = 0;
|
||||||
|
List<String> successIds = new ArrayList<>();
|
||||||
|
|
||||||
|
for (Long commentId : pendingIds) {
|
||||||
|
try {
|
||||||
|
DynamicComment comment = dynamicCommentService.getById(commentId);
|
||||||
|
if (comment == null) {
|
||||||
|
// 评论已被删除,从待处理集合移除
|
||||||
|
successIds.add(commentId.toString());
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 只更新根评论
|
||||||
|
if (comment.getRootCommentId() == null) {
|
||||||
|
updateCommentHotScore(comment);
|
||||||
|
updated++;
|
||||||
|
}
|
||||||
|
|
||||||
|
// 更新成功,从待处理集合移除
|
||||||
|
successIds.add(commentId.toString());
|
||||||
|
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("更新评论热度失败,commentId={}", commentId, e);
|
||||||
|
// 失败的不移除,下次继续尝试
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 批量移除已处理的ID
|
||||||
|
if (CollectionUtils.isNotEmpty(successIds)) {
|
||||||
|
try {
|
||||||
|
redisService.setRemove(PENDING_SET_KEY, successIds.toArray(new String[0]));
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("移除已处理的评论ID失败", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return updated;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 兜底更新最近24小时的评论(应对时间衰减)
|
||||||
|
*/
|
||||||
|
private int updateRecentComments(Set<Long> excludeIds) {
|
||||||
|
int updated = 0;
|
||||||
|
|
||||||
|
try {
|
||||||
|
long oneDayAgo = System.currentTimeMillis() - (24 * 60 * 60 * 1000L);
|
||||||
|
Long lastId = null;
|
||||||
|
boolean hasMore = true;
|
||||||
|
int batchSize = 500;
|
||||||
|
|
||||||
|
while (hasMore && updated < 2000) { // 限制最多2000条,避免执行时间过长
|
||||||
|
List<DynamicComment> comments = dynamicCommentService.query()
|
||||||
|
.isNull(DynamicComment::getRootCommentId)
|
||||||
|
.ge(DynamicComment::getCreateTime, new java.sql.Timestamp(oneDayAgo))
|
||||||
|
.gt(lastId != null, DynamicComment::getId, lastId)
|
||||||
|
.orderByAsc(DynamicComment::getId)
|
||||||
|
.last(PageConstant.formatLimit(batchSize))
|
||||||
|
.list();
|
||||||
|
|
||||||
|
if (CollectionUtils.isEmpty(comments)) {
|
||||||
|
hasMore = false;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
for (DynamicComment comment : comments) {
|
||||||
|
// 跳过已经在增量更新中处理过的
|
||||||
|
if (excludeIds != null && excludeIds.contains(comment.getId())) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
updateCommentHotScore(comment);
|
||||||
|
updated++;
|
||||||
|
}
|
||||||
|
|
||||||
|
lastId = comments.get(comments.size() - 1).getId();
|
||||||
|
hasMore = comments.size() >= batchSize;
|
||||||
|
}
|
||||||
|
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("兜底更新最近评论失败", e);
|
||||||
|
}
|
||||||
|
|
||||||
|
return updated;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@ -123,7 +230,7 @@ public class DynamicCommentHotScoreTask {
|
|||||||
.execute();
|
.execute();
|
||||||
|
|
||||||
if (totalChildLikes > 0) {
|
if (totalChildLikes > 0) {
|
||||||
log.debug("更新评论热度 - 根评论ID:{}, 根评论点赞:{}, 子评论点赞:{}, 总点赞:{}, 热度分数:{}",
|
log.debug("更新评论热度 - 根评论ID:{}, 根评论点赞:{}, 子评论点赞:{}, 总点赞:{}, 热度分数:{}",
|
||||||
rootComment.getId(), rootLikeCount, totalChildLikes, totalLikeCount, hotScore);
|
rootComment.getId(), rootLikeCount, totalChildLikes, totalLikeCount, hotScore);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -140,7 +247,7 @@ public class DynamicCommentHotScoreTask {
|
|||||||
long currentTime = System.currentTimeMillis();
|
long currentTime = System.currentTimeMillis();
|
||||||
long ageMillis = currentTime - comment.getCreateTime().getTime();
|
long ageMillis = currentTime - comment.getCreateTime().getTime();
|
||||||
double ageHours = Math.max(ageMillis / (1000.0 * 60 * 60), 0.1);
|
double ageHours = Math.max(ageMillis / (1000.0 * 60 * 60), 0.1);
|
||||||
|
|
||||||
return totalLikeCount / Math.pow(ageHours + 2, 1.5);
|
return totalLikeCount / Math.pow(ageHours + 2, 1.5);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user