hy/aslan-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

火箭能量更新roomKey处理

Browse Source
This commit is contained in:
tianfeng 2025-11-18 10:47:24 +08:00
parent 1be939fd0e
commit ac034f9cc4
1 changed files with 19 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -15,6 +15,7 @@ import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManager
import com.red.circle.other.infra.gateway.RocketStatusGatewayImpl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@ -43,6 +44,8 @@ public class RocketEnergyAggregator {
private final RocketLaunchManager rocketLaunchManager;
private final RoomProfileManagerService roomProfileManagerService;
private final StringRedisTemplate redisTemplate;
@Value("${spring.profiles.active}")
private String activeProfile;
/**
* Redis 队列 Key 前缀
@ -79,20 +82,29 @@ public class RocketEnergyAggregator {
*/
private final Map<Long, Long> lastPushTime = new ConcurrentHashMap<>();
// 根据环境动态生成 Key 前缀
private String getKey(String roomId) {
return activeProfile + ":" + RocketEnergyAggregator.ENERGY_QUEUE_PREFIX + roomId;
}
private String getActiveRoomsKey() {
return activeProfile + ":" + RocketEnergyAggregator.ACTIVE_ROOMS_KEY;
}
/**
* 接收能量增加请求（写入Redis队列）
*/
public void addEnergy(RocketEnergyAddCmd cmd) {
try {
Long roomId = cmd.getRoomId();
String queueKey = ENERGY_QUEUE_PREFIX + roomId;
String queueKey = getKey(String.valueOf(roomId)) ;
String value = JSON.toJSONString(cmd);
// 写入队列
redisTemplate.opsForList().rightPush(queueKey, value);
// 记录活跃房间（使用 Set 去重）
redisTemplate.opsForSet().add(ACTIVE_ROOMS_KEY, roomId.toString());
redisTemplate.opsForSet().add(getActiveRoomsKey(), roomId.toString());
// 设置队列过期时间（24小时）
redisTemplate.expire(queueKey, 24, TimeUnit.HOURS);
@ -113,7 +125,7 @@ public class RocketEnergyAggregator {
public void flushToDatabase() {
try {
// 获取所有活跃房间
Set<String> activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY);
Set<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
if (activeRooms == null || activeRooms.isEmpty()) {
log.debug("没有待处理的房间");
@ -172,14 +184,14 @@ public class RocketEnergyAggregator {
* @return 是否处理了数据
*/
private boolean processRoomEnergyFromRedis(Long roomId) {
String queueKey = ENERGY_QUEUE_PREFIX + roomId;
String queueKey = getKey(String.valueOf(roomId)) ;
try {
// 检查队列长度
Long queueSize = redisTemplate.opsForList().size(queueKey);
if (queueSize == null || queueSize == 0) {
// 队列为空，从活跃房间列表中移除
redisTemplate.opsForSet().remove(ACTIVE_ROOMS_KEY, roomId.toString());
redisTemplate.opsForSet().remove(getActiveRoomsKey(), roomId.toString());
log.debug("房间{}队列为空，从活跃列表移除", roomId);
return false;
}
@ -338,7 +350,7 @@ public class RocketEnergyAggregator {
public void cleanupExpiredData() {
try {
// 清理本地推送时间缓存
Set<String> activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY);
Set<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
if (activeRooms != null) {
Set<Long> activeRoomIds = new HashSet<>();
for (String roomIdStr : activeRooms) {
@ -385,7 +397,7 @@ public class RocketEnergyAggregator {
public Map<String, Object> getQueueStatus() {
Map<String, Object> status = new HashMap<>();
Set<String> activeRooms = redisTemplate.opsForSet().members(ACTIVE_ROOMS_KEY);
Set<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
status.put("activeRoomCount", activeRooms != null ? activeRooms.size() : 0);
status.put("localPushCacheSize", lastPushTime.size());