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

fix prod runtime logging and room discovery

Browse Source
This commit is contained in:
hy001 2026-04-29 14:47:46 +08:00
parent 096e37612e
commit e2e59eb1b2
6 changed files with 277 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
nacos_config/common/mongo-config.yml
View File

@ -2,8 +2,3 @@ spring:
data:
mongodb:
uri: ${LIKEI_MONGO_URI}
logging:
level:
org.springframework.data.mongodb.core.MongoTemplate: DEBUG

159
rc-gateway/src/main/java/com/red/circle/gateway/filter/ApiLoggingFilter.java
View File

@ -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<HttpMessageReader<?>> 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<HttpMessageReader<?>> messageReaders = HandlerStrategies.withDefaults()
.messageReaders();
@Override
public Mono<Void> 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<? extends DataBuffer> 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,

8
rc-gateway/src/main/java/com/red/circle/gateway/logging/LoggingUploadService.java
View File

@ -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";

171
rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java
View File

@ -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<RoomVoiceProfileCO> 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<RoomVoiceProfileCO> roomList = JSON.parseArray(regionsRoom, RoomVoiceProfileCO.class);
if (CollectionUtils.isEmpty(roomList)) {
logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, true, 0);
return CollectionUtils.newArrayList();
}
List<RoomVoiceProfileCO> 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<ActiveVoiceRoom> 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<ActiveVoiceRoom> emptyRooms = getEmptyRoomsFromCache();
stepStart = logDiscoveryStep("scan-empty-room-cache", stepStart, reqUserId, sysOrigin, region);
if (CollectionUtils.isNotEmpty(emptyRooms)) {
List<ActiveVoiceRoom> 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<RoomVoiceProfileCO> roomList = roomVoiceProfileCommon.toListRoomVoiceProfileCO(activeVoiceRooms);
stepStart = logDiscoveryStep("build-profile-list", stepStart, reqUserId, sysOrigin, region);
if (roomList.size() < DISCOVERY_LIMIT) {
List<RoomProfileManager> 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<Long, SVIPLevelEnum> svipLevelEnumMap = userSVipGateway.checkSVipIdentity(
roomList.stream().map(RoomVoiceProfileCO::getUserId).collect(
Collectors.toSet()));
}
Map<Long, SVIPLevelEnum> 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<RoomVoiceProfileCO> 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<RoomVoiceProfileCO> mergeRoomList(List<RoomVoiceProfileCO> currentRooms,
List<RoomVoiceProfileCO> fallbackRooms, int limit) {
if (CollectionUtils.isEmpty(fallbackRooms)) {
@ -348,12 +404,12 @@ public class RoomVoiceDiscoverQryExe {
*
* @return
*/
private List<ActiveVoiceRoom> getEmptyRoomsFromCache() {
try {
Set<String> keys = redisTemplate.keys(EMPTY_ROOM_CACHE_KEY);
if (CollectionUtils.isEmpty(keys)) {
return new ArrayList<>();
}
private List<ActiveVoiceRoom> getEmptyRoomsFromCache() {
try {
Set<String> keys = scanEmptyRoomKeys();
if (CollectionUtils.isEmpty(keys)) {
return new ArrayList<>();
}
List<ActiveVoiceRoom> 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<String> scanEmptyRoomKeys() {
Set<String> 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

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

@ -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<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
@Scheduled(fixedDelay = FLUSH_INTERVAL)
public void flushToDatabase() {
if (!rocketEnergyEnabled) {
return;
}
try {
// 获取所有活跃房间
Set<String> 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<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
@Scheduled(fixedRate = 3600000)
public void cleanupExpiredData() {
if (!rocketEnergyEnabled) {
return;
}
try {
// 清理本地推送时间缓存
Set<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
if (activeRooms != null) {
Set<Long> 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<String, Object> getQueueStatus() {
Map<String, Object> status = new HashMap<>();
Set<String> activeRooms = redisTemplate.opsForSet().members(getActiveRoomsKey());
public Map<String, Object> getQueueStatus() {
Map<String, Object> status = new HashMap<>();
status.put("enabled", rocketEnergyEnabled);
if (!rocketEnergyEnabled) {
return status;
}
Set<String> 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;
}
}
}

2
rc-service/rc-service-wallet/wallet-infrastructure/src/main/java/com/red/circle/wallet/infra/enums/LogTopic.java
View File

@ -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;