在他人房间活跃逻辑处理

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) {
+    if (roomId == null || roomId <= 0) {
-      RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody();
+        return;
-      if (roomProfile != null && Objects.equals(roomProfile.getUserId(), event.getUserId())) {
-        List<LiveRoomUser> liveRoomUsers = liveMicUserCacheService.listUser(event.getUserId());
-        handleOnlineUserCountTask(event.getUserId(), liveRoomUsers.size());
-      }
     }
 
+    RoomProfileDTO roomProfile = roomManagerClient.getById(roomId).getBody();
+    if (roomProfile == null) {
+      return;
+    }
+
+    // 房间内同时在线用户数
+    if (Objects.equals(roomProfile.getUserId(), event.getUserId())) {
+      List<LiveRoomUser> 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)) {

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();
+  }
+
+}