diff --git a/nacos_config/common/mongo-config.yml b/nacos_config/common/mongo-config.yml index 05e7d637..8459e530 100644 --- a/nacos_config/common/mongo-config.yml +++ b/nacos_config/common/mongo-config.yml @@ -2,8 +2,3 @@ spring: data: mongodb: uri: ${LIKEI_MONGO_URI} - - -logging: - level: - org.springframework.data.mongodb.core.MongoTemplate: DEBUG diff --git a/rc-gateway/src/main/java/com/red/circle/gateway/filter/ApiLoggingFilter.java b/rc-gateway/src/main/java/com/red/circle/gateway/filter/ApiLoggingFilter.java index 4716da43..27bed08a 100644 --- a/rc-gateway/src/main/java/com/red/circle/gateway/filter/ApiLoggingFilter.java +++ b/rc-gateway/src/main/java/com/red/circle/gateway/filter/ApiLoggingFilter.java @@ -17,10 +17,9 @@ import org.springframework.cloud.gateway.filter.factory.rewrite.CachedBodyOutput import org.springframework.cloud.gateway.support.BodyInserterContext; import org.springframework.cloud.gateway.support.ServerWebExchangeUtils; import org.springframework.core.Ordered; -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferFactory; -import org.springframework.core.io.buffer.DataBufferUtils; -import org.springframework.core.io.buffer.DefaultDataBufferFactory; +import org.springframework.core.io.buffer.DataBuffer; +import org.springframework.core.io.buffer.DataBufferFactory; +import org.springframework.core.io.buffer.DataBufferUtils; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatusCode; import org.springframework.http.MediaType; @@ -36,14 +35,15 @@ import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.server.HandlerStrategies; import org.springframework.web.reactive.function.server.ServerRequest; import org.springframework.web.server.ServerWebExchange; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.util.List; -import java.util.Objects; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Objects; import java.util.Optional; import java.util.zip.GZIPInputStream; @@ -55,11 +55,13 @@ import java.util.zip.GZIPInputStream; @Slf4j @Component @RequiredArgsConstructor -public class ApiLoggingFilter implements GlobalFilter, Ordered { - - private final LoggingUploadService loggingUploadService; - private final List> messageReaders = HandlerStrategies.withDefaults() - .messageReaders(); +public class ApiLoggingFilter implements GlobalFilter, Ordered { + + private static final int MAX_RESPONSE_BODY_LOG_LENGTH = 500; + + private final LoggingUploadService loggingUploadService; + private final List> messageReaders = HandlerStrategies.withDefaults() + .messageReaders(); @Override public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { @@ -160,56 +162,79 @@ public class ApiLoggingFilter implements GlobalFilter, Ordered { GatewayLoggingResponse loggingResponse = gatewayLog.getResponse(); loggingResponse.setTime(System.currentTimeMillis()); gatewayLog.calculateExecuteTime(); - loggingResponse.setCode(Optional.ofNullable(response.getStatusCode()) - .map(HttpStatusCode::value) - .map(Objects::toString) - .orElse("?")); - if (body instanceof Flux) { - // 获取响应类型，如果是 json 就打印 - String originalResponseContentType = exchange.getAttribute(ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR); - - if (StringUtils.isNotBlank(originalResponseContentType) && originalResponseContentType.contains("application/json")) { - - Flux fluxBody = Flux.from(body); - return super.writeWith(fluxBody.buffer().map(dataBuffers -> { - - // 合并多个流集合，解决返回体分段传输 - DataBufferFactory dataBufferFactory = new DefaultDataBufferFactory(); - DataBuffer join = dataBufferFactory.join(dataBuffers); - byte[] content = new byte[join.readableByteCount()]; - join.read(content); - - // 释放掉内存 - DataBufferUtils.release(join); - - try (ByteArrayOutputStream out = new ByteArrayOutputStream()) { - try (GZIPInputStream in = new GZIPInputStream(new java.io.ByteArrayInputStream(content))) { - byte[] buffer = new byte[1024]; - int len; - while ((len = in.read(buffer)) > 0) { - out.write(buffer, 0, len); - } - } - String responseResult = out.toString(StandardCharsets.UTF_8); - // 截取500个字符 - responseResult = responseResult.substring(0, Math.min(responseResult.length(), 500)); - loggingResponse.setBody(responseResult); - } catch (IOException e) { - log.error("Failed to decompress data", e); - } - - return bufferFactory.wrap(content); - })); - } - } - return super.writeWith(body); - } - }; - } - - /** - * 请求装饰器，重新计算 headers. - */ + loggingResponse.setCode(Optional.ofNullable(response.getStatusCode()) + .map(HttpStatusCode::value) + .map(Objects::toString) + .orElse("?")); + if (body instanceof Flux && !isSuccessResponse(response)) { + // 获取响应类型，如果是 json 就打印 + if (isJsonResponse(exchange, response)) { + return super.writeWith(DataBufferUtils.join(Flux.from(body)).map(dataBuffer -> { + byte[] content = new byte[dataBuffer.readableByteCount()]; + dataBuffer.read(content); + DataBufferUtils.release(dataBuffer); + + String responseResult = readResponseBody(exchange, content); + responseResult = responseResult.substring(0, Math.min(responseResult.length(), MAX_RESPONSE_BODY_LOG_LENGTH)); + loggingResponse.setBody(responseResult); + return bufferFactory.wrap(content); + })); + } + } + return super.writeWith(body); + } + }; + } + + private boolean isSuccessResponse(ServerHttpResponse response) { + HttpStatusCode statusCode = response.getStatusCode(); + return statusCode == null || statusCode.is2xxSuccessful(); + } + + private boolean isJsonResponse(ServerWebExchange exchange, ServerHttpResponse response) { + String originalResponseContentType = exchange.getAttribute(ServerWebExchangeUtils.ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR); + if (StringUtils.isNotBlank(originalResponseContentType) && originalResponseContentType.contains(MediaType.APPLICATION_JSON_VALUE)) { + return true; + } + MediaType contentType = response.getHeaders().getContentType(); + return contentType != null && MediaType.APPLICATION_JSON.isCompatibleWith(contentType); + } + + private String readResponseBody(ServerWebExchange exchange, byte[] content) { + if (!isGzipResponse(exchange, content)) { + return new String(content, StandardCharsets.UTF_8); + } + try (ByteArrayOutputStream out = new ByteArrayOutputStream(); + GZIPInputStream in = new GZIPInputStream(new ByteArrayInputStream(content))) { + byte[] buffer = new byte[1024]; + int len; + while ((len = in.read(buffer)) > 0) { + out.write(buffer, 0, len); + } + return out.toString(StandardCharsets.UTF_8); + } catch (IOException e) { + log.warn("Failed to read gzip response body for logging: {}", e.getMessage()); + return ""; + } + } + + private boolean isGzipResponse(ServerWebExchange exchange, byte[] content) { + boolean gzipHeader = exchange.getResponse().getHeaders().getOrEmpty(HttpHeaders.CONTENT_ENCODING) + .stream() + .anyMatch(value -> StringUtils.isNotBlank(value) && value.toLowerCase().contains("gzip")); + return gzipHeader || hasGzipMagic(content); + } + + private boolean hasGzipMagic(byte[] content) { + return content != null + && content.length >= 2 + && (content[0] & 0xff) == 0x1f + && (content[1] & 0xff) == 0x8b; + } + + /** + * 请求装饰器，重新计算 headers. + */ @SuppressWarnings("all") private ServerHttpRequestDecorator requestDecorate(ServerWebExchange exchange, HttpHeaders headers, diff --git a/rc-gateway/src/main/java/com/red/circle/gateway/logging/LoggingUploadService.java b/rc-gateway/src/main/java/com/red/circle/gateway/logging/LoggingUploadService.java index 7c911c24..fa65a673 100644 --- a/rc-gateway/src/main/java/com/red/circle/gateway/logging/LoggingUploadService.java +++ b/rc-gateway/src/main/java/com/red/circle/gateway/logging/LoggingUploadService.java @@ -22,10 +22,10 @@ public class LoggingUploadService { private final ClsService clsService; private String getTopic() { - log.info("profileActive: {}", profileActive); - if (Objects.equals(profileActive, "prod")) { - return "4d467d31-9165-42c6-b80a-2ea23ae7777e"; - } + log.info("profileActive: {}", profileActive); + if (Objects.equals(profileActive, "prod")) { + return "4fd8ef93-1c0c-4802-a999-9268831a3ee5"; + } // if (Objects.equals(profileActive, "dev")) { // return "180afe72-b0fe-4502-bbc2-55a8a5193e78"; diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java index 09cd5657..6378c7c8 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java +++ b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java @@ -2,7 +2,6 @@ package com.red.circle.other.app.command.room.query; import com.alibaba.fastjson.JSON; import com.google.common.collect.Sets; -import com.google.gson.Gson; import com.red.circle.common.business.dto.cmd.AppExtCommand; import com.red.circle.common.business.enums.SVIPLevelEnum; import com.red.circle.live.inner.endpoint.LiveMicClient; @@ -28,20 +27,18 @@ import com.red.circle.other.infra.database.mongo.service.live.RoomUserBlacklistS import com.red.circle.other.infra.database.rds.entity.sys.SysCountryCode; import com.red.circle.other.infra.database.rds.service.sys.SysCountryCodeService; import com.red.circle.other.infra.database.rds.service.user.device.LatestMobileDeviceService; -import com.red.circle.other.inner.model.cmd.live.AppActiveVoiceRoomQryCmd; -import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO; -import com.red.circle.tool.core.collection.CollectionUtils; -import com.google.gson.reflect.TypeToken; -import java.lang.reflect.Type; - -import java.util.*; -import java.util.function.Function; -import java.util.stream.Collectors; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.BeanUtils; -import org.springframework.data.redis.core.RedisTemplate; -import org.springframework.stereotype.Component; +import com.red.circle.other.inner.model.cmd.live.AppActiveVoiceRoomQryCmd; +import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO; +import com.red.circle.tool.core.collection.CollectionUtils; + +import java.util.*; +import java.util.stream.Collectors; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.BeanUtils; +import org.springframework.data.redis.core.ScanOptions; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.stereotype.Component; /** * 发现房间集合. @@ -72,18 +69,29 @@ public class RoomVoiceDiscoverQryExe { private static final String EMPTY_ROOM_CACHE_KEY = "empty_room_cache:*"; private static final String MEMBER_QUANTITY_EXT_KEY = "memberQuantity"; private static final int DISCOVERY_LIMIT = 50; + private static final long DISCOVERY_SLOW_STEP_MILLIS = 100L; + private static final long DISCOVERY_SLOW_TOTAL_MILLIS = 300L; + private static final long EMPTY_ROOM_SCAN_COUNT = 200L; public List execute(AppExtCommand cmd) { - boolean allRegionWhiteListUser = roomRegionFilterCommon.canViewAllRegions(cmd.requiredReqUserId()); - RegionConfig currentRegion = roomRegionFilterCommon.requireRegionConfig(cmd.requiredReqUserId()); - String region = currentRegion.getRegionCode(); - UserProfile userProfile = userProfileGateway.getByUserId(cmd.requiredReqUserId()); + long totalStart = System.currentTimeMillis(); + long stepStart = totalStart; + Long reqUserId = cmd.requiredReqUserId(); + String sysOrigin = cmd.requireReqSysOrigin(); - String key = discoveryCacheKey(cmd.requireReqSysOrigin(), region, allRegionWhiteListUser); + boolean allRegionWhiteListUser = roomRegionFilterCommon.canViewAllRegions(reqUserId); + RegionConfig currentRegion = roomRegionFilterCommon.requireRegionConfig(reqUserId); + String region = currentRegion.getRegionCode(); + UserProfile userProfile = userProfileGateway.getByUserId(reqUserId); + stepStart = logDiscoveryStep("prepare", stepStart, reqUserId, sysOrigin, region); + + String key = discoveryCacheKey(sysOrigin, region, allRegionWhiteListUser); String regionsRoom = regionRoomCacheService.getRegionsRoom(key); + stepStart = logDiscoveryStep("read-region-cache", stepStart, reqUserId, sysOrigin, region); if (Objects.nonNull(regionsRoom)) { List roomList = JSON.parseArray(regionsRoom, RoomVoiceProfileCO.class); if (CollectionUtils.isEmpty(roomList)) { + logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, true, 0); return CollectionUtils.newArrayList(); } List availableRooms = roomVoiceProfileCommon @@ -91,32 +99,36 @@ public class RoomVoiceDiscoverQryExe { if (availableRooms.size() != roomList.size()) { regionRoomCacheService.remove(key); } + stepStart = logDiscoveryStep("filter-cache", stepStart, reqUserId, sysOrigin, region); + logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, true, availableRooms.size()); return availableRooms; } List activeVoiceRooms = allRegionWhiteListUser ? activeVoiceRoomService.listDiscover( - cmd.requireReqSysOrigin(), + sysOrigin, true, region, Optional.ofNullable(userProfile).map(UserProfile::getCountryCode).orElse(null), DISCOVERY_LIMIT ) : roomRegionFilterCommon.filterActiveVoiceRoomsByRegion( - activeVoiceRoomService.listByRegion(cmd.requireReqSysOrigin(), region, DISCOVERY_LIMIT), + activeVoiceRoomService.listByRegion(sysOrigin, region, DISCOVERY_LIMIT), region ); + stepStart = logDiscoveryStep("query-active-rooms", stepStart, reqUserId, sysOrigin, region); // 查询没人的房间缓存，加入进去 List emptyRooms = getEmptyRoomsFromCache(); + stepStart = logDiscoveryStep("scan-empty-room-cache", stepStart, reqUserId, sysOrigin, region); if (CollectionUtils.isNotEmpty(emptyRooms)) { List matchedEmptyRooms = allRegionWhiteListUser ? emptyRooms.stream() - .filter(room -> Objects.equals(cmd.requireReqSysOrigin(), room.getSysOrigin())) + .filter(room -> Objects.equals(sysOrigin, room.getSysOrigin())) .toList() : roomRegionFilterCommon.filterActiveVoiceRoomsByRegion( emptyRooms.stream() - .filter(room -> Objects.equals(cmd.requireReqSysOrigin(), room.getSysOrigin())) + .filter(room -> Objects.equals(sysOrigin, room.getSysOrigin())) .toList(), region ); @@ -131,10 +143,13 @@ public class RoomVoiceDiscoverQryExe { activeVoiceRooms = allRooms; } } + stepStart = logDiscoveryStep("merge-empty-rooms", stepStart, reqUserId, sysOrigin, region); + List roomList = roomVoiceProfileCommon.toListRoomVoiceProfileCO(activeVoiceRooms); + stepStart = logDiscoveryStep("build-profile-list", stepStart, reqUserId, sysOrigin, region); if (roomList.size() < DISCOVERY_LIMIT) { List fallbackRoomManagersSource = roomProfileManagerService.listSelectiveLimitByCountryCodes( - cmd.requireReqSysOrigin(), + sysOrigin, allRegionWhiteListUser ? List.of() : roomRegionFilterCommon.listCountryCodes(currentRegion), roomList.stream().map(RoomVoiceProfileCO::getId).toList(), DISCOVERY_LIMIT @@ -148,24 +163,33 @@ public class RoomVoiceDiscoverQryExe { roomList = mergeRoomList(roomList, roomVoiceProfileCommon.listRoomVoiceProfilesV2(fallbackRoomManagers), DISCOVERY_LIMIT); } + stepStart = logDiscoveryStep("fallback-room-profiles", stepStart, reqUserId, sysOrigin, region); roomList = sortByMemberQuantityDesc(roomList); + stepStart = logDiscoveryStep("sort-rooms", stepStart, reqUserId, sysOrigin, region); if (CollectionUtils.isEmpty(roomList)) { + logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, 0); return CollectionUtils.newArrayList(); - } - Map svipLevelEnumMap = userSVipGateway.checkSVipIdentity( - roomList.stream().map(RoomVoiceProfileCO::getUserId).collect( - Collectors.toSet())); + } + Map svipLevelEnumMap = userSVipGateway.checkSVipIdentity( + roomList.stream().map(RoomVoiceProfileCO::getUserId).collect( + Collectors.toSet())); roomList = roomList.stream() .peek(room -> { - room.setUserSVipLevel(svipLevelEnumMap.get(room.getUserId())); - }) - .toList(); - - // 填充火箭状态 - fillRocketStatus(roomList); - // 填充游戏图标 - fillGameIcon(roomList); + room.setUserSVipLevel(svipLevelEnumMap.get(room.getUserId())); + }) + .toList(); + stepStart = logDiscoveryStep("fill-svip", stepStart, reqUserId, sysOrigin, region); + + // 填充火箭状态 + fillRocketStatus(roomList); + stepStart = logDiscoveryStep("fill-rocket-status", stepStart, reqUserId, sysOrigin, region); + // 填充游戏图标 + fillGameIcon(roomList); + stepStart = logDiscoveryStep("fill-game-icon", stepStart, reqUserId, sysOrigin, region); + saveDiscoveryCache(key, roomList, reqUserId, sysOrigin, region); + logDiscoveryStep("write-region-cache", stepStart, reqUserId, sysOrigin, region); + logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, roomList.size()); return roomList; } @@ -201,6 +225,38 @@ public class RoomVoiceDiscoverQryExe { return allRegionWhiteListUser ? sysOrigin + region + ":ALL_REGION" : sysOrigin + region; } + private void saveDiscoveryCache(String key, List roomList, Long userId, + String sysOrigin, String region) { + if (CollectionUtils.isEmpty(roomList)) { + return; + } + try { + regionRoomCacheService.saveRegionsRoom(JSON.toJSONString(roomList), key); + } catch (Exception e) { + log.warn("写入房间发现缓存失败, key={}, userId={}, sysOrigin={}, region={}", + key, userId, sysOrigin, region, e); + } + } + + private long logDiscoveryStep(String step, long stepStart, Long userId, String sysOrigin, String region) { + long now = System.currentTimeMillis(); + long cost = now - stepStart; + if (cost >= DISCOVERY_SLOW_STEP_MILLIS) { + log.warn("房间发现慢步骤, step={}, cost={}ms, userId={}, sysOrigin={}, region={}", + step, cost, userId, sysOrigin, region); + } + return now; + } + + private void logDiscoveryTotal(long totalStart, Long userId, String sysOrigin, String region, + boolean allRegionWhiteListUser, boolean cacheHit, int roomCount) { + long cost = System.currentTimeMillis() - totalStart; + if (cost >= DISCOVERY_SLOW_TOTAL_MILLIS) { + log.warn("房间发现总耗时慢, cost={}ms, userId={}, sysOrigin={}, region={}, allRegion={}, cacheHit={}, roomCount={}", + cost, userId, sysOrigin, region, allRegionWhiteListUser, cacheHit, roomCount); + } + } + private List mergeRoomList(List currentRooms, List fallbackRooms, int limit) { if (CollectionUtils.isEmpty(fallbackRooms)) { @@ -348,12 +404,12 @@ public class RoomVoiceDiscoverQryExe { * * @return */ - private List getEmptyRoomsFromCache() { - try { - Set keys = redisTemplate.keys(EMPTY_ROOM_CACHE_KEY); - if (CollectionUtils.isEmpty(keys)) { - return new ArrayList<>(); - } + private List getEmptyRoomsFromCache() { + try { + Set keys = scanEmptyRoomKeys(); + if (CollectionUtils.isEmpty(keys)) { + return new ArrayList<>(); + } List emptyRooms = new ArrayList<>(); for (String key : keys) { @@ -376,12 +432,29 @@ public class RoomVoiceDiscoverQryExe { return emptyRooms; } catch (Exception e) { log.error("获取空房间缓存失败", e); - return new ArrayList<>(); - } - } - - /** - * 将ActiveVoiceRoomCO转换为ActiveVoiceRoom + return new ArrayList<>(); + } + } + + private Set scanEmptyRoomKeys() { + Set keys = new HashSet<>(); + ScanOptions options = ScanOptions.scanOptions() + .match(EMPTY_ROOM_CACHE_KEY) + .count(EMPTY_ROOM_SCAN_COUNT) + .build(); + try (var cursor = redisTemplate.scan(options)) { + if (Objects.isNull(cursor)) { + return keys; + } + while (cursor.hasNext()) { + keys.add(Objects.toString(cursor.next(), "")); + } + } + return keys; + } + + /** + * 将ActiveVoiceRoomCO转换为ActiveVoiceRoom * * @param roomCO * @return 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 9c983575..38fa19d8 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 @@ -44,9 +44,11 @@ public class RocketEnergyAggregator { private final RocketImPushManager rocketImPushManager; private final RocketLaunchManager rocketLaunchManager; private final RoomProfileManagerService roomProfileManagerService; - private final StringRedisTemplate redisTemplate; - @Value("${spring.profiles.active}") - private String activeProfile; + private final StringRedisTemplate redisTemplate; + @Value("${spring.profiles.active}") + private String activeProfile; + @Value("${rocket.energy.enabled:false}") + private boolean rocketEnergyEnabled; /** * Redis 队列 Key 前缀 @@ -92,13 +94,16 @@ public class RocketEnergyAggregator { return activeProfile + ":" + RocketEnergyAggregator.ACTIVE_ROOMS_KEY; } - /** - * 接收能量增加请求（写入Redis队列） - */ - public void addEnergy(RocketEnergyAddCmd cmd) { - try { - Long roomId = cmd.getRoomId(); - String queueKey = getKey(String.valueOf(roomId)) ; + /** + * 接收能量增加请求（写入Redis队列） + */ + public void addEnergy(RocketEnergyAddCmd cmd) { + if (!rocketEnergyEnabled) { + return; + } + try { + Long roomId = cmd.getRoomId(); + String queueKey = getKey(String.valueOf(roomId)) ; String value = JSON.toJSONString(cmd); // 写入队列 @@ -122,11 +127,14 @@ public class RocketEnergyAggregator { /** * 定时批量刷新 */ - @Scheduled(fixedDelay = FLUSH_INTERVAL) - public void flushToDatabase() { - try { - // 获取所有活跃房间 - Set activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey()); + @Scheduled(fixedDelay = FLUSH_INTERVAL) + public void flushToDatabase() { + if (!rocketEnergyEnabled) { + return; + } + try { + // 获取所有活跃房间 + Set activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey()); if (activeRooms == null || activeRooms.isEmpty()) { log.debug("没有待处理的房间"); @@ -353,11 +361,14 @@ public class RocketEnergyAggregator { /** * 定时清理过期数据（每小时） */ - @Scheduled(fixedRate = 3600000) - public void cleanupExpiredData() { - try { - // 清理本地推送时间缓存 - Set activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey()); + @Scheduled(fixedRate = 3600000) + public void cleanupExpiredData() { + if (!rocketEnergyEnabled) { + return; + } + try { + // 清理本地推送时间缓存 + Set activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey()); if (activeRooms != null) { Set activeRoomIds = new HashSet<>(); for (String roomIdStr : activeRooms) { @@ -380,31 +391,42 @@ public class RocketEnergyAggregator { /** * 应用关闭时强制刷新 */ - @javax.annotation.PreDestroy - public void shutdown() { - log.info("应用关闭，强制刷新所有待处理能量"); - try { - flushToDatabase(); + @javax.annotation.PreDestroy + public void shutdown() { + if (!rocketEnergyEnabled) { + return; + } + log.info("应用关闭，强制刷新所有待处理能量"); + try { + flushToDatabase(); } catch (Exception e) { log.error("关闭时刷新失败", e); } } - /** - * 手动触发刷新（供运维使用） - */ - public void manualFlush() { - log.info("手动触发刷新"); - flushToDatabase(); + /** + * 手动触发刷新（供运维使用） + */ + public void manualFlush() { + if (!rocketEnergyEnabled) { + log.info("火箭能量聚合未启用，跳过手动刷新"); + return; + } + log.info("手动触发刷新"); + flushToDatabase(); } /** * 获取队列状态（监控使用） */ - public Map getQueueStatus() { - Map status = new HashMap<>(); - - Set activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey()); + public Map getQueueStatus() { + Map status = new HashMap<>(); + status.put("enabled", rocketEnergyEnabled); + if (!rocketEnergyEnabled) { + return status; + } + + Set activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey()); status.put("activeRoomCount", activeRooms != null ? activeRooms.size() : 0); status.put("localPushCacheSize", lastPushTime.size()); @@ -424,4 +446,4 @@ public class RocketEnergyAggregator { return status; } -} \ No newline at end of file +} diff --git a/rc-service/rc-service-wallet/wallet-infrastructure/src/main/java/com/red/circle/wallet/infra/enums/LogTopic.java b/rc-service/rc-service-wallet/wallet-infrastructure/src/main/java/com/red/circle/wallet/infra/enums/LogTopic.java index 91108f6c..3429c812 100644 --- a/rc-service/rc-service-wallet/wallet-infrastructure/src/main/java/com/red/circle/wallet/infra/enums/LogTopic.java +++ b/rc-service/rc-service-wallet/wallet-infrastructure/src/main/java/com/red/circle/wallet/infra/enums/LogTopic.java @@ -8,7 +8,7 @@ public enum LogTopic { /** * 钱包金币资产. */ - WALLET_GOLD_ASSET_PROD("45b22e07-3cc4-41e5-944c-ebda65330e39"), + WALLET_GOLD_ASSET_PROD("656b398e-9713-4de9-8975-b6fac77a3f61"), WALLET_GOLD_ASSET_DEVELOP("931e4d86-fde3-455e-b225-59f83ce30235"); private final String topic;