feat: async quit room compensation

rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java

@@ -12,10 +12,12 @@ import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatServic
 import com.red.circle.other.inner.endpoint.live.ActiveVoiceRoomClient;
 import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO;
 import com.red.circle.tool.core.collection.CollectionUtils;
+import com.red.circle.tool.core.thread.ThreadPoolManager;
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
 /**
@@ -23,6 +25,7 @@ import org.springframework.stereotype.Component;
  *
  * @author pengliang on 2023/12/12
  */
+@Slf4j
 @Component
 @RequiredArgsConstructor
 public class LiveQuitRoomExe {
@@ -37,45 +40,66 @@ public class LiveQuitRoomExe {
   private static final int CACHE_EXPIRE_MINUTES = 10;
 
   public void execute(AppRoomIdCmd cmd) {
-    // 移除在线用户
+    Long roomId = cmd.getRoomId();
+    Long userId = cmd.requiredReqUserId();
+
+    // 同步删除房间本地在线状态，远端补偿交给后台线程处理
     liveMicrophoneGateway.removeRoomOnlineUser(
-        cmd.getRoomId(),
+        roomId,
-        cmd.requiredReqUserId()
+        userId
     );
 
-    // 下麦克风
+    // 同步下掉当前用户占用的本地麦位，避免客户端继续感知到在房状态
-    List<LiveMicrophone> liveMicrophones = liveMicCacheService.listLiveMicrophone(cmd.getRoomId(),
+    List<LiveMicrophone> liveMicrophones = liveMicCacheService.listLiveMicrophone(roomId, userId);
-        cmd.requiredReqUserId());
+    boolean micChanged = false;
     if (CollectionUtils.isNotEmpty(liveMicrophones)) {
-      liveMicrophones.forEach(mic ->
+      for (LiveMicrophone mic : liveMicrophones) {
-          liveMicrophoneGateway.tryMicOpsLock(cmd.getRoomId(), mic.getMicIndex(), () -> {
+        final boolean[] removed = {false};
-            liveMicCacheService.goDown(cmd.getRoomId(), mic.getMicIndex());
+        liveMicrophoneGateway.tryMicOpsLock(roomId, mic.getMicIndex(), () -> {
-            liveMicrophoneGateway.micUserChangeNotice(cmd.getRoomId());
+          liveMicCacheService.goDown(roomId, mic.getMicIndex());
-          })
+          removed[0] = true;
-      );
+        });
+        micChanged = micChanged || removed[0];
+      }
     }
 
-    // 移除播放音乐
+    // 同步清理本地音乐状态
-    LiveMusicStatus musicStatus = liveMusicHeartbeatService.getMusicUser(cmd.getRoomId());
+    LiveMusicStatus musicStatus = liveMusicHeartbeatService.getMusicUser(roomId);
     if (Objects.nonNull(musicStatus)
-        && Objects.equals(musicStatus.getUserId(), cmd.requiredReqUserId())) {
+        && Objects.equals(musicStatus.getUserId(), userId)) {
-      liveMusicHeartbeatService.removeMickUser(cmd.getRoomId());
+      liveMusicHeartbeatService.removeMickUser(roomId);
     }
 
-    //将没人的房间存入缓存10分钟
+    boolean finalMicChanged = micChanged;
+    ThreadPoolManager.getInstance().execute(() -> {
+      if (finalMicChanged) {
         try {
-      ResultResponse<ActiveVoiceRoomCO> noBodyRoom = activeVoiceRoomClient.createNoBodyRoom(cmd.getRoomId());
+          liveMicrophoneGateway.micUserChangeNotice(roomId);
+        } catch (Exception e) {
+          log.error("quitRoom micUserChangeNotice error roomId={}, userId={}", roomId, userId, e);
+        }
+      }
+
+      cacheEmptyRoom(roomId, userId);
+    });
+  }
+
+  private void cacheEmptyRoom(Long roomId, Long userId) {
+    try {
+      ResultResponse<ActiveVoiceRoomCO> noBodyRoom = activeVoiceRoomClient.createNoBodyRoom(roomId);
       ActiveVoiceRoomCO body = noBodyRoom.getBody();
       if (Objects.nonNull(body)) {
-        // 缓存空房间信息，设置10分钟过期
+        String cacheKey = EMPTY_ROOM_CACHE_KEY + roomId;
-        String cacheKey = EMPTY_ROOM_CACHE_KEY + cmd.getRoomId();
+        redisService.setIfAbsent(
-        redisService.setIfAbsent(cacheKey, JSON.toJSONString(body), CACHE_EXPIRE_MINUTES, TimeUnit.MINUTES);
+            cacheKey,
+            JSON.toJSONString(body),
+            CACHE_EXPIRE_MINUTES,
+            TimeUnit.MINUTES
+        );
       }
     } catch (Exception e) {
-      //ignore
+      log.error("quitRoom cacheEmptyRoom error roomId={}, userId={}", roomId, userId, e);
     }
-
-
   }
 
 }

rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/repository/LiveMicrophoneGatewayImpl.java

@@ -649,20 +649,10 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
 
   @Override
   public void removeRoomOnlineUser(Long roomId, Long userId) {
-    //刷新在线用户信息
+    LiveRoomUserRefresh liveRoomUserRefresh = liveMicUserCacheService.removeUser(roomId, userId);
-    userOnlineClient.updateStatusAndSessionId(userId, UserOnlineStatusEnum.IDLE,0L);
+    ThreadPoolManager.getInstance().execute(
-
+        () -> compensateRemoveRoomOnlineUser(roomId, userId, liveRoomUserRefresh)
-    this.refreshUserProcess(roomId, liveMicUserCacheService.removeUser(roomId, userId));
+    );
-
-    //删除用户的心动值
-    List<LiveHeartbeatCmd> liveHeartbeats = liveHeartbeatClient.listLiveHeartbeat(roomId).getBody();
-    if(CollectionUtils.isNotEmpty(liveHeartbeats)){
-      liveHeartbeats= liveHeartbeats.stream()
-          .filter(item -> !Objects.equals(item.getId(),userId)).collect(
-              Collectors.toList());
-      liveHeartbeatClient.create(roomId, liveHeartbeats);
-    }
-
   }
 
   @Override
@@ -792,4 +782,34 @@ public class LiveMicrophoneGatewayImpl implements LiveMicrophoneGateway {
     return list.stream().collect(Collectors.toMap(LiveHeartbeatCmd::getId,LiveHeartbeatCmd::getHeartbeatVal));
   }
 
+  private void compensateRemoveRoomOnlineUser(Long roomId, Long userId,
+      LiveRoomUserRefresh liveRoomUserRefresh) {
+    try {
+      userOnlineClient.updateStatusAndSessionId(userId, UserOnlineStatusEnum.IDLE, 0L);
+    } catch (Exception e) {
+      log.error("removeRoomOnlineUser updateStatusAndSessionId error roomId={}, userId={}", roomId,
+          userId, e);
+    }
+
+    try {
+      refreshUserProcess(roomId, liveRoomUserRefresh);
+    } catch (Exception e) {
+      log.error("removeRoomOnlineUser refreshUserProcess error roomId={}, userId={}", roomId,
+          userId, e);
+    }
+
+    try {
+      List<LiveHeartbeatCmd> liveHeartbeats = liveHeartbeatClient.listLiveHeartbeat(roomId).getBody();
+      if (CollectionUtils.isNotEmpty(liveHeartbeats)) {
+        liveHeartbeats = liveHeartbeats.stream()
+            .filter(item -> !Objects.equals(item.getId(), userId))
+            .collect(Collectors.toList());
+        liveHeartbeatClient.create(roomId, liveHeartbeats);
+      }
+    } catch (Exception e) {
+      log.error("removeRoomOnlineUser refreshHeartbeat error roomId={}, userId={}", roomId, userId,
+          e);
+    }
+  }
+
 }