From 147992ffdfb3205515fc8ae2fa7029b978ae47a2 Mon Sep 17 00:00:00 2001 From: tianfeng <769204422@qq.com> Date: Thu, 5 Mar 2026 10:44:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=9C=A8=E4=BB=96=E4=BA=BA=E6=88=BF=E9=97=B4?= =?UTF-8?q?=E6=B4=BB=E8=B7=83=E9=80=BB=E8=BE=91=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/listener/UserHeartbeatListener.java | 53 ++++++++++++++++--- .../infra/util/DateTimeAsiaRiyadhUtils.java | 19 +++++++ 2 files changed, 66 insertions(+), 6 deletions(-) create mode 100644 rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/util/DateTimeAsiaRiyadhUtils.java diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java index 609e83c4..6a3ff67d 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/listener/UserHeartbeatListener.java @@ -6,6 +6,7 @@ import com.red.circle.component.mq.config.RocketMqMessageListener; import com.red.circle.component.mq.service.Action; import com.red.circle.component.mq.service.ConsumerMessage; import com.red.circle.component.mq.service.MessageListener; +import com.red.circle.component.redis.service.RedisService; import com.red.circle.live.app.dto.cmd.HeartbeatStatusEnum; import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; import com.red.circle.live.domain.live.LiveMicrophone; @@ -13,6 +14,7 @@ import com.red.circle.live.domain.online.LiveRoomUser; import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; import com.red.circle.live.infra.database.cache.service.LiveMicUserCacheService; import com.red.circle.live.infra.database.cache.service.LiveRoomCacheService; +import com.red.circle.live.infra.util.DateTimeAsiaRiyadhUtils; import com.red.circle.mq.business.model.event.user.UserHeartbeatEvent; import com.red.circle.mq.rocket.business.producer.UserMqMessageService; import com.red.circle.mq.rocket.business.streams.UserHeartbeatSink; @@ -26,15 +28,19 @@ import com.red.circle.other.inner.model.cmd.user.OnlineStatusUploadCmd; import com.red.circle.other.inner.model.dto.live.RoomProfileDTO; import com.red.circle.other.inner.model.dto.user.UserProfileDTO; import com.red.circle.tool.core.collection.CollectionUtils; +import com.red.circle.tool.core.date.ZonedDateTimeAsiaRiyadhUtils; import com.red.circle.tool.core.parse.DataTypeUtils; import com.red.circle.tool.core.regex.RegexConstant; import com.red.circle.tool.core.sequence.IdWorkerUtils; import com.red.circle.tool.core.text.StringUtils; import java.util.List; import java.util.Objects; +import java.util.concurrent.TimeUnit; + import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Profile; +import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; /** @@ -58,6 +64,9 @@ public class UserHeartbeatListener implements MessageListener { private final LiveRoomCacheService liveRoomCacheService; private final LiveMicUserCacheService liveMicUserCacheService; private final RoomDailyTaskClient roomDailyTaskClient; + private final RedisService redisService; + + private static final int HEARTBEAT_INTERVAL_SECONDS = 60; @Override public Action consume(ConsumerMessage message) { @@ -96,14 +105,26 @@ public class UserHeartbeatListener implements MessageListener { uploadHistoryHeartBeat(event); Long roomId = DataTypeUtils.toLong(event.getSessionId()); - if (roomId != null && roomId > 0) { - RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody(); - if (roomProfile != null && Objects.equals(roomProfile.getUserId(), event.getUserId())) { - List liveRoomUsers = liveMicUserCacheService.listUser(event.getUserId()); - handleOnlineUserCountTask(event.getUserId(), liveRoomUsers.size()); - } + if (roomId == null || roomId <= 0) { + return; } + RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody(); + if (roomProfile == null) { + return; + } + + // 房间内同时在线用户数 + if (Objects.equals(roomProfile.getUserId(), event.getUserId())) { + List liveRoomUsers = liveMicUserCacheService.listUser(event.getUserId()); + handleOnlineUserCountTask(event.getUserId(), liveRoomUsers.size()); + } else { + // 在他人房间活跃 + if (!tryAcquireMinuteTick(event.getUserId())) { + return; + } + updateDailyTask(roomId, event.getUserId(), RoomDailyTaskCode.PERSONAL_ACTIVE_IN_ROOM); + } //log.warn("consume UserHeartbeat end with {}", System.currentTimeMillis() - startTime); } @@ -117,6 +138,26 @@ public class UserHeartbeatListener implements MessageListener { ); } + private void updateDailyTask(Long roomId, Long userId, RoomDailyTaskCode taskType) { + String redisKey = "room:daily:task:progress:" + taskType.name() + ":" + userId + ":" + ZonedDateTimeAsiaRiyadhUtils.now().toLocalDate(); + Long total = redisService.increment(redisKey, 1); + redisService.expire(redisKey, DateTimeAsiaRiyadhUtils.getSecondsUntilMidnight(), TimeUnit.SECONDS); + roomDailyTaskClient.updateTaskProgress( + new RoomDailyTaskProgressUpdateInnerCmd() + .setUserId(userId) + .setTaskCode(taskType.name()) + .setProgressValue(total.intValue()) + ); + } + + /** + * 限流：60秒内只允许累加一次，返回 true 表示本次可以计算 + */ + private boolean tryAcquireMinuteTick(Long userId) { + String key = "user:heartbeat:tick:" + userId ; + return redisService.setIfAbsent(key, "1", HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS); + } + private void processLiveHeartBeat(UserHeartbeatEvent event, UserProfileDTO userProfile) { if (StringUtils.isBlank(event.getSessionId()) || !event.getSessionId().trim().matches(RegexConstant.NUMBER)) { diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/util/DateTimeAsiaRiyadhUtils.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/util/DateTimeAsiaRiyadhUtils.java new file mode 100644 index 00000000..1aa3b8a3 --- /dev/null +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/util/DateTimeAsiaRiyadhUtils.java @@ -0,0 +1,19 @@ +package com.red.circle.live.infra.util; + +import com.red.circle.tool.core.date.ZonedDateTimeAsiaRiyadhUtils; + +import java.time.ZonedDateTime; + +public class DateTimeAsiaRiyadhUtils { + + /** + * 获取当前时间到凌晨的秒数 + * @return 秒数 + */ + public static long getSecondsUntilMidnight() { + ZonedDateTime now = ZonedDateTimeAsiaRiyadhUtils.now(); + ZonedDateTime midnight = now.toLocalDate().plusDays(1).atStartOfDay(now.getZone()); + return java.time.Duration.between(now, midnight).getSeconds(); + } + +}