火箭能量异步刷新数据丢失 问题处理2

rc-service/rc-service-other/other-adapter/src/main/java/com/red/circle/other/adapter/app/RocketMonitorRestController.java

@@ -0,0 +1,35 @@
+package com.red.circle.other.adapter.app;
+
+import com.red.circle.other.app.manager.RocketEnergyAggregator;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Map;
+
+/**
+ * 火箭能量队列监控接口
+ */
+@RestController
+@RequestMapping("/rocket/monitor")
+@RequiredArgsConstructor
+public class RocketMonitorRestController {
+
+    private final RocketEnergyAggregator rocketEnergyAggregator;
+
+    /**
+     * 获取队列状态
+     */
+    @GetMapping("/queue/status")
+    public Map<String, Object> getQueueStatus() {
+        return rocketEnergyAggregator.getQueueStatus();
+    }
+
+    /**
+     * 手动触发刷新
+     */
+    @PostMapping("/queue/flush")
+    public String manualFlush() {
+        rocketEnergyAggregator.manualFlush();
+        return "刷新任务已触发";
+    }
+}

rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/rocket/RocketLaunchCmdExe.java

@@ -62,7 +62,7 @@ public class RocketLaunchCmdExe {
      * @param triggerUserId 触发升级的用户ID
      */
     public void execute(Long roomId, Long triggerUserId) {
-        String lockKey = RocketKeys.LOCK.getKey(roomId);
+        String lockKey = RocketKeys.LAUNCH.getKey(roomId);
         
         distributedLockUtil.executeWithLock(lockKey, 10L, () -> {
             doLaunch(roomId, triggerUserId);

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

@@ -1,5 +1,6 @@
 package com.red.circle.other.app.manager;
 
+import com.alibaba.fastjson.JSON;
 import com.red.circle.other.app.convertor.RocketConvertor;
 import com.red.circle.other.app.dto.clientobject.RocketStatusCO;
 import com.red.circle.other.app.dto.cmd.RocketEnergyAddCmd;
@@ -8,27 +9,23 @@ import com.red.circle.other.domain.gateway.RocketConfigGateway;
 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.domain.rocket.RocketStatusEnum;
 import com.red.circle.other.infra.database.cache.key.RocketKeys;
 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.StringRedisTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
 import java.time.LocalDate;
-import java.util.ArrayList;
+import java.util.*;
-import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
 
 /**
- * 火箭能量聚合器
+ * 火箭能量聚合器（Redis队列 + 多实例消费版本）
- * 职责：
- * 1. 接收能量增加请求，缓存到内存
- * 2. 定时批量刷新到数据库
- * 3. 定时推送能量更新消息
  *
  * @author system
  * @date 2025-01-15
@@ -45,145 +42,235 @@ public class RocketEnergyAggregator {
     private final RocketImPushManager rocketImPushManager;
     private final RocketLaunchManager rocketLaunchManager;
     private final RoomProfileManagerService roomProfileManagerService;
+    private final StringRedisTemplate redisTemplate;
 
     /**
-     * 房间能量聚合缓存（高性能线程安全队列）
+     * Redis 队列 Key 前缀
-     * Key: roomId
-     * Value: 该房间待聚合的能量队列
      */
-    private final Map<Long, ConcurrentLinkedQueue<RocketEnergyAddCmd>> energyBuffer = new ConcurrentHashMap<>();
+    private static final String ENERGY_QUEUE_PREFIX = "rocket:energy:queue:";
+    
+    /**
+     * Redis 房间列表 Key（记录所有有数据的房间）
+     */
+    private static final String ACTIVE_ROOMS_KEY = "rocket:energy:active:rooms";
+    
+    /**
+     * 推送间隔
+     */
+    private static final long PUSH_INTERVAL = 3000L;
+    
+    /**
+     * 批量刷新间隔
+     */
+    private static final long FLUSH_INTERVAL = 5000L;
+    
+    /**
+     * 每次最大处理数量
+     */
+    private static final int MAX_BATCH_SIZE = 100;
+    
+    /**
+     * 锁超时时间（秒）
+     */
+    private static final long LOCK_TIMEOUT = 10L;
 
     /**
-     * 上次推送时间
+     * 上次推送时间（本地缓存）
-     * Key: roomId
-     * Value: 上次推送的时间戳
      */
     private final Map<Long, Long> lastPushTime = new ConcurrentHashMap<>();
 
     /**
-     * 推送间隔（毫秒）
+     * 接收能量增加请求（写入Redis队列）
-     */
-    private static final long PUSH_INTERVAL = 3000L; // 3秒推送一次
-
-    /**
-     * 批量刷新间隔（毫秒）
-     */
-    private static final long FLUSH_INTERVAL = 5000L; // 5秒刷新一次
-
-    /**
-     * 接收能量增加请求
      */
     public void addEnergy(RocketEnergyAddCmd cmd) {
-        energyBuffer.computeIfAbsent(cmd.getRoomId(), k -> new ConcurrentLinkedQueue<>())
+        try {
-                    .offer(cmd);
+            Long roomId = cmd.getRoomId();
-        
+            String queueKey = ENERGY_QUEUE_PREFIX + roomId;
-        log.debug("缓存火箭能量: roomId={}, userId={}, energy={}", 
+            String value = JSON.toJSONString(cmd);
-                cmd.getRoomId(), cmd.getUserId(), cmd.getGiftGoldValue());
+            
+            // 写入队列
+            redisTemplate.opsForList().rightPush(queueKey, value);
+            
+            // 记录活跃房间（使用 Set 去重）
+            redisTemplate.opsForSet().add(ACTIVE_ROOMS_KEY, roomId.toString());
+            
+            // 设置队列过期时间（24小时）
+            redisTemplate.expire(queueKey, 24, TimeUnit.HOURS);
+            
+            log.debug("能量写入Redis队列: roomId={}, userId={}, energy={}", 
+                    roomId, cmd.getUserId(), cmd.getGiftGoldValue());
+                    
+        } catch (Exception e) {
+            log.error("写入能量队列失败: roomId={}", cmd.getRoomId(), e);
+            throw e;
+        }
     }
 
     /**
-     * 定时批量刷新到数据库（每5秒）
+     * 定时批量刷新
      */
     @Scheduled(fixedDelay = FLUSH_INTERVAL)
     public void flushToDatabase() {
-        if (energyBuffer.isEmpty()) {
+        try {
-            return;
+            // 获取所有活跃房间
-        }
+            Set<String> activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY);
-
+            
-        log.info("开始批量刷新火箭能量, 待处理房间数={}", energyBuffer.size());
+            if (activeRooms == null || activeRooms.isEmpty()) {
-
+                log.debug("没有待处理的房间");
-        // 遍历所有房间
-        energyBuffer.forEach((roomId, queue) -> {
-            if (queue.isEmpty()) {
                 return;
             }
 
-            try {
+            log.info("开始批量刷新火箭能量, 待处理房间数={}", activeRooms.size());
-                String lockKey = RocketKeys.LOCK.getKey(roomId);
+            
+            int successCount = 0;
+            int skipCount = 0;
+            int errorCount = 0;
 
-                distributedLockUtil.executeWithLock(lockKey, 10L, () -> {
+            // 遍历所有活跃房间，尝试获取锁并处理
-
+            for (String roomIdStr : activeRooms) {
-                    // 先 peek 不删除，等处理成功后再删除
+                Long roomId = null;
-                    List<RocketEnergyAddCmd> batch = new ArrayList<>(queue); // 复制数据，不删除原队列
+                try {
-
+                    roomId = Long.parseLong(roomIdStr);
-                    if (batch.isEmpty()) {
+                    
-                        return null;
+                    // 尝试获取房间锁（非阻塞）
+                    String lockKey = RocketKeys.LOCK.getKey(roomId);
+                    Boolean locked = distributedLockUtil.tryLock(lockKey, LOCK_TIMEOUT);
+                    
+                    if (!locked) {
+                        // 其他实例正在处理这个房间，跳过
+                        skipCount++;
+                        log.debug("房间{}被其他实例占用，跳过", roomId);
+                        continue;
                     }
 
                     try {
-                        // 处理这个房间的能量增加
+                        // 获取锁成功，处理这个房间
-                        processRoomEnergyBatch(roomId, batch);
+                        boolean processed = processRoomEnergyFromRedis(roomId);
-
+                        if (processed) {
-                        // 处理成功后才删除队列中的数据
+                            successCount++;
-                        for (int i = 0; i < batch.size(); i++) {
-                            queue.poll();
                         }
-
+                    } finally {
-                    } catch (Exception e) {
+                        // 释放锁
-                        log.error("处理房间{}火箭能量失败，数据保留在队列中等待重试", roomId, e);
+                        distributedLockUtil.unlock(lockKey);
-                        throw e; // 重新抛出，让外层处理
                     }
-
+                    
-                    return null;
+                } catch (Exception e) {
-
+                    errorCount++;
-                }, "火箭能量刷新操作中");
+                    log.error("处理房间{}能量失败", roomId, e);
-                
+                }
-            } catch (Exception e) {
-                log.error("刷新房间{}火箭能量失败", roomId, e);
             }
-        });
+
+            log.info("批量刷新完成: 成功={}, 跳过={}, 失败={}", successCount, skipCount, errorCount);
+            
+        } catch (Exception e) {
+            log.error("批量刷新火箭能量失败", e);
+        }
+    }
+
+    /**
+     * 从Redis队列处理房间能量
+     * @return 是否处理了数据
+     */
+    private boolean processRoomEnergyFromRedis(Long roomId) {
+        String queueKey = ENERGY_QUEUE_PREFIX + roomId;
+        
+        try {
+            // 检查队列长度
+            Long queueSize = redisTemplate.opsForList().size(queueKey);
+            if (queueSize == null || queueSize == 0) {
+                // 队列为空，从活跃房间列表中移除
+                redisTemplate.opsForSet().remove(ACTIVE_ROOMS_KEY, roomId.toString());
+                log.debug("房间{}队列为空，从活跃列表移除", roomId);
+                return false;
+            }
+
+            // 批量取出数据（使用 LRANGE 查看，不删除）
+            List<String> values = redisTemplate.opsForList().range(queueKey, 0, MAX_BATCH_SIZE - 1);
+            
+            if (values == null || values.isEmpty()) {
+                return false;
+            }
+
+            // 解析数据
+            List<RocketEnergyAddCmd> batch = new ArrayList<>();
+            for (String value : values) {
+                try {
+                    RocketEnergyAddCmd cmd = JSON.parseObject(value, RocketEnergyAddCmd.class);
+                    batch.add(cmd);
+                } catch (Exception e) {
+                    log.error("解析能量数据失败，写入死信队列: {}", value, e);
+                    redisTemplate.opsForList().rightPush("rocket:energy:dead:letter", value);
+                }
+            }
+
+            if (batch.isEmpty()) {
+                // 数据解析失败，删除这些无效数据
+                redisTemplate.opsForList().trim(queueKey, values.size(), -1);
+                return false;
+            }
+
+            // 处理能量增加
+            processRoomEnergyBatch(roomId, batch);
+            
+            // 处理成功后才删除Redis中的数据（使用 LTRIM）
+            redisTemplate.opsForList().trim(queueKey, batch.size(), -1);
+            
+            log.info("成功处理房间{}能量, 处理数量={}, 剩余={}", 
+                    roomId, batch.size(), queueSize - batch.size());
+            
+            return true;
+            
+        } catch (Exception e) {
+            log.error("处理房间{}能量失败，数据保留在Redis中等待重试", roomId, e);
+            throw e;
+        }
     }
 
     /**
      * 处理单个房间的能量批次
      */
     private void processRoomEnergyBatch(Long roomId, List<RocketEnergyAddCmd> batch) {
-        // 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;
+                throw new RuntimeException("一级火箭配置不存在");
             }
-
-            // 使用配置初始化
             rocketStatus = RocketStatus.init(roomId, level1Config);
             log.info("初始化房间火箭, roomId={}, maxEnergy={}", roomId, level1Config.getMaxEnergy());
         }
 
-        // 2. 检查状态
+        // 校验状态
         if (!rocketStatus.canAddEnergy()) {
             log.warn("火箭不可充能, roomId={}, status={}", roomId, rocketStatus.getStatus());
-            return;
+            throw new RuntimeException("火箭状态不允许充能");
         }
 
-        // 3. 记录增加能量前的状态
+        // 记录增加前的状态
         long energyBeforeAdd = rocketStatus.getCurrentEnergy();
         long maxEnergy = rocketStatus.getMaxEnergy();
         
-        // 4. 批量增加能量，并找出触发升级的用户
+        // 批量增加能量
         long totalEnergy = 0L;
         Long triggerUserId = null;
+        
         for (RocketEnergyAddCmd cmd : batch) {
-            // 增加前的能量
             long beforeEnergy = rocketStatus.getCurrentEnergy();
             
             // 增加能量
-            rocketStatus.addEnergy(
+            rocketStatus.addEnergy(cmd.getUserId(), cmd.getGiftGoldValue());
-                    cmd.getUserId(),
-                    cmd.getGiftGoldValue()
-            );
             totalEnergy += cmd.getGiftGoldValue();
             
-            // 增加后的能量
             long afterEnergy = rocketStatus.getCurrentEnergy();
             
-            // 判断是否是这个用户触发了满能量
+            // 记录触发满能量的用户
             if (beforeEnergy < maxEnergy && afterEnergy >= maxEnergy && triggerUserId == null) {
                 triggerUserId = cmd.getUserId();
                 log.info("找到触发升级的用户: userId={}, beforeEnergy={}, afterEnergy={}, maxEnergy={}",
@@ -191,22 +278,27 @@ public class RocketEnergyAggregator {
             }
         }
 
-        // 5. 保存状态（一次性保存）
+        // 保存状态
-        rocketStatusGateway.save(rocketStatus);
+        try {
+            rocketStatusGateway.save(rocketStatus);
+        } catch (Exception e) {
+            log.error("保存火箭状态失败: roomId={}", roomId, e);
+            throw e;
+        }
 
-        log.info("批量增加火箭能量: roomId={}, 批次数量={}, 总能量={}, 当前能量={}/{}", 
+        log.info("批量增加火箭能量: roomId={}, 批次数量={}, 总能量={}, 更新前={}, 更新后={}/{}", 
-                roomId, batch.size(), totalEnergy, 
+                roomId, batch.size(), totalEnergy, energyBeforeAdd,
                 rocketStatus.getCurrentEnergy(), rocketStatus.getMaxEnergy());
 
-        // 6. 推送更新消息（节流）
+        // 推送更新消息（节流）
         pushEnergyUpdate(roomId, rocketStatus);
 
-        // 7. 检查是否需要发射
+        // 检查是否需要发射（添加状态检查）
-        if (rocketStatus.isFullEnergy()) {
+        if (rocketStatus.isFullEnergy() && rocketStatus.getStatus() == RocketStatusEnum.CHARGING) {
             log.info("火箭能量已满, 准备发射, roomId={}, level={}, 触发用户={}", 
                     roomId, rocketStatus.getLevel().getLevel(), triggerUserId);
             
-            // 异步触发发射，传入触发用户信息
+            // 异步触发发射
             rocketLaunchManager.launchAsync(roomId, triggerUserId);
         }
     }
@@ -240,17 +332,30 @@ public class RocketEnergyAggregator {
     }
 
     /**
-     * 定时清理空闲房间的资源（每小时）
+     * 定时清理过期数据（每小时）
      */
     @Scheduled(fixedRate = 3600000)
-    public void cleanupIdleRooms() {
+    public void cleanupExpiredData() {
-        // 清理buffer中的空队列
+        try {
-        energyBuffer.entrySet().removeIf(entry -> entry.getValue().isEmpty());
+            // 清理本地推送时间缓存
-        
+            Set<String> activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY);
-        // 清理推送时间
+            if (activeRooms != null) {
-        lastPushTime.keySet().removeIf(roomId -> !energyBuffer.containsKey(roomId));
+                Set<Long> activeRoomIds = new HashSet<>();
-        
+                for (String roomIdStr : activeRooms) {
-        log.info("清理空闲房间资源, 当前活跃房间数={}", energyBuffer.size());
+                    try {
+                        activeRoomIds.add(Long.parseLong(roomIdStr));
+                    } catch (Exception ignored) {
+                    }
+                }
+
+                // 移除不活跃房间的推送时间
+                lastPushTime.keySet().removeIf(roomId -> !activeRoomIds.contains(roomId));
+            }
+
+            log.info("清理过期数据完成, 当前活跃房间数={}", activeRooms != null ? activeRooms.size() : 0);
+        } catch (Exception e) {
+            log.error("清理过期数据失败", e);
+        }
     }
 
     /**
@@ -259,6 +364,45 @@ public class RocketEnergyAggregator {
     @javax.annotation.PreDestroy
     public void shutdown() {
         log.info("应用关闭，强制刷新所有待处理能量");
+        try {
+            flushToDatabase();
+        } catch (Exception e) {
+            log.error("关闭时刷新失败", e);
+        }
+    }
+
+    /**
+     * 手动触发刷新（供运维使用）
+     */
+    public void manualFlush() {
+        log.info("手动触发刷新");
         flushToDatabase();
     }
-}
+
+    /**
+     * 获取队列状态（监控使用）
+     */
+    public Map<String, Object> getQueueStatus() {
+        Map<String, Object> status = new HashMap<>();
+        
+        Set<String> activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY);
+        status.put("activeRoomCount", activeRooms != null ? activeRooms.size() : 0);
+        status.put("localPushCacheSize", lastPushTime.size());
+        
+        if (activeRooms != null && !activeRooms.isEmpty()) {
+            Map<Long, Long> queueSizes = new HashMap<>();
+            for (String roomIdStr : activeRooms) {
+                try {
+                    Long roomId = Long.parseLong(roomIdStr);
+                    String queueKey = ENERGY_QUEUE_PREFIX + roomId;
+                    Long size = redisTemplate.opsForList().size(queueKey);
+                    queueSizes.put(roomId, size != null ? size : 0L);
+                } catch (Exception ignored) {
+                }
+            }
+            status.put("queueSizes", queueSizes);
+        }
+        
+        return status;
+    }
+}

rc-service/rc-service-other/other-domain/src/main/java/com/red/circle/other/domain/rocket/RocketHistory.java

@@ -89,7 +89,7 @@ public class RocketHistory {
         history.setFinalEnergy(status.getCurrentEnergy());
 
         // 计算超出能量
-        Long overEnergy = status.getCurrentEnergy() - status.getLevel().getMaxEnergy();
+        Long overEnergy = status.getCurrentEnergy() - status.getMaxEnergy();
         history.setOverEnergy(Math.max(overEnergy, 0L));
 
         // 转换贡献者

rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/key/RocketKeys.java

@@ -42,6 +42,8 @@ public enum RocketKeys implements RedisKeys {
      */
     CLAIMED,
 
+    LAUNCH,
+
     /**
      * 火箭配置缓存
      * KEY: ROCKET:CONFIG:{level}