火箭能量更新性能优化

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/manager/RocketEnergyAggregator.java

@@ -9,22 +9,19 @@ import com.red.circle.other.domain.gateway.RocketStatusGateway;
 import com.red.circle.other.domain.rocket.RocketConfig;
 import com.red.circle.other.domain.rocket.RocketStatus;
 import com.red.circle.other.infra.database.cache.key.RocketKeys;
-import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager;
 import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManagerService;
 import com.red.circle.other.infra.gateway.RocketStatusGatewayImpl;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
-import javax.annotation.PostConstruct;
 import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 /**
  * 火箭能量聚合器
@@ -50,18 +47,18 @@ public class RocketEnergyAggregator {
     private final RoomProfileManagerService roomProfileManagerService;
 
     /**
-     * 房间能量聚合缓存
+     * 房间能量聚合缓存（高性能线程安全队列）
      * Key: roomId
-     * Value: 该房间待聚合的能量列表
+     * Value: 该房间待聚合的能量队列
      */
-    private final Map<Long, List<RocketEnergyAddCmd>> energyBuffer = new ConcurrentHashMap<>();
+    private final Map<Long, ConcurrentLinkedQueue<RocketEnergyAddCmd>> energyBuffer = new ConcurrentHashMap<>();
 
     /**
-     * 房间级别的细粒度锁
+     * 上次推送时间
      * Key: roomId
-     * Value: 该房间的锁
+     * Value: 上次推送的时间戳
      */
-    private final Map<Long, ReentrantLock> roomLocks = new ConcurrentHashMap<>();
+    private final Map<Long, Long> lastPushTime = new ConcurrentHashMap<>();
 
     /**
      * 推送间隔（毫秒）
@@ -74,21 +71,14 @@ public class RocketEnergyAggregator {
     private static final long FLUSH_INTERVAL = 5000L; // 5秒刷新一次
 
     /**
-     * 上次推送时间
+     * 接收能量增加请求
-     * Key: roomId
-     * Value: 上次推送的时间戳
-     */
-    private final Map<Long, Long> lastPushTime = new ConcurrentHashMap<>();
-
-    /**
-     * 接收能量增加请求（无锁，直接缓存）
      */
     public void addEnergy(RocketEnergyAddCmd cmd) {
-        energyBuffer.computeIfAbsent(cmd.getRoomId(), k -> new ArrayList<>()).add(cmd);
+        energyBuffer.computeIfAbsent(cmd.getRoomId(), k -> new ConcurrentLinkedQueue<>())
+                    .offer(cmd);
         
-        log.debug("缓存火箭能量: roomId={}, userId={}, energy={}, bufferSize={}", 
+        log.debug("缓存火箭能量: roomId={}, userId={}, energy={}", 
-                cmd.getRoomId(), cmd.getUserId(), cmd.getGiftGoldValue(),
+                cmd.getRoomId(), cmd.getUserId(), cmd.getGiftGoldValue());
-                energyBuffer.get(cmd.getRoomId()).size());
     }
 
     /**
@@ -103,30 +93,33 @@ public class RocketEnergyAggregator {
         log.info("开始批量刷新火箭能量, 待处理房间数={}", energyBuffer.size());
 
         // 遍历所有房间
-        energyBuffer.forEach((roomId, cmdList) -> {
+        energyBuffer.forEach((roomId, queue) -> {
-            if (cmdList.isEmpty()) {
+            if (queue.isEmpty()) {
                 return;
             }
 
             try {
-                // 获取该房间的细粒度锁
+                String lockKey = RocketKeys.LOCK.getKey(roomId);
-                ReentrantLock roomLock = roomLocks.computeIfAbsent(roomId, k -> new ReentrantLock());
                 
-                if (roomLock.tryLock()) {
+                distributedLockUtil.executeWithLock(lockKey, 10L, () -> {
-                    try {
+                    
-                        // 取出当前批次的数据
+                    // 原子性地取出所有数据
-                        List<RocketEnergyAddCmd> batch = new ArrayList<>(cmdList);
+                    List<RocketEnergyAddCmd> batch = new ArrayList<>();
-                        cmdList.clear();
+                    RocketEnergyAddCmd cmd;
-
+                    while ((cmd = queue.poll()) != null) {
-                        // 处理这个房间的能量增加
+                        batch.add(cmd);
-                        processRoomEnergy(roomId, batch);
-                        
-                    } finally {
-                        roomLock.unlock();
                     }
-                } else {
+                    
-                    log.debug("房间{}正在处理中，跳过本次刷新", roomId);
+                    if (batch.isEmpty()) {
-                }
+                        return null;
+                    }
+
+                    // 处理这个房间的能量增加
+                    processRoomEnergyBatch(roomId, batch);
+                    return null;
+                    
+                }, "火箭能量刷新操作中");
+                
             } catch (Exception e) {
                 log.error("刷新房间{}火箭能量失败", roomId, e);
             }
@@ -134,80 +127,67 @@ public class RocketEnergyAggregator {
     }
 
     /**
-     * 处理单个房间的能量增加
+     * 处理单个房间的能量批次
      */
