From d1e1a07016034ff465b1fd1e56c801723e6cad32 Mon Sep 17 00:00:00 2001 From: tianfeng <769204422@qq.com> Date: Thu, 13 Nov 2025 10:49:27 +0800 Subject: [PATCH] =?UTF-8?q?=E7=81=AB=E7=AE=AD=E8=83=BD=E9=87=8F=E6=9B=B4?= =?UTF-8?q?=E6=96=B0=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../app/manager/RocketEnergyAggregator.java | 216 ++++++++---------- 1 file changed, 94 insertions(+), 122 deletions(-) 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 b37cd635..3e6a6963 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 @@ -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> energyBuffer = new ConcurrentHashMap<>(); + private final Map> energyBuffer = new ConcurrentHashMap<>(); /** - * 房间级别的细粒度锁 + * 上次推送时间 * Key: roomId - * Value: 该房间的锁 + * Value: 上次推送的时间戳 */ - private final Map roomLocks = new ConcurrentHashMap<>(); + private final Map lastPushTime = new ConcurrentHashMap<>(); /** * 推送间隔（毫秒） @@ -74,21 +71,14 @@ public class RocketEnergyAggregator { private static final long FLUSH_INTERVAL = 5000L; // 5秒刷新一次 /** - * 上次推送时间 - * Key: roomId - * Value: 上次推送的时间戳 - */ - private final Map 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={}", - cmd.getRoomId(), cmd.getUserId(), cmd.getGiftGoldValue(), - energyBuffer.get(cmd.getRoomId()).size()); + log.debug("缓存火箭能量: roomId={}, userId={}, energy={}", + cmd.getRoomId(), cmd.getUserId(), cmd.getGiftGoldValue()); } /** @@ -103,30 +93,33 @@ public class RocketEnergyAggregator { log.info("开始批量刷新火箭能量, 待处理房间数={}", energyBuffer.size()); // 遍历所有房间 - energyBuffer.forEach((roomId, cmdList) -> { - if (cmdList.isEmpty()) { + energyBuffer.forEach((roomId, queue) -> { + if (queue.isEmpty()) { return; } try { - // 获取该房间的细粒度锁 - ReentrantLock roomLock = roomLocks.computeIfAbsent(roomId, k -> new ReentrantLock()); + String lockKey = RocketKeys.LOCK.getKey(roomId); - if (roomLock.tryLock()) { - try { - // 取出当前批次的数据 - List batch = new ArrayList<>(cmdList); - cmdList.clear(); - - // 处理这个房间的能量增加 - processRoomEnergy(roomId, batch); - - } finally { - roomLock.unlock(); + distributedLockUtil.executeWithLock(lockKey, 10L, () -> { + + // 原子性地取出所有数据 + List batch = new ArrayList<>(); + RocketEnergyAddCmd cmd; + while ((cmd = queue.poll()) != null) { + batch.add(cmd); } - } 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 cmdList) { - if (cmdList.isEmpty()) { + 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; + } + + // 使用配置初始化 + 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); - - // 使用Redis分布式锁，避免多实例冲突 - distributedLockUtil.executeWithLock(lockKey, 10L, () -> { + // 3. 批量增加能量 + long totalEnergy = 0L; + for (RocketEnergyAddCmd cmd : batch) { + 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(); - 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; - }, "火箭能量刷新操作中，请稍后"); + // 异步触发发射 + rocketLaunchManager.launchAsync(roomId); + } } /** * 推送能量更新（节流：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 { - // 查询火箭状态 - RocketStatus fullStatus = rocketStatusGatewayImpl - .findByRoomIdAndDateWithoutContributors(roomId, rocketStatus.getDate()); + String roomAccount = roomProfileManagerService.getRoomAccount(roomId); + RocketStatusCO statusCO = RocketConvertor.toStatusCO(rocketStatus); + rocketImPushManager.pushEnergyUpdate(roomAccount, statusCO); - if (fullStatus != null) { - RocketStatusCO statusCO = RocketConvertor.toStatusCO(fullStatus); - rocketImPushManager.pushEnergyUpdate(roomAccount, statusCO); - - // 更新推送时间 - lastPushTime.put(roomId, now); - - log.info("推送火箭能量更新: roomId={}, energy={}/{}", - roomId, fullStatus.getCurrentEnergy(), fullStatus.getMaxEnergy()); - } + // 更新推送时间 + lastPushTime.put(roomId, now); + + log.info("推送火箭能量更新: roomId={}, energy={}/{}", + roomId, rocketStatus.getCurrentEnergy(), rocketStatus.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));