From 1be939fd0e6052b5820369fa2b8d9c7716894138 Mon Sep 17 00:00:00 2001 From: tianfeng <769204422@qq.com> Date: Tue, 18 Nov 2025 01:18:09 +0800 Subject: [PATCH] =?UTF-8?q?=E7=81=AB=E7=AE=AD=E8=83=BD=E9=87=8F=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E5=88=B7=E6=96=B0=E6=95=B0=E6=8D=AE=E4=B8=A2=E5=A4=B1?= =?UTF-8?q?=20=E9=97=AE=E9=A2=98=E5=A4=84=E7=90=862?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/RocketMonitorRestController.java | 35 ++ .../command/rocket/RocketLaunchCmdExe.java | 2 +- .../app/manager/RocketEnergyAggregator.java | 350 ++++++++++++------ .../other/domain/rocket/RocketHistory.java | 2 +- .../infra/database/cache/key/RocketKeys.java | 2 + 5 files changed, 286 insertions(+), 105 deletions(-) create mode 100644 rc-service/rc-service-other/other-adapter/src/main/java/com/red/circle/other/adapter/app/RocketMonitorRestController.java diff --git a/rc-service/rc-service-other/other-adapter/src/main/java/com/red/circle/other/adapter/app/RocketMonitorRestController.java b/rc-service/rc-service-other/other-adapter/src/main/java/com/red/circle/other/adapter/app/RocketMonitorRestController.java new file mode 100644 index 00000000..6a65090e --- /dev/null +++ b/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 getQueueStatus() { + return rocketEnergyAggregator.getQueueStatus(); + } + + /** + * 手动触发刷新 + */ + @PostMapping("/queue/flush") + public String manualFlush() { + rocketEnergyAggregator.manualFlush(); + return "刷新任务已触发"; + } +} \ No newline at end of file diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/rocket/RocketLaunchCmdExe.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/rocket/RocketLaunchCmdExe.java index 52a2ed58..43ee6bb0 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/rocket/RocketLaunchCmdExe.java +++ b/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); diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/manager/RocketEnergyAggregator.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/manager/RocketEnergyAggregator.java index f9af6bbd..2a7b76c6 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/manager/RocketEnergyAggregator.java +++ b/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.List; -import java.util.Map; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.TimeUnit; /** - * 火箭能量聚合器 - * 职责： - * 1. 接收能量增加请求，缓存到内存 - * 2. 定时批量刷新到数据库 - * 3. 定时推送能量更新消息 + * 火箭能量聚合器（Redis队列 + 多实例消费版本） * * @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; /** - * 房间能量聚合缓存（高性能线程安全队列） - * Key: roomId - * Value: 该房间待聚合的能量队列 + * Redis 队列 Key 前缀 */ - private final Map> 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 lastPushTime = new ConcurrentHashMap<>(); /** - * 推送间隔（毫秒） - */ - private static final long PUSH_INTERVAL = 3000L; // 3秒推送一次 - - /** - * 批量刷新间隔（毫秒） - */ - private static final long FLUSH_INTERVAL = 5000L; // 5秒刷新一次 - - /** - * 接收能量增加请求 + * 接收能量增加请求（写入Redis队列） */ public void addEnergy(RocketEnergyAddCmd cmd) { - energyBuffer.computeIfAbsent(cmd.getRoomId(), k -> new ConcurrentLinkedQueue<>()) - .offer(cmd); - - log.debug("缓存火箭能量: roomId={}, userId={}, energy={}", - cmd.getRoomId(), cmd.getUserId(), cmd.getGiftGoldValue()); + try { + Long roomId = cmd.getRoomId(); + String queueKey = ENERGY_QUEUE_PREFIX + roomId; + String value = JSON.toJSONString(cmd); + + // 写入队列 + 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()) { - return; - } - - log.info("开始批量刷新火箭能量, 待处理房间数={}", energyBuffer.size()); - - // 遍历所有房间 - energyBuffer.forEach((roomId, queue) -> { - if (queue.isEmpty()) { + try { + // 获取所有活跃房间 + Set activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY); + + if (activeRooms == null || activeRooms.isEmpty()) { + log.debug("没有待处理的房间"); return; } - try { - String lockKey = RocketKeys.LOCK.getKey(roomId); + log.info("开始批量刷新火箭能量, 待处理房间数={}", activeRooms.size()); + + int successCount = 0; + int skipCount = 0; + int errorCount = 0; - distributedLockUtil.executeWithLock(lockKey, 10L, () -> { - - // 先 peek 不删除，等处理成功后再删除 - List batch = new ArrayList<>(queue); // 复制数据，不删除原队列 - - if (batch.isEmpty()) { - return null; + // 遍历所有活跃房间，尝试获取锁并处理 + for (String roomIdStr : activeRooms) { + Long roomId = null; + try { + roomId = Long.parseLong(roomIdStr); + + // 尝试获取房间锁（非阻塞） + String lockKey = RocketKeys.LOCK.getKey(roomId); + Boolean locked = distributedLockUtil.tryLock(lockKey, LOCK_TIMEOUT); + + if (!locked) { + // 其他实例正在处理这个房间，跳过 + skipCount++; + log.debug("房间{}被其他实例占用，跳过", roomId); + continue; } try { - // 处理这个房间的能量增加 - processRoomEnergyBatch(roomId, batch); - - // 处理成功后才删除队列中的数据 - for (int i = 0; i < batch.size(); i++) { - queue.poll(); + // 获取锁成功，处理这个房间 + boolean processed = processRoomEnergyFromRedis(roomId); + if (processed) { + successCount++; } - - } catch (Exception e) { - log.error("处理房间{}火箭能量失败，数据保留在队列中等待重试", roomId, e); - throw e; // 重新抛出，让外层处理 + } finally { + // 释放锁 + distributedLockUtil.unlock(lockKey); } - - return null; - - }, "火箭能量刷新操作中"); - - } catch (Exception e) { - log.error("刷新房间{}火箭能量失败", roomId, e); + + } catch (Exception e) { + errorCount++; + 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 values = redisTemplate.opsForList().range(queueKey, 0, MAX_BATCH_SIZE - 1); + + if (values == null || values.isEmpty()) { + return false; + } + + // 解析数据 + List 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 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( - cmd.getUserId(), - cmd.getGiftGoldValue() - ); + rocketStatus.addEnergy(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={}, 批次数量={}, 总能量={}, 当前能量={}/{}", - roomId, batch.size(), totalEnergy, + log.info("批量增加火箭能量: roomId={}, 批次数量={}, 总能量={}, 更新前={}, 更新后={}/{}", + 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() { - // 清理buffer中的空队列 - energyBuffer.entrySet().removeIf(entry -> entry.getValue().isEmpty()); - - // 清理推送时间 - lastPushTime.keySet().removeIf(roomId -> !energyBuffer.containsKey(roomId)); - - log.info("清理空闲房间资源, 当前活跃房间数={}", energyBuffer.size()); + public void cleanupExpiredData() { + try { + // 清理本地推送时间缓存 + Set activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY); + if (activeRooms != null) { + Set activeRoomIds = new HashSet<>(); + for (String roomIdStr : activeRooms) { + 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 getQueueStatus() { + Map status = new HashMap<>(); + + Set 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 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; + } +} \ No newline at end of file diff --git a/rc-service/rc-service-other/other-domain/src/main/java/com/red/circle/other/domain/rocket/RocketHistory.java b/rc-service/rc-service-other/other-domain/src/main/java/com/red/circle/other/domain/rocket/RocketHistory.java index 54181034..94b30e56 100644 --- a/rc-service/rc-service-other/other-domain/src/main/java/com/red/circle/other/domain/rocket/RocketHistory.java +++ b/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)); // 转换贡献者 diff --git a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/key/RocketKeys.java b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/key/RocketKeys.java index 0391d665..9ae8c7d4 100644 --- a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/key/RocketKeys.java +++ b/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}