-    private void processRoomEnergy(Long roomId, List<RocketEnergyAddCmd> cmdList) {
+    private void processRoomEnergyBatch(Long roomId, List<RocketEnergyAddCmd> batch) {
-        if (cmdList.isEmpty()) {
+        // 1. 查询当前火箭状态（不包含贡献者列表，性能优化）
+        String today = LocalDate.now().toString();
+        RocketStatus rocketStatus = rocketStatusGatewayImpl
+                .findByRoomIdAndDateWithoutContributors(roomId, today);
+
+        if (rocketStatus == null) {
+            // 查询一级火箭配置
+            RocketConfig level1Config = rocketConfigGateway.findByLevel(1);
+            if (level1Config == null || level1Config.getStatus() != 1) {
+                log.error("一级火箭配置不存在或未启用");
+                return;
+            }
+
+            // 使用配置初始化
+            rocketStatus = RocketStatus.init(roomId, level1Config);
+            log.info("初始化房间火箭, roomId={}, maxEnergy={}", roomId, level1Config.getMaxEnergy());
+        }
+
+        // 2. 检查状态
+        if (!rocketStatus.canAddEnergy()) {
+            log.warn("火箭不可充能, roomId={}, status={}", roomId, rocketStatus.getStatus());
             return;
         }
 
-        String lockKey = RocketKeys.LOCK.getKey(roomId);
+        // 3. 批量增加能量
-        
+        long totalEnergy = 0L;
-        // 使用Redis分布式锁，避免多实例冲突
+        for (RocketEnergyAddCmd cmd : batch) {
-        distributedLockUtil.executeWithLock(lockKey, 10L, () -> {
+            rocketStatus.addEnergy(
+                    cmd.getUserId(),
+                    cmd.getGiftGoldValue()
+            );
+            totalEnergy += cmd.getGiftGoldValue();
+        }
+
+        // 4. 保存状态（一次性保存）
+        rocketStatusGateway.save(rocketStatus);
+
+        log.info("批量增加火箭能量: roomId={}, 批次数量={}, 总能量={}, 当前能量={}/{}", 
+                roomId, batch.size(), totalEnergy, 
+                rocketStatus.getCurrentEnergy(), rocketStatus.getMaxEnergy());
+
+        // 5. 推送更新消息（节流）
+        pushEnergyUpdate(roomId, rocketStatus);
+
+        // 6. 检查是否需要发射
+        if (rocketStatus.isFullEnergy()) {
+            log.info("火箭能量已满, 准备发射, roomId={}, level={}", 
+                    roomId, rocketStatus.getLevel().getLevel());
             
-            // 1. 查询当前火箭状态（不包含贡献者列表，性能优化）
+            // 异步触发发射
-            String today = LocalDate.now().toString();
+            rocketLaunchManager.launchAsync(roomId);
-            RocketStatus rocketStatus = rocketStatusGatewayImpl
+        }
-                    .findByRoomIdAndDateWithoutContributors(roomId, today);
-
-            if (rocketStatus == null) {
-                // 查询一级火箭配置
-                RocketConfig level1Config = rocketConfigGateway.findByLevel(1);
-                if (level1Config == null || level1Config.getStatus() != 1) {
-                    log.error("一级火箭配置不存在或未启用");
-                    return null;
-                }
-
-                // 使用配置初始化
-                rocketStatus = RocketStatus.init(roomId, level1Config);
-                log.info("初始化房间火箭, roomId={}, maxEnergy={}", roomId, level1Config.getMaxEnergy());
-            }
-
-            // 2. 检查状态
-            if (!rocketStatus.canAddEnergy()) {
-                log.warn("火箭不可充能, roomId={}, status={}", roomId, rocketStatus.getStatus());
-                return null;
-            }
-
-            // 3. 批量增加能量
-            long totalEnergy = 0L;
-            for (RocketEnergyAddCmd cmd : cmdList) {
-                rocketStatus.addEnergy(
-                        cmd.getUserId(),
-                        cmd.getGiftGoldValue()
-                );
-                totalEnergy += cmd.getGiftGoldValue();
-            }
-
-            // 4. 保存状态（一次性保存）
-            rocketStatusGateway.save(rocketStatus);
-
-            log.info("批量增加火箭能量: roomId={}, 批次数量={}, 总能量={}, 当前能量={}/{}", 
-                    roomId, cmdList.size(), totalEnergy, 
-                    rocketStatus.getCurrentEnergy(), rocketStatus.getMaxEnergy());
-
-            // 5. 推送更新消息（节流）
-            String roomAccount = roomProfileManagerService.getRoomAccount(roomId);
-            pushEnergyUpdate(roomId, roomAccount, rocketStatus);
-
-            // 6. 检查是否需要发射
-            if (rocketStatus.isFullEnergy()) {
-                log.info("火箭能量已满, 准备发射, roomId={}, level={}", 
-                        roomId, rocketStatus.getLevel().getLevel());
-                
-                // 异步触发发射
-                rocketLaunchManager.launchAsync(roomId);
-            }
-
-            return null;
-        }, "火箭能量刷新操作中，请稍后");
     }
 
     /**
      * 推送能量更新（节流：3秒推送一次）
      */
-    private void pushEnergyUpdate(Long roomId, String roomAccount, RocketStatus rocketStatus) {
+    private void pushEnergyUpdate(Long roomId, RocketStatus rocketStatus) {
         long now = System.currentTimeMillis();
         Long lastPush = lastPushTime.get(roomId);
         
@@ -218,36 +198,28 @@ public class RocketEnergyAggregator {
         }
 
         try {
-            // 查询火箭状态
+            String roomAccount = roomProfileManagerService.getRoomAccount(roomId);
-            RocketStatus fullStatus = rocketStatusGatewayImpl
+            RocketStatusCO statusCO = RocketConvertor.toStatusCO(rocketStatus);
-                    .findByRoomIdAndDateWithoutContributors(roomId, rocketStatus.getDate());
+            rocketImPushManager.pushEnergyUpdate(roomAccount, statusCO);
             
-            if (fullStatus != null) {
+            // 更新推送时间
-                RocketStatusCO statusCO = RocketConvertor.toStatusCO(fullStatus);
+            lastPushTime.put(roomId, now);
-                rocketImPushManager.pushEnergyUpdate(roomAccount, statusCO);
+            
-                
+            log.info("推送火箭能量更新: roomId={}, energy={}/{}", 
-                // 更新推送时间
+                    roomId, rocketStatus.getCurrentEnergy(), rocketStatus.getMaxEnergy());
-                lastPushTime.put(roomId, now);
-                
-                log.info("推送火箭能量更新: roomId={}, energy={}/{}", 
-                        roomId, fullStatus.getCurrentEnergy(), fullStatus.getMaxEnergy());
-            }
         } catch (Exception e) {
             log.error("推送火箭能量更新失败: roomId={}", roomId, e);
         }
     }
 
     /**
-     * 定时清理空闲房间的锁（每小时）
+     * 定时清理空闲房间的资源（每小时）
      */
     @Scheduled(fixedRate = 3600000)
     public void cleanupIdleRooms() {
-        // 清理buffer中的空列表
+        // 清理buffer中的空队列
         energyBuffer.entrySet().removeIf(entry -> entry.getValue().isEmpty());
         
-        // 清理不在buffer中的锁
-        roomLocks.keySet().removeIf(roomId -> !energyBuffer.containsKey(roomId));
-        
         // 清理推送时间
         lastPushTime.keySet().removeIf(roomId -> !energyBuffer.containsKey(roomId));