fix(other): index empty rooms and singleflight discovery rebuild

This commit is contained in:
hy001 2026-07-11 02:33:14 +08:00
parent 0177ca0dce
commit 691ef81b12
13 changed files with 1446 additions and 307 deletions

View File

@ -0,0 +1,35 @@
package com.red.circle.other.inner.model.cache;
/**
* 空房间缓存的跨服务 Redis key 约定
*
* <p>live-service 负责写入空房间快照other-service 负责消费索引生成发现页key 只能在这个
* 共享模型中定义避免两个服务独立拼接后发生不可见的数据分叉</p>
*/
public final class EmptyRoomCacheKeys {
/** 保留旧前缀,使滚动发布期间尚未升级的 other-service 仍可读取房间快照。 */
private static final String PAYLOAD_PREFIX = "empty_room_cache:";
/**
* 索引必须避开 {@code empty_room_cache:*}旧版 other-service 会扫描该通配符并对结果执行
* GET如果索引也使用旧前缀旧实例读到 ZSet 会产生 WRONGTYPE
*/
private static final String INDEX_PREFIX = "empty_room_index:v2:";
private static final String ALL_REGIONS = "__ALL__";
private EmptyRoomCacheKeys() {
}
public static String payload(Long roomId) {
return PAYLOAD_PREFIX + roomId;
}
public static String regionIndex(String sysOrigin, String region) {
return INDEX_PREFIX + sysOrigin + ":" + region;
}
public static String allRegionsIndex(String sysOrigin) {
return INDEX_PREFIX + sysOrigin + ":" + ALL_REGIONS;
}
}

View File

@ -1,12 +1,12 @@
package com.red.circle.live.app.command.user; package com.red.circle.live.app.command.user;
import com.alibaba.fastjson.JSON;
import com.red.circle.common.business.dto.cmd.app.AppRoomIdCmd; import com.red.circle.common.business.dto.cmd.app.AppRoomIdCmd;
import com.red.circle.component.redis.service.RedisService; import com.red.circle.component.redis.service.RedisService;
import com.red.circle.framework.dto.ResultResponse; import com.red.circle.framework.dto.ResultResponse;
import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; import com.red.circle.live.domain.gateway.LiveMicrophoneGateway;
import com.red.circle.live.domain.live.LiveMicrophone; import com.red.circle.live.domain.live.LiveMicrophone;
import com.red.circle.live.domain.live.LiveMusicStatus; import com.red.circle.live.domain.live.LiveMusicStatus;
import com.red.circle.live.infra.database.cache.service.EmptyRoomCacheWriter;
import com.red.circle.live.infra.database.cache.service.LiveMicCacheService; import com.red.circle.live.infra.database.cache.service.LiveMicCacheService;
import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService; import com.red.circle.live.infra.database.cache.service.LiveMusicHeartbeatService;
import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService; import com.red.circle.live.infra.database.cache.service.UserAgoraTokenCacheService;
@ -16,7 +16,6 @@ import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.tool.core.thread.ThreadPoolManager; import com.red.circle.tool.core.thread.ThreadPoolManager;
import java.util.List; import java.util.List;
import java.util.Objects; import java.util.Objects;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
@ -37,9 +36,7 @@ public class LiveQuitRoomExe {
private final ActiveVoiceRoomClient activeVoiceRoomClient; private final ActiveVoiceRoomClient activeVoiceRoomClient;
private final RedisService redisService; private final RedisService redisService;
private final UserAgoraTokenCacheService userAgoraTokenCacheService; private final UserAgoraTokenCacheService userAgoraTokenCacheService;
private final EmptyRoomCacheWriter emptyRoomCacheWriter;
private static final String EMPTY_ROOM_CACHE_KEY = "empty_room_cache:";
private static final int CACHE_EXPIRE_MINUTES = 10;
public void execute(AppRoomIdCmd cmd) { public void execute(AppRoomIdCmd cmd) {
Long roomId = cmd.getRoomId(); Long roomId = cmd.getRoomId();
@ -94,13 +91,8 @@ public class LiveQuitRoomExe {
ResultResponse<ActiveVoiceRoomCO> noBodyRoom = activeVoiceRoomClient.createNoBodyRoom(roomId); ResultResponse<ActiveVoiceRoomCO> noBodyRoom = activeVoiceRoomClient.createNoBodyRoom(roomId);
ActiveVoiceRoomCO body = noBodyRoom.getBody(); ActiveVoiceRoomCO body = noBodyRoom.getBody();
if (Objects.nonNull(body)) { if (Objects.nonNull(body)) {
String cacheKey = EMPTY_ROOM_CACHE_KEY + roomId; // live 同时维护 legacy payload ZSet使它可以先于 other 发布且不影响旧 reader
redisService.setIfAbsent( emptyRoomCacheWriter.save(body);
cacheKey,
JSON.toJSONString(body),
CACHE_EXPIRE_MINUTES,
TimeUnit.MINUTES
);
} }
} catch (Exception e) { } catch (Exception e) {
log.error("quitRoom cacheEmptyRoom error roomId={}, userId={}", roomId, userId, e); log.error("quitRoom cacheEmptyRoom error roomId={}, userId={}", roomId, userId, e);

View File

@ -0,0 +1,16 @@
package com.red.circle.live.infra.database.cache.service;
import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO;
/**
* 维护发现页所需的空房间快照与索引
*/
public interface EmptyRoomCacheWriter {
/**
* 保存空房间快照并把房间登记到所属区域和同系统全区域索引
*
* @param room other-service 生成的完整空房间快照
*/
void save(ActiveVoiceRoomCO room);
}

View File

@ -0,0 +1,112 @@
package com.red.circle.live.infra.database.cache.service.impl;
import com.alibaba.fastjson.JSON;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.live.infra.database.cache.service.EmptyRoomCacheWriter;
import com.red.circle.other.inner.model.cache.EmptyRoomCacheKeys;
import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO;
import com.red.circle.tool.core.text.StringUtils;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* 使用单房间快照 + ZSet 到期索引维护空房间
*
* <p>ZSet score 是快照实际过期毫秒这样 Set/Hash 不具备的成员级 TTL 可以在读取时通过
* ZREMRANGEBYSCORE 有界清理同时发现页不再遍历整个 Redis DB</p>
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class EmptyRoomCacheWriterImpl implements EmptyRoomCacheWriter {
private static final long PAYLOAD_TTL_MINUTES = 10L;
private static final long INDEX_TTL_MINUTES = PAYLOAD_TTL_MINUTES + 1L;
private final RedisService redisService;
@Override
public void save(ActiveVoiceRoomCO room) {
if (!isIndexable(room)) {
log.warn("跳过字段不完整的空房间索引, roomId={}, sysOrigin={}, region={}",
Objects.isNull(room) ? null : room.getId(),
Objects.isNull(room) ? null : room.getSysOrigin(),
Objects.isNull(room) ? null : room.getRegion());
return;
}
String payloadKey = EmptyRoomCacheKeys.payload(room.getId());
try {
// 保持原有 SETNX 语义普通成员频繁退出不能持续延长同一空房间的展示窗口
redisService.setIfAbsent(
payloadKey,
JSON.toJSONString(room),
PAYLOAD_TTL_MINUTES,
TimeUnit.MINUTES
);
Long remainingMillis = ensurePayloadTtl(payloadKey, room);
if (Objects.isNull(remainingMillis) || remainingMillis <= 0L) {
log.warn("空房间快照在建立索引前已过期, roomId={}, payloadKey={}", room.getId(), payloadKey);
return;
}
double expiresAtMillis = System.currentTimeMillis() + remainingMillis;
String roomMember = room.getId().toString();
addIndexMember(
EmptyRoomCacheKeys.regionIndex(room.getSysOrigin(), room.getRegion()),
roomMember,
expiresAtMillis
);
addIndexMember(
EmptyRoomCacheKeys.allRegionsIndex(room.getSysOrigin()),
roomMember,
expiresAtMillis
);
} catch (RuntimeException e) {
// 空房间只是发现页补充数据索引故障不能反向破坏已经完成的退房主链路
log.error("维护空房间索引失败, roomId={}, sysOrigin={}, region={}",
room.getId(), room.getSysOrigin(), room.getRegion(), e);
}
}
private Long ensurePayloadTtl(String payloadKey, ActiveVoiceRoomCO room) {
Long remainingMillis = redisService.getExpire(payloadKey, TimeUnit.MILLISECONDS);
if (Objects.nonNull(remainingMillis) && remainingMillis > 0L) {
return remainingMillis;
}
if (Objects.equals(remainingMillis, -1L)) {
// 修复历史上可能残留的无 TTL 快照避免索引成员永久存在
redisService.expire(payloadKey, PAYLOAD_TTL_MINUTES, TimeUnit.MINUTES);
return TimeUnit.MINUTES.toMillis(PAYLOAD_TTL_MINUTES);
}
// 快照可能恰好在 SETNX TTL 查询之间过期仅重试一次避免异常 Redis 下无界重试
redisService.setIfAbsent(
payloadKey,
JSON.toJSONString(room),
PAYLOAD_TTL_MINUTES,
TimeUnit.MINUTES
);
return redisService.getExpire(payloadKey, TimeUnit.MILLISECONDS);
}
private void addIndexMember(String indexKey, String roomMember, double expiresAtMillis) {
// 每次写入先增量淘汰已过期成员持续活跃的 ALL 索引也只保留最近十分钟窗口避免集中大删除
redisService.zsetRemoveByScore(indexKey, 0D, System.currentTimeMillis());
redisService.zsetAdd(indexKey, roomMember, expiresAtMillis);
// 索引 TTL 比成员窗口多一分钟无新增写入时最后一批成员清理后索引也会自然释放
redisService.expire(indexKey, INDEX_TTL_MINUTES, TimeUnit.MINUTES);
}
private boolean isIndexable(ActiveVoiceRoomCO room) {
return Objects.nonNull(room)
&& Objects.nonNull(room.getId())
&& StringUtils.isNotBlank(room.getSysOrigin())
&& StringUtils.isNotBlank(room.getRegion());
}
}

View File

@ -0,0 +1,276 @@
package com.red.circle.live.infra.database.cache.service.impl;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import com.alibaba.fastjson.JSON;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO;
import java.lang.reflect.Proxy;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
public class EmptyRoomCacheWriterImplTest {
private static final long ROOM_ID = 1001L;
private static final long PAYLOAD_TTL_MILLIS = TimeUnit.MINUTES.toMillis(10L);
@Test
public void saveShouldKeepLegacyPayloadAndWriteRegionAndAllRegionIndexes() {
RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
redisTemplate.remainingTtls.add(TimeUnit.MINUTES.toMillis(7L));
EmptyRoomCacheWriterImpl writer = writer(redisTemplate);
long beforeSave = System.currentTimeMillis();
writer.save(room(ROOM_ID, "YOLO", "AE"));
long afterSave = System.currentTimeMillis();
assertEquals(1, redisTemplate.payloadWrites.size());
PayloadWrite payloadWrite = redisTemplate.payloadWrites.get(0);
assertEquals("empty_room_cache:1001", payloadWrite.key());
assertEquals(10L, payloadWrite.ttl());
assertEquals(TimeUnit.MINUTES, payloadWrite.unit());
ActiveVoiceRoomCO cachedRoom = JSON.parseObject(payloadWrite.value().toString(),
ActiveVoiceRoomCO.class);
assertEquals(Long.valueOf(ROOM_ID), cachedRoom.getId());
assertEquals("YOLO", cachedRoom.getSysOrigin());
assertEquals("AE", cachedRoom.getRegion());
assertEquals(2, redisTemplate.indexWrites.size());
assertEquals(2, redisTemplate.indexCleanupCalls.size());
IndexWrite regionIndex = redisTemplate.indexWrites.get(0);
IndexWrite allRegionsIndex = redisTemplate.indexWrites.get(1);
assertEquals("empty_room_index:v2:YOLO:AE", regionIndex.key());
assertEquals("empty_room_index:v2:YOLO:__ALL__", allRegionsIndex.key());
assertEquals("1001", regionIndex.member());
assertEquals("1001", allRegionsIndex.member());
assertEquals(regionIndex.score(), allRegionsIndex.score(), 0D);
assertEquals("empty_room_index:v2:YOLO:AE", redisTemplate.indexCleanupCalls.get(0).key());
assertEquals("empty_room_index:v2:YOLO:__ALL__", redisTemplate.indexCleanupCalls.get(1).key());
long remainingMillis = TimeUnit.MINUTES.toMillis(7L);
assertTrue(regionIndex.score() >= beforeSave + remainingMillis);
assertTrue(regionIndex.score() <= afterSave + remainingMillis);
assertEquals(List.of(
new ExpireCall("empty_room_index:v2:YOLO:AE", 11L, TimeUnit.MINUTES),
new ExpireCall("empty_room_index:v2:YOLO:__ALL__", 11L, TimeUnit.MINUTES)
), redisTemplate.expireCalls);
assertTrue(redisTemplate.events.indexOf("payload:empty_room_cache:1001")
< redisTemplate.events.indexOf("index:empty_room_index:v2:YOLO:AE"));
}
@Test
public void saveShouldRepairLegacyPayloadWithoutTtlBeforeIndexing() {
RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
redisTemplate.remainingTtls.add(-1L);
EmptyRoomCacheWriterImpl writer = writer(redisTemplate);
long beforeSave = System.currentTimeMillis();
writer.save(room(ROOM_ID, "YOLO", "AE"));
long afterSave = System.currentTimeMillis();
assertEquals(new ExpireCall("empty_room_cache:1001", 10L, TimeUnit.MINUTES),
redisTemplate.expireCalls.get(0));
assertEquals(3, redisTemplate.expireCalls.size());
assertEquals(2, redisTemplate.indexWrites.size());
double score = redisTemplate.indexWrites.get(0).score();
assertTrue(score >= beforeSave + PAYLOAD_TTL_MILLIS);
assertTrue(score <= afterSave + PAYLOAD_TTL_MILLIS);
}
@Test
public void saveShouldRetrySnapshotThatExpiredBeforeTtlRead() {
RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
redisTemplate.remainingTtls.add(-2L);
redisTemplate.remainingTtls.add(TimeUnit.MINUTES.toMillis(5L));
EmptyRoomCacheWriterImpl writer = writer(redisTemplate);
writer.save(room(ROOM_ID, "YOLO", "AE"));
assertEquals(2, redisTemplate.payloadWrites.size());
assertEquals("empty_room_cache:1001", redisTemplate.payloadWrites.get(0).key());
assertEquals("empty_room_cache:1001", redisTemplate.payloadWrites.get(1).key());
assertEquals(2, redisTemplate.indexWrites.size());
}
@Test
public void saveShouldIgnoreRoomsThatCannotBePartitioned() {
RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
EmptyRoomCacheWriterImpl writer = writer(redisTemplate);
writer.save(null);
writer.save(room(null, "YOLO", "AE"));
writer.save(room(ROOM_ID, " ", "AE"));
writer.save(room(ROOM_ID, "YOLO", " "));
assertTrue(redisTemplate.payloadWrites.isEmpty());
assertTrue(redisTemplate.indexWrites.isEmpty());
assertTrue(redisTemplate.expireCalls.isEmpty());
assertEquals(0, redisTemplate.getExpireCalls);
}
@Test
public void saveShouldNotPropagatePayloadWriteFailure() {
RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
redisTemplate.payloadWriteFailure = new IllegalStateException("redis unavailable");
EmptyRoomCacheWriterImpl writer = writer(redisTemplate);
// 空房缓存是退房后的附加效果Redis 故障不能逃逸到调用者并破坏主链路
writer.save(room(ROOM_ID, "YOLO", "AE"));
assertEquals(1, redisTemplate.payloadWriteAttempts);
assertTrue(redisTemplate.indexWrites.isEmpty());
}
@Test
public void saveShouldNotPropagateIndexWriteFailure() {
RecordingRedisTemplate redisTemplate = new RecordingRedisTemplate();
redisTemplate.remainingTtls.add(TimeUnit.MINUTES.toMillis(5L));
redisTemplate.indexWriteFailure = new IllegalStateException("zset unavailable");
EmptyRoomCacheWriterImpl writer = writer(redisTemplate);
writer.save(room(ROOM_ID, "YOLO", "AE"));
assertEquals(1, redisTemplate.payloadWrites.size());
assertEquals(1, redisTemplate.indexWriteAttempts);
assertTrue(redisTemplate.indexWrites.isEmpty());
assertTrue(redisTemplate.expireCalls.isEmpty());
}
private static EmptyRoomCacheWriterImpl writer(RecordingRedisTemplate redisTemplate) {
return new EmptyRoomCacheWriterImpl(new RedisService(redisTemplate));
}
private static ActiveVoiceRoomCO room(Long roomId, String sysOrigin, String region) {
ActiveVoiceRoomCO room = new ActiveVoiceRoomCO();
room.setId(roomId);
room.setSysOrigin(sysOrigin);
room.setRegion(region);
room.setUserId(2002L);
room.setRoomAccount("10001");
return room;
}
private record PayloadWrite(String key, Object value, long ttl, TimeUnit unit) {
}
private record IndexWrite(String key, Object member, double score) {
}
private record IndexCleanupCall(String key, double minScore, double maxScore) {
}
private record ExpireCall(String key, long ttl, TimeUnit unit) {
}
/**
* 只实现 writer 允许使用的 Redis 命令生产代码若意外增加扫描或其他命令测试会立即失败
* 从而守住房间发现链路不能再退化为全库遍历的边界
*/
private static final class RecordingRedisTemplate extends RedisTemplate<String, Object> {
private final List<PayloadWrite> payloadWrites = new ArrayList<>();
private final List<IndexWrite> indexWrites = new ArrayList<>();
private final List<IndexCleanupCall> indexCleanupCalls = new ArrayList<>();
private final List<ExpireCall> expireCalls = new ArrayList<>();
private final List<String> events = new ArrayList<>();
private final Queue<Long> remainingTtls = new ArrayDeque<>();
private RuntimeException payloadWriteFailure;
private RuntimeException indexWriteFailure;
private int payloadWriteAttempts;
private int indexWriteAttempts;
private int getExpireCalls;
@Override
@SuppressWarnings("unchecked")
public ValueOperations<String, Object> opsForValue() {
return proxy(ValueOperations.class, (methodName, args) -> {
if (Objects.equals("setIfAbsent", methodName) && args.length == 4) {
payloadWriteAttempts++;
if (payloadWriteFailure != null) {
throw payloadWriteFailure;
}
payloadWrites.add(new PayloadWrite(
args[0].toString(), args[1], ((Number) args[2]).longValue(), (TimeUnit) args[3]
));
events.add("payload:" + args[0]);
return Boolean.TRUE;
}
throw new UnsupportedOperationException(methodName);
});
}
@Override
@SuppressWarnings("unchecked")
public ZSetOperations<String, Object> opsForZSet() {
return proxy(ZSetOperations.class, (methodName, args) -> {
if (Objects.equals("add", methodName) && args.length == 3) {
indexWriteAttempts++;
if (indexWriteFailure != null) {
throw indexWriteFailure;
}
indexWrites.add(new IndexWrite(
args[0].toString(), args[1], ((Number) args[2]).doubleValue()
));
events.add("index:" + args[0]);
return Boolean.TRUE;
}
if (Objects.equals("removeRangeByScore", methodName) && args.length == 3) {
indexCleanupCalls.add(new IndexCleanupCall(
args[0].toString(),
((Number) args[1]).doubleValue(),
((Number) args[2]).doubleValue()
));
return 0L;
}
throw new UnsupportedOperationException(methodName);
});
}
@Override
public Long getExpire(String key, TimeUnit timeUnit) {
getExpireCalls++;
assertEquals(TimeUnit.MILLISECONDS, timeUnit);
if (remainingTtls.isEmpty()) {
throw new AssertionError("Missing getExpire result for " + key);
}
return remainingTtls.remove();
}
@Override
public Boolean expire(String key, long timeout, TimeUnit unit) {
expireCalls.add(new ExpireCall(key, timeout, unit));
return Boolean.TRUE;
}
@SuppressWarnings("unchecked")
private static <T> T proxy(Class<T> type, Operation operation) {
return (T) Proxy.newProxyInstance(type.getClassLoader(), new Class<?>[] {type},
(proxy, method, args) -> {
if (method.getDeclaringClass().equals(Object.class)) {
return switch (method.getName()) {
case "toString" -> type.getSimpleName() + "RecordingProxy";
case "hashCode" -> System.identityHashCode(proxy);
case "equals" -> proxy == args[0];
default -> throw new UnsupportedOperationException(method.getName());
};
}
return operation.invoke(method.getName(), args == null ? new Object[0] : args);
});
}
}
@FunctionalInterface
private interface Operation {
Object invoke(String methodName, Object[] args);
}
}

View File

@ -17,6 +17,7 @@ import com.red.circle.other.domain.gateway.user.ability.UserSVipGateway;
import com.red.circle.other.domain.model.user.UserProfile; import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.domain.rocket.RocketStatus; import com.red.circle.other.domain.rocket.RocketStatus;
import com.red.circle.other.infra.common.sys.CountryCodeAliasSupport; import com.red.circle.other.infra.common.sys.CountryCodeAliasSupport;
import com.red.circle.other.infra.database.cache.service.other.EmptyRoomCacheReader;
import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService; import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom; import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager; import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager;
@ -28,16 +29,13 @@ 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.sys.SysCountryCodeService;
import com.red.circle.other.infra.database.rds.service.user.device.LatestMobileDeviceService; 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.cmd.live.AppActiveVoiceRoomQryCmd;
import com.red.circle.other.inner.model.dto.live.ActiveVoiceRoomCO;
import com.red.circle.tool.core.collection.CollectionUtils; import com.red.circle.tool.core.collection.CollectionUtils;
import java.util.*; import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; 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; import org.springframework.stereotype.Component;
/** /**
@ -60,19 +58,20 @@ public class RoomVoiceDiscoverQryExe {
private final RoomProfileManagerService roomProfileManagerService; private final RoomProfileManagerService roomProfileManagerService;
private final RegionRoomCacheService regionRoomCacheService; private final RegionRoomCacheService regionRoomCacheService;
private final LiveMicClient liveMicClient; private final LiveMicClient liveMicClient;
private final RedisTemplate<String, Object> redisTemplate; private final EmptyRoomCacheReader emptyRoomCacheReader;
private final RocketStatusCacheService rocketStatusCacheService; private final RocketStatusCacheService rocketStatusCacheService;
private final GameLudoService gameLudoService; private final GameLudoService gameLudoService;
private final UserProfileGateway userProfileGateway; private final UserProfileGateway userProfileGateway;
private final CountryCodeAliasSupport countryCodeAliasSupport; private final CountryCodeAliasSupport countryCodeAliasSupport;
private static final String EMPTY_ROOM_CACHE_KEY = "empty_room_cache:*";
private static final String MEMBER_QUANTITY_EXT_KEY = "memberQuantity"; private static final String MEMBER_QUANTITY_EXT_KEY = "memberQuantity";
private static final String FIXED_WEIGHTS_EXT_KEY = "fixedWeights"; private static final String FIXED_WEIGHTS_EXT_KEY = "fixedWeights";
private static final int DISCOVERY_LIMIT = 50; private static final int DISCOVERY_LIMIT = 50;
private static final long DISCOVERY_SLOW_STEP_MILLIS = 100L; private static final long DISCOVERY_SLOW_STEP_MILLIS = 100L;
private static final long DISCOVERY_SLOW_TOTAL_MILLIS = 300L; private static final long DISCOVERY_SLOW_TOTAL_MILLIS = 300L;
private static final long EMPTY_ROOM_SCAN_COUNT = 200L; private static final int REBUILD_WAIT_ATTEMPTS = 20;
private static final long REBUILD_WAIT_MILLIS = 50L;
private static final long EMPTY_RESULT_CACHE_SECONDS = 30L;
public List<RoomVoiceProfileCO> execute(AppExtCommand cmd) { public List<RoomVoiceProfileCO> execute(AppExtCommand cmd) {
long totalStart = System.currentTimeMillis(); long totalStart = System.currentTimeMillis();
@ -87,99 +86,160 @@ public class RoomVoiceDiscoverQryExe {
stepStart = logDiscoveryStep("prepare", stepStart, reqUserId, sysOrigin, region); stepStart = logDiscoveryStep("prepare", stepStart, reqUserId, sysOrigin, region);
String key = discoveryCacheKey(sysOrigin, region, allRegionWhiteListUser); String key = discoveryCacheKey(sysOrigin, region, allRegionWhiteListUser);
String regionsRoom = regionRoomCacheService.getRegionsRoom(key); List<RoomVoiceProfileCO> cachedRooms = readDiscoveryCache(key);
stepStart = logDiscoveryStep("read-region-cache", stepStart, reqUserId, sysOrigin, region); stepStart = logDiscoveryStep("read-region-cache", stepStart, reqUserId, sysOrigin, region);
if (Objects.nonNull(regionsRoom)) { if (Objects.nonNull(cachedRooms)) {
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
.filterAvailableRoomProfiles(roomList);
if (availableRooms.size() != roomList.size()) {
regionRoomCacheService.remove(key);
}
stepStart = logDiscoveryStep("filter-cache", stepStart, reqUserId, sysOrigin, region); stepStart = logDiscoveryStep("filter-cache", stepStart, reqUserId, sysOrigin, region);
logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, true, availableRooms.size()); logDiscoveryTotal(
return availableRooms; totalStart,
reqUserId,
sysOrigin,
region,
allRegionWhiteListUser,
true,
cachedRooms.size()
);
return cachedRooms;
} }
List<ActiveVoiceRoom> activeVoiceRooms = allRegionWhiteListUser String rebuildToken = UUID.randomUUID().toString();
? activeVoiceRoomService.listDiscover( boolean rebuildLockAcquired = false;
boolean rebuildLockUnavailable = false;
try {
rebuildLockAcquired = regionRoomCacheService.tryAcquireRebuildLock(key, rebuildToken);
} catch (RuntimeException e) {
// Redis 锁异常时保留发现页可用性后续缓存/索引访问仍各自 fail-open不扩大主链路故障
rebuildLockUnavailable = true;
log.error("发现缓存重建锁不可用,进入无锁降级, key={}, userId={}", key, reqUserId, e);
}
if (!rebuildLockAcquired && !rebuildLockUnavailable) {
List<RoomVoiceProfileCO> publishedRooms = waitForPublishedDiscoveryCache(key);
stepStart = logDiscoveryStep("wait-region-cache", stepStart, reqUserId, sysOrigin, region);
if (Objects.nonNull(publishedRooms)) {
logDiscoveryTotal(
totalStart,
reqUserId,
sysOrigin, sysOrigin,
true,
region, region,
Optional.ofNullable(userProfile).map(UserProfile::getCountryCode).orElse(null), allRegionWhiteListUser,
DISCOVERY_LIMIT true,
) publishedRooms.size()
: roomRegionFilterCommon.filterActiveVoiceRoomsByRegion( );
activeVoiceRoomService.listByRegion(sysOrigin, region, DISCOVERY_LIMIT), return publishedRooms;
}
try {
// leader 可能失败且已释放锁等待结束后只允许重新竞争禁止 follower 直接并发回源
rebuildLockAcquired = regionRoomCacheService.tryAcquireRebuildLock(key, rebuildToken);
} catch (RuntimeException e) {
rebuildLockUnavailable = true;
log.error("等待后重试发现缓存重建锁失败,进入无锁降级, key={}, userId={}", key, reqUserId, e);
}
if (!rebuildLockAcquired && !rebuildLockUnavailable) {
log.warn("等待发现缓存重建超时,拒绝并发回源, key={}, userId={}", key, reqUserId);
logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, 0);
return CollectionUtils.newArrayList();
}
}
try {
if (rebuildLockAcquired) {
// 获取锁后必须二次检查覆盖首次 miss SETNX 之间已有 leader 发布的竞态窗口
List<RoomVoiceProfileCO> doubleCheckedRooms = readDiscoveryCache(key);
stepStart = logDiscoveryStep(
"double-check-region-cache",
stepStart,
reqUserId,
sysOrigin,
region region
); );
stepStart = logDiscoveryStep("query-active-rooms", stepStart, reqUserId, sysOrigin, region); if (Objects.nonNull(doubleCheckedRooms)) {
logDiscoveryTotal(
totalStart,
reqUserId,
sysOrigin,
region,
allRegionWhiteListUser,
true,
doubleCheckedRooms.size()
);
return doubleCheckedRooms;
}
}
// 查询没人的房间缓存加入进去 List<ActiveVoiceRoom> activeVoiceRooms = allRegionWhiteListUser
List<ActiveVoiceRoom> emptyRooms = getEmptyRoomsFromCache(); ? activeVoiceRoomService.listDiscover(
stepStart = logDiscoveryStep("scan-empty-room-cache", stepStart, reqUserId, sysOrigin, region); sysOrigin,
if (CollectionUtils.isNotEmpty(emptyRooms)) { true,
List<ActiveVoiceRoom> matchedEmptyRooms = allRegionWhiteListUser region,
? emptyRooms.stream() Optional.ofNullable(userProfile).map(UserProfile::getCountryCode).orElse(null),
.filter(room -> Objects.equals(sysOrigin, room.getSysOrigin())) DISCOVERY_LIMIT
.toList() )
: roomRegionFilterCommon.filterActiveVoiceRoomsByRegion( : roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(
emptyRooms.stream() activeVoiceRoomService.listByRegion(sysOrigin, region, DISCOVERY_LIMIT),
.filter(room -> Objects.equals(sysOrigin, room.getSysOrigin()))
.toList(),
region region
); );
if (CollectionUtils.isNotEmpty(matchedEmptyRooms)) { stepStart = logDiscoveryStep("query-active-rooms", stepStart, reqUserId, sysOrigin, region);
// 索引已经按系统/区域收敛读取成本只与当前候选空房数量相关不再遍历 DB0 全量 key
List<ActiveVoiceRoom> emptyRooms = emptyRoomCacheReader.list(
sysOrigin,
region,
allRegionWhiteListUser,
DISCOVERY_LIMIT
);
stepStart = logDiscoveryStep("read-empty-room-index", stepStart, reqUserId, sysOrigin, region);
if (CollectionUtils.isNotEmpty(emptyRooms)) {
List<ActiveVoiceRoom> allRooms = new ArrayList<>(activeVoiceRooms); List<ActiveVoiceRoom> allRooms = new ArrayList<>(activeVoiceRooms);
Set<Long> existingRoomIds = activeVoiceRooms.stream() Set<Long> existingRoomIds = activeVoiceRooms.stream()
.map(ActiveVoiceRoom::getId) .map(ActiveVoiceRoom::getId)
.collect(Collectors.toSet()); .collect(Collectors.toSet());
matchedEmptyRooms.stream() emptyRooms.stream()
.filter(room -> !existingRoomIds.contains(room.getId())) .filter(room -> !existingRoomIds.contains(room.getId()))
.forEach(allRooms::add); .forEach(allRooms::add);
activeVoiceRooms = allRooms; activeVoiceRooms = allRooms;
} }
} stepStart = logDiscoveryStep("merge-empty-rooms", stepStart, reqUserId, sysOrigin, region);
stepStart = logDiscoveryStep("merge-empty-rooms", stepStart, reqUserId, sysOrigin, region);
List<RoomVoiceProfileCO> roomList = roomVoiceProfileCommon.toListRoomVoiceProfileCO(activeVoiceRooms); List<RoomVoiceProfileCO> roomList = roomVoiceProfileCommon.toListRoomVoiceProfileCO(activeVoiceRooms);
stepStart = logDiscoveryStep("build-profile-list", stepStart, reqUserId, sysOrigin, region); stepStart = logDiscoveryStep("build-profile-list", stepStart, reqUserId, sysOrigin, region);
if (roomList.size() < DISCOVERY_LIMIT) { if (roomList.size() < DISCOVERY_LIMIT) {
List<RoomProfileManager> fallbackRoomManagersSource = roomProfileManagerService.listSelectiveLimitByCountryCodes( List<RoomProfileManager> fallbackRoomManagersSource =
sysOrigin, roomProfileManagerService.listSelectiveLimitByCountryCodes(
allRegionWhiteListUser ? List.of() : roomRegionFilterCommon.listCountryCodes(currentRegion), sysOrigin,
roomList.stream().map(RoomVoiceProfileCO::getId).toList(), allRegionWhiteListUser ? List.of() : roomRegionFilterCommon.listCountryCodes(currentRegion),
DISCOVERY_LIMIT roomList.stream().map(RoomVoiceProfileCO::getId).toList(),
); DISCOVERY_LIMIT
List<RoomProfileManager> fallbackRoomManagers = allRegionWhiteListUser );
? fallbackRoomManagersSource List<RoomProfileManager> fallbackRoomManagers = allRegionWhiteListUser
: roomRegionFilterCommon.filterRoomProfileManagersByRegion( ? fallbackRoomManagersSource
fallbackRoomManagersSource, : roomRegionFilterCommon.filterRoomProfileManagersByRegion(
region fallbackRoomManagersSource,
); region
roomList = mergeRoomList(roomList, roomVoiceProfileCommon.listRoomVoiceProfilesV2(fallbackRoomManagers), );
DISCOVERY_LIMIT); roomList = mergeRoomList(
} roomList,
stepStart = logDiscoveryStep("fallback-room-profiles", stepStart, reqUserId, sysOrigin, region); roomVoiceProfileCommon.listRoomVoiceProfilesV2(fallbackRoomManagers),
roomList = sortByTopWeightAndMemberQuantityDesc(roomList); DISCOVERY_LIMIT
stepStart = logDiscoveryStep("sort-rooms", stepStart, reqUserId, sysOrigin, region); );
}
stepStart = logDiscoveryStep("fallback-room-profiles", stepStart, reqUserId, sysOrigin, region);
roomList = sortByTopWeightAndMemberQuantityDesc(roomList);
stepStart = logDiscoveryStep("sort-rooms", stepStart, reqUserId, sysOrigin, region);
if (CollectionUtils.isEmpty(roomList)) { if (CollectionUtils.isEmpty(roomList)) {
logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, 0); // 短负缓存让空区域的 follower 也能结束等待但不会把新开房间隐藏十分钟
return CollectionUtils.newArrayList(); saveDiscoveryCache(key, roomList, reqUserId, sysOrigin, region);
} logDiscoveryStep("write-empty-region-cache", stepStart, reqUserId, sysOrigin, region);
Map<Long, SVIPLevelEnum> svipLevelEnumMap = userSVipGateway.checkSVipIdentity( logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, 0);
roomList.stream().map(RoomVoiceProfileCO::getUserId).collect( return CollectionUtils.newArrayList();
Collectors.toSet())); }
roomList = roomList.stream() Map<Long, SVIPLevelEnum> svipLevelEnumMap = userSVipGateway.checkSVipIdentity(
.peek(room -> { roomList.stream().map(RoomVoiceProfileCO::getUserId).collect(Collectors.toSet()));
room.setUserSVipLevel(svipLevelEnumMap.get(room.getUserId())); roomList = roomList.stream()
}) .peek(room -> room.setUserSVipLevel(svipLevelEnumMap.get(room.getUserId())))
.toList(); .toList();
stepStart = logDiscoveryStep("fill-svip", stepStart, reqUserId, sysOrigin, region); stepStart = logDiscoveryStep("fill-svip", stepStart, reqUserId, sysOrigin, region);
// 填充火箭状态 // 填充火箭状态
@ -192,7 +252,12 @@ public class RoomVoiceDiscoverQryExe {
logDiscoveryStep("write-region-cache", stepStart, reqUserId, sysOrigin, region); logDiscoveryStep("write-region-cache", stepStart, reqUserId, sysOrigin, region);
logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, roomList.size()); logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, roomList.size());
return roomList; return roomList;
} finally {
if (rebuildLockAcquired) {
releaseRebuildLock(key, rebuildToken, reqUserId);
}
}
} }
private List<RoomVoiceProfileCO> sortByTopWeightAndMemberQuantityDesc( private List<RoomVoiceProfileCO> sortByTopWeightAndMemberQuantityDesc(
@ -236,13 +301,105 @@ public class RoomVoiceDiscoverQryExe {
return allRegionWhiteListUser ? sysOrigin + region + ":ALL_REGION" : sysOrigin + region; return allRegionWhiteListUser ? sysOrigin + region + ":ALL_REGION" : sysOrigin + region;
} }
private void saveDiscoveryCache(String key, List<RoomVoiceProfileCO> roomList, Long userId, /**
String sysOrigin, String region) { * 读取并校验发现页缓存返回 null 表示真正 miss List 表示命中的负缓存
if (CollectionUtils.isEmpty(roomList)) { */
return; private List<RoomVoiceProfileCO> readDiscoveryCache(String key) {
return readDiscoveryCache(key, true);
}
private List<RoomVoiceProfileCO> readDiscoveryCache(String key, boolean logReadFailure) {
String regionsRoom;
try {
regionsRoom = regionRoomCacheService.getRegionsRoom(key);
} catch (RuntimeException e) {
// Redis GET 超时按真正 miss 处理后续仍由分布式锁或无锁降级决定是否回源
if (logReadFailure) {
log.warn("读取发现缓存失败,按 miss 降级, key={}", key, e);
}
return null;
}
if (Objects.isNull(regionsRoom)) {
return null;
} }
try { try {
regionRoomCacheService.saveRegionsRoom(JSON.toJSONString(roomList), key); List<RoomVoiceProfileCO> roomList = JSON.parseArray(regionsRoom, RoomVoiceProfileCO.class);
if (Objects.isNull(roomList)) {
removeDiscoveryCache(key);
return null;
}
if (CollectionUtils.isEmpty(roomList)) {
return CollectionUtils.newArrayList();
}
List<RoomVoiceProfileCO> availableRooms = roomVoiceProfileCommon
.filterAvailableRoomProfiles(roomList);
if (availableRooms.size() != roomList.size()) {
// 本次仍返回过滤后的安全结果下一请求通过 singleflight 重建完整缓存
removeDiscoveryCache(key);
}
return availableRooms;
} catch (RuntimeException e) {
removeDiscoveryCache(key);
log.warn("清理无法解析的发现页缓存, key={}", key, e);
return null;
}
}
private List<RoomVoiceProfileCO> waitForPublishedDiscoveryCache(String key) {
for (int attempt = 0; attempt < REBUILD_WAIT_ATTEMPTS; attempt++) {
// follower 轮询期间抑制重复 GET 异常日志初次读取和锁竞争仍会保留故障证据
List<RoomVoiceProfileCO> cachedRooms = readDiscoveryCache(key, false);
if (Objects.nonNull(cachedRooms)) {
return cachedRooms;
}
if (attempt + 1 >= REBUILD_WAIT_ATTEMPTS) {
break;
}
try {
Thread.sleep(REBUILD_WAIT_MILLIS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return null;
}
}
return null;
}
private void removeDiscoveryCache(String key) {
try {
regionRoomCacheService.remove(key);
} catch (RuntimeException e) {
// 清理失败不能覆盖本次安全返回残留坏值由 TTL 或下一次成功清理收敛
log.warn("删除失效发现缓存失败, key={}", key, e);
}
}
private void releaseRebuildLock(String key, String token, Long userId) {
try {
if (!regionRoomCacheService.releaseRebuildLock(key, token)) {
// token 不一致通常表示租约已过期Lua 保证不会误删新的持有者
log.warn("发现缓存重建锁未释放,等待 TTL 自愈, key={}, userId={}", key, userId);
}
} catch (RuntimeException e) {
log.error("释放发现缓存重建锁失败, key={}, userId={}", key, userId, e);
}
}
private void saveDiscoveryCache(String key, List<RoomVoiceProfileCO> roomList, Long userId,
String sysOrigin, String region) {
try {
String payload = JSON.toJSONString(roomList);
if (CollectionUtils.isEmpty(roomList)) {
regionRoomCacheService.saveRegionsRoom(
payload,
key,
EMPTY_RESULT_CACHE_SECONDS,
TimeUnit.SECONDS
);
} else {
regionRoomCacheService.saveRegionsRoom(payload, key);
}
} catch (Exception e) { } catch (Exception e) {
log.warn("写入房间发现缓存失败, key={}, userId={}, sysOrigin={}, region={}", log.warn("写入房间发现缓存失败, key={}, userId={}, sysOrigin={}, region={}",
key, userId, sysOrigin, region, e); key, userId, sysOrigin, region, e);
@ -410,86 +567,6 @@ public class RoomVoiceDiscoverQryExe {
} }
} }
/**
* 从缓存中获取没人的房间
*
* @return
*/
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) {
String cachedRoom = (String) redisTemplate.opsForValue().get(key);
if (cachedRoom != null) {
try {
// 将缓存对象转换为ActiveVoiceRoomCO再转换为ActiveVoiceRoom
ActiveVoiceRoomCO roomCO = JSON.parseObject(cachedRoom, ActiveVoiceRoomCO.class);
if (roomCO != null) {
ActiveVoiceRoom room = convertToActiveVoiceRoom(roomCO);
if (room != null) {
emptyRooms.add(room);
}
}
} catch (Exception e) {
log.warn("转换缓存房间对象失败, key: {}", key, e);
}
}
}
return emptyRooms;
} catch (Exception e) {
log.error("获取空房间缓存失败", e);
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
*/
private ActiveVoiceRoom convertToActiveVoiceRoom(ActiveVoiceRoomCO roomCO) {
try {
ActiveVoiceRoom room = new ActiveVoiceRoom();
BeanUtils.copyProperties(roomCO, room);
// 设置在线人数为0因为这是没人的房间
room.setOnlineQuantity(0L);
// 设置默认权重如果未设置的话
if (room.getWeights() == null) {
room.setWeights(0);
}
if (room.getFixedWeights() == null) {
room.setFixedWeights(0);
}
return room;
} catch (Exception e) {
log.error("转换ActiveVoiceRoomCO失败", e);
return null;
}
}
/** /**
* 去除拉黑和被上锁的的房间 * 去除拉黑和被上锁的的房间
* *

View File

@ -1,6 +1,7 @@
package com.red.circle.other.app.command.room.query; package com.red.circle.other.app.command.room.query;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.anySet; import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never; import static org.mockito.Mockito.never;
@ -22,6 +23,7 @@ import com.red.circle.other.domain.gateway.user.ability.UserSVipGateway;
import com.red.circle.other.domain.model.user.UserProfile; import com.red.circle.other.domain.model.user.UserProfile;
import com.red.circle.other.domain.model.user.ability.RegionConfig; import com.red.circle.other.domain.model.user.ability.RegionConfig;
import com.red.circle.other.infra.common.sys.CountryCodeAliasSupport; import com.red.circle.other.infra.common.sys.CountryCodeAliasSupport;
import com.red.circle.other.infra.database.cache.service.other.EmptyRoomCacheReader;
import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService; import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom; import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager; import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager;
@ -31,9 +33,8 @@ import com.red.circle.other.infra.database.mongo.service.live.RoomUserBlacklistS
import com.red.circle.other.infra.database.rds.service.user.device.LatestMobileDeviceService; import com.red.circle.other.infra.database.rds.service.user.device.LatestMobileDeviceService;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.springframework.data.redis.core.RedisTemplate;
class RoomVoiceDiscoverQryExeTest { class RoomVoiceDiscoverQryExeTest {
@ -51,7 +52,7 @@ class RoomVoiceDiscoverQryExeTest {
RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class); RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class);
RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class); RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class);
LiveMicClient liveMicClient = mock(LiveMicClient.class); LiveMicClient liveMicClient = mock(LiveMicClient.class);
RedisTemplate<String, Object> redisTemplate = mock(RedisTemplate.class); EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class);
RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class); RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class);
GameLudoService gameLudoService = mock(GameLudoService.class); GameLudoService gameLudoService = mock(GameLudoService.class);
UserProfileGateway userProfileGateway = mock(UserProfileGateway.class); UserProfileGateway userProfileGateway = mock(UserProfileGateway.class);
@ -67,7 +68,7 @@ class RoomVoiceDiscoverQryExeTest {
roomProfileManagerService, roomProfileManagerService,
regionRoomCacheService, regionRoomCacheService,
liveMicClient, liveMicClient,
redisTemplate, emptyRoomCacheReader,
rocketStatusCacheService, rocketStatusCacheService,
gameLudoService, gameLudoService,
userProfileGateway, userProfileGateway,
@ -94,10 +95,12 @@ class RoomVoiceDiscoverQryExeTest {
when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion); when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion);
when(userProfileGateway.getByUserId(1L)).thenReturn(userProfile); when(userProfileGateway.getByUserId(1L)).thenReturn(userProfile);
when(regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null); when(regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null);
when(regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString())).thenReturn(true);
when(regionRoomCacheService.releaseRebuildLock(anyString(), anyString())).thenReturn(true);
when(activeVoiceRoomService.listByRegion("YOLO", "AE", 50)).thenReturn(List.of(activeVoiceRoom)); when(activeVoiceRoomService.listByRegion("YOLO", "AE", 50)).thenReturn(List.of(activeVoiceRoom));
when(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE")) when(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE"))
.thenReturn(List.of(activeVoiceRoom)); .thenReturn(List.of(activeVoiceRoom));
when(redisTemplate.keys("empty_room_cache:*")).thenReturn(Set.of()); when(emptyRoomCacheReader.list("YOLO", "AE", false, 50)).thenReturn(List.of());
when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn( when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn(
List.of(activeRoomCo) List.of(activeRoomCo)
); );
@ -121,6 +124,8 @@ class RoomVoiceDiscoverQryExeTest {
assertEquals(List.of(fallbackRoomCo, activeRoomCo), roomList); assertEquals(List.of(fallbackRoomCo, activeRoomCo), roomList);
verify(activeVoiceRoomService).listByRegion("YOLO", "AE", 50); verify(activeVoiceRoomService).listByRegion("YOLO", "AE", 50);
verify(emptyRoomCacheReader).list("YOLO", "AE", false, 50);
verify(regionRoomCacheService).releaseRebuildLock(anyString(), anyString());
verify(activeVoiceRoomService, never()).listDiscover("YOLO", true, "AE", null, 50); verify(activeVoiceRoomService, never()).listDiscover("YOLO", true, "AE", null, 50);
} }
@ -136,7 +141,7 @@ class RoomVoiceDiscoverQryExeTest {
RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class); RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class);
RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class); RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class);
LiveMicClient liveMicClient = mock(LiveMicClient.class); LiveMicClient liveMicClient = mock(LiveMicClient.class);
RedisTemplate<String, Object> redisTemplate = mock(RedisTemplate.class); EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class);
RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class); RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class);
GameLudoService gameLudoService = mock(GameLudoService.class); GameLudoService gameLudoService = mock(GameLudoService.class);
UserProfileGateway userProfileGateway = mock(UserProfileGateway.class); UserProfileGateway userProfileGateway = mock(UserProfileGateway.class);
@ -152,7 +157,7 @@ class RoomVoiceDiscoverQryExeTest {
roomProfileManagerService, roomProfileManagerService,
regionRoomCacheService, regionRoomCacheService,
liveMicClient, liveMicClient,
redisTemplate, emptyRoomCacheReader,
rocketStatusCacheService, rocketStatusCacheService,
gameLudoService, gameLudoService,
userProfileGateway, userProfileGateway,
@ -179,9 +184,12 @@ class RoomVoiceDiscoverQryExeTest {
when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion); when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion);
when(userProfileGateway.getByUserId(1L)).thenReturn(userProfile); when(userProfileGateway.getByUserId(1L)).thenReturn(userProfile);
when(regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null); when(regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null);
when(regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString())).thenReturn(true);
when(regionRoomCacheService.releaseRebuildLock(anyString(), anyString())).thenReturn(true);
when(activeVoiceRoomService.listByRegion("YOLO", "AE", 50)).thenReturn(List.of(activeVoiceRoom)); when(activeVoiceRoomService.listByRegion("YOLO", "AE", 50)).thenReturn(List.of(activeVoiceRoom));
when(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE")) when(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE"))
.thenReturn(List.of(activeVoiceRoom)); .thenReturn(List.of(activeVoiceRoom));
when(emptyRoomCacheReader.list("YOLO", "AE", false, 50)).thenReturn(List.of());
when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn( when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn(
List.of(topRoomCo) List.of(topRoomCo)
); );
@ -218,7 +226,7 @@ class RoomVoiceDiscoverQryExeTest {
RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class); RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class);
RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class); RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class);
LiveMicClient liveMicClient = mock(LiveMicClient.class); LiveMicClient liveMicClient = mock(LiveMicClient.class);
RedisTemplate<String, Object> redisTemplate = mock(RedisTemplate.class); EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class);
RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class); RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class);
GameLudoService gameLudoService = mock(GameLudoService.class); GameLudoService gameLudoService = mock(GameLudoService.class);
UserProfileGateway userProfileGateway = mock(UserProfileGateway.class); UserProfileGateway userProfileGateway = mock(UserProfileGateway.class);
@ -234,7 +242,7 @@ class RoomVoiceDiscoverQryExeTest {
roomProfileManagerService, roomProfileManagerService,
regionRoomCacheService, regionRoomCacheService,
liveMicClient, liveMicClient,
redisTemplate, emptyRoomCacheReader,
rocketStatusCacheService, rocketStatusCacheService,
gameLudoService, gameLudoService,
userProfileGateway, userProfileGateway,
@ -260,9 +268,11 @@ class RoomVoiceDiscoverQryExeTest {
when(roomRegionFilterCommon.requireRegionConfig(ALL_REGION_WHITE_USER_ID)).thenReturn(currentRegion); when(roomRegionFilterCommon.requireRegionConfig(ALL_REGION_WHITE_USER_ID)).thenReturn(currentRegion);
when(userProfileGateway.getByUserId(ALL_REGION_WHITE_USER_ID)).thenReturn(userProfile); when(userProfileGateway.getByUserId(ALL_REGION_WHITE_USER_ID)).thenReturn(userProfile);
when(regionRoomCacheService.getRegionsRoom("YOLOAE:ALL_REGION")).thenReturn(null); when(regionRoomCacheService.getRegionsRoom("YOLOAE:ALL_REGION")).thenReturn(null);
when(regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString())).thenReturn(true);
when(regionRoomCacheService.releaseRebuildLock(anyString(), anyString())).thenReturn(true);
when(activeVoiceRoomService.listDiscover("YOLO", true, "AE", "AE", 50)) when(activeVoiceRoomService.listDiscover("YOLO", true, "AE", "AE", 50))
.thenReturn(List.of(activeVoiceRoom)); .thenReturn(List.of(activeVoiceRoom));
when(redisTemplate.keys("empty_room_cache:*")).thenReturn(Set.of()); when(emptyRoomCacheReader.list("YOLO", "AE", true, 50)).thenReturn(List.of());
when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn( when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn(
List.of(activeRoomCo) List.of(activeRoomCo)
); );
@ -282,7 +292,151 @@ class RoomVoiceDiscoverQryExeTest {
assertEquals(List.of(activeRoomCo, fallbackRoomCo), roomList); assertEquals(List.of(activeRoomCo, fallbackRoomCo), roomList);
verify(activeVoiceRoomService, never()).listByRegion("YOLO", "AE", 50); verify(activeVoiceRoomService, never()).listByRegion("YOLO", "AE", 50);
verify(emptyRoomCacheReader).list("YOLO", "AE", true, 50);
verify(roomRegionFilterCommon, never()).filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE"); verify(roomRegionFilterCommon, never()).filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE");
verify(roomRegionFilterCommon, never()).filterRoomProfileManagersByRegion(List.of(fallbackRoomManager), "AE"); verify(roomRegionFilterCommon, never()).filterRoomProfileManagersByRegion(List.of(fallbackRoomManager), "AE");
} }
@Test
void execute_shouldDoubleCheckCacheAfterWinningRebuildLock() {
DiscoveryFixture fixture = new DiscoveryFixture();
AppExtCommand cmd = fixture.stubCurrentRegionRequest();
// 首次读取 miss 另一实例可能已经发布leader 获锁后必须二次检查不能重复回源
when(fixture.regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null, "[]");
when(fixture.regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString()))
.thenReturn(true);
when(fixture.regionRoomCacheService.releaseRebuildLock(anyString(), anyString()))
.thenReturn(true);
assertEquals(List.of(), fixture.exe.execute(cmd));
verify(fixture.activeVoiceRoomService, never()).listByRegion("YOLO", "AE", 50);
verify(fixture.emptyRoomCacheReader, never()).list("YOLO", "AE", false, 50);
verify(fixture.regionRoomCacheService).releaseRebuildLock(anyString(), anyString());
}
@Test
void execute_shouldWaitForLeaderCacheWithoutFollowerOriginQuery() {
DiscoveryFixture fixture = new DiscoveryFixture();
AppExtCommand cmd = fixture.stubCurrentRegionRequest();
// follower 第一次 miss 后立即读到 leader 发布的负缓存不能自行查询 Mongo 或空房索引
when(fixture.regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null, "[]");
when(fixture.regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString()))
.thenReturn(false);
assertEquals(List.of(), fixture.exe.execute(cmd));
verify(fixture.activeVoiceRoomService, never()).listByRegion("YOLO", "AE", 50);
verify(fixture.emptyRoomCacheReader, never()).list("YOLO", "AE", false, 50);
verify(fixture.regionRoomCacheService, never()).releaseRebuildLock(anyString(), anyString());
}
@Test
void execute_shouldPublishShortNegativeCacheForEmptyRegion() {
DiscoveryFixture fixture = new DiscoveryFixture();
AppExtCommand cmd = fixture.stubCurrentRegionRequest();
when(fixture.regionRoomCacheService.getRegionsRoom("YOLOAE")).thenReturn(null);
when(fixture.regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString()))
.thenReturn(true);
when(fixture.regionRoomCacheService.releaseRebuildLock(anyString(), anyString()))
.thenReturn(true);
fixture.stubEmptyRegionOrigin();
assertEquals(List.of(), fixture.exe.execute(cmd));
verify(fixture.regionRoomCacheService)
.saveRegionsRoom("[]", "YOLOAE", 30L, TimeUnit.SECONDS);
verify(fixture.regionRoomCacheService).releaseRebuildLock(anyString(), anyString());
}
@Test
void execute_shouldFailOpenToOriginWhenDiscoveryRedisIsUnavailable() {
DiscoveryFixture fixture = new DiscoveryFixture();
AppExtCommand cmd = fixture.stubCurrentRegionRequest();
when(fixture.regionRoomCacheService.getRegionsRoom("YOLOAE"))
.thenThrow(new IllegalStateException("redis get timeout"));
when(fixture.regionRoomCacheService.tryAcquireRebuildLock(anyString(), anyString()))
.thenThrow(new IllegalStateException("redis setnx timeout"));
fixture.stubEmptyRegionOrigin();
assertEquals(List.of(), fixture.exe.execute(cmd));
verify(fixture.activeVoiceRoomService).listByRegion("YOLO", "AE", 50);
verify(fixture.emptyRoomCacheReader).list("YOLO", "AE", false, 50);
verify(fixture.regionRoomCacheService, never()).releaseRebuildLock(anyString(), anyString());
}
/**
* 只为 singleflight 分支集中装配依赖避免测试中的大量 mock 掩盖真正的回源边界
*/
private static final class DiscoveryFixture {
private final UserSVipGateway userSVipGateway = mock(UserSVipGateway.class);
private final UserRegionGateway userRegionGateway = mock(UserRegionGateway.class);
private final RoomRegionFilterCommon roomRegionFilterCommon = mock(RoomRegionFilterCommon.class);
private final RoomVoiceProfileCommon roomVoiceProfileCommon = mock(RoomVoiceProfileCommon.class);
private final ActiveVoiceRoomService activeVoiceRoomService = mock(ActiveVoiceRoomService.class);
private final RoomUserBlacklistService roomUserBlacklistService =
mock(RoomUserBlacklistService.class);
private final LatestMobileDeviceService latestMobileDeviceService =
mock(LatestMobileDeviceService.class);
private final RoomProfileManagerService roomProfileManagerService =
mock(RoomProfileManagerService.class);
private final RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class);
private final LiveMicClient liveMicClient = mock(LiveMicClient.class);
private final EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class);
private final RocketStatusCacheService rocketStatusCacheService =
mock(RocketStatusCacheService.class);
private final GameLudoService gameLudoService = mock(GameLudoService.class);
private final UserProfileGateway userProfileGateway = mock(UserProfileGateway.class);
private final CountryCodeAliasSupport countryCodeAliasSupport =
mock(CountryCodeAliasSupport.class);
private final RegionConfig currentRegion =
new RegionConfig().setRegionCode("AE").setCountryCodes("AE,SA");
private final RoomVoiceDiscoverQryExe exe = new RoomVoiceDiscoverQryExe(
userSVipGateway,
userRegionGateway,
roomRegionFilterCommon,
roomVoiceProfileCommon,
activeVoiceRoomService,
roomUserBlacklistService,
latestMobileDeviceService,
roomProfileManagerService,
regionRoomCacheService,
liveMicClient,
emptyRoomCacheReader,
rocketStatusCacheService,
gameLudoService,
userProfileGateway,
countryCodeAliasSupport
);
private AppExtCommand stubCurrentRegionRequest() {
when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion);
when(userProfileGateway.getByUserId(1L)).thenReturn(new UserProfile());
AppExtCommand cmd = new AppExtCommand();
cmd.setReqUserId(1L);
cmd.setReqSysOrigin(ReqSysOrigin.of("YOLO", "YOLO"));
return cmd;
}
private void stubEmptyRegionOrigin() {
when(activeVoiceRoomService.listByRegion("YOLO", "AE", 50)).thenReturn(List.of());
when(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(), "AE"))
.thenReturn(List.of());
when(emptyRoomCacheReader.list("YOLO", "AE", false, 50)).thenReturn(List.of());
when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of())).thenReturn(List.of());
when(roomRegionFilterCommon.listCountryCodes(currentRegion)).thenReturn(List.of("AE", "SA"));
when(roomProfileManagerService.listSelectiveLimitByCountryCodes(
"YOLO", List.of("AE", "SA"), List.of(), 50
)).thenReturn(List.of());
when(roomRegionFilterCommon.filterRoomProfileManagersByRegion(List.of(), "AE"))
.thenReturn(List.of());
when(roomVoiceProfileCommon.listRoomVoiceProfilesV2(List.of())).thenReturn(List.of());
}
}
} }

View File

@ -0,0 +1,21 @@
package com.red.circle.other.infra.database.cache.service.other;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import java.util.List;
/**
* 按系统和区域读取发现页的空房间补充数据
*/
public interface EmptyRoomCacheReader {
/**
* 从单个有界索引批量读取空房间不允许退回全库 SCAN/KEYS
*
* @param sysOrigin 业务系统
* @param region 当前区域
* @param allRegions 是否读取同系统的全区域聚合索引
* @param limit 最大返回数量
* @return 有效空房间快照
*/
List<ActiveVoiceRoom> list(String sysOrigin, String region, boolean allRegions, int limit);
}

View File

@ -0,0 +1,176 @@
package com.red.circle.other.infra.database.cache.service.other.impl;
import com.alibaba.fastjson.JSON;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.other.infra.database.cache.service.other.EmptyRoomCacheReader;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.inner.model.cache.EmptyRoomCacheKeys;
import com.red.circle.tool.core.collection.CollectionUtils;
import com.red.circle.tool.core.text.StringUtils;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
/**
* 基于 ZSet 到期索引批量加载空房间快照
*
* <p> reader 不允许回退 legacy 全库 SCAN滚动发布必须先全量升级 live writer等待旧快照
* 十分钟窗口收敛后再升级 other确保任何在途 legacy-only payload 都已过期</p>
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class EmptyRoomCacheReaderImpl implements EmptyRoomCacheReader {
private final RedisService redisService;
@Override
public List<ActiveVoiceRoom> list(String sysOrigin, String region, boolean allRegions, int limit) {
if (StringUtils.isBlank(sysOrigin)
|| (!allRegions && StringUtils.isBlank(region))
|| limit <= 0) {
return new ArrayList<>();
}
String indexKey = allRegions
? EmptyRoomCacheKeys.allRegionsIndex(sysOrigin)
: EmptyRoomCacheKeys.regionIndex(sysOrigin, region);
try {
long nowMillis = System.currentTimeMillis();
// 过期成员在读取前先删除索引开销只与该系统/区域的空房数量相关
redisService.zsetRemoveByScore(indexKey, 0D, nowMillis);
// score 越大代表快照剩余窗口越长 rank Redis 端截断网络与 MGET 都最多处理 limit
Set<Object> indexedMembers = redisService.zsetGetReverseRange(indexKey, 0L, limit - 1L);
if (CollectionUtils.isEmpty(indexedMembers)) {
return new ArrayList<>();
}
List<String> roomMembers = indexedMembers.stream()
.filter(Objects::nonNull)
.map(Objects::toString)
.filter(StringUtils::isNotBlank)
.distinct()
.toList();
if (CollectionUtils.isEmpty(roomMembers)) {
return new ArrayList<>();
}
Set<String> staleMembers = new LinkedHashSet<>();
List<String> validMembers = new ArrayList<>();
List<Long> roomIds = new ArrayList<>();
for (String member : roomMembers) {
Long roomId = parseRoomId(member);
if (Objects.isNull(roomId)) {
staleMembers.add(member);
continue;
}
validMembers.add(member);
roomIds.add(roomId);
}
if (CollectionUtils.isEmpty(validMembers)) {
// 全部成员都损坏时不要向 Redis 发送空 MGET先清掉当前索引避免每次发现请求重复解析
redisService.zsetRemove(indexKey, staleMembers.toArray());
return new ArrayList<>();
}
List<String> payloadKeys = roomIds.stream()
.map(EmptyRoomCacheKeys::payload)
.toList();
List<String> payloads = Objects.requireNonNullElse(
redisService.multiGetString(payloadKeys),
List.of()
);
List<ActiveVoiceRoom> rooms = new ArrayList<>();
for (int index = 0; index < validMembers.size(); index++) {
String member = validMembers.get(index);
String payload = index < payloads.size() ? payloads.get(index) : null;
ActiveVoiceRoom room = parseRoom(
payload,
member,
roomIds.get(index),
payloadKeys.get(index),
staleMembers
);
if (Objects.isNull(room)
|| !matchesIndex(room, sysOrigin, region, allRegions)) {
staleMembers.add(member);
continue;
}
normalizeEmptyRoom(room);
rooms.add(room);
}
if (CollectionUtils.isNotEmpty(staleMembers)) {
// payload 已过期损坏或索引归属错误时只清理当前小索引绝不回退全库扫描
redisService.zsetRemove(indexKey, staleMembers.toArray());
}
return rooms;
} catch (RuntimeException e) {
// 索引是发现页补充数据异常时返回活跃房间主链路结果而不是放大 Redis 故障
log.error("读取空房间索引失败, indexKey={}", indexKey, e);
return new ArrayList<>();
}
}
private ActiveVoiceRoom parseRoom(
String payload,
String member,
Long expectedRoomId,
String payloadKey,
Set<String> staleMembers
) {
if (StringUtils.isBlank(payload)) {
staleMembers.add(member);
return null;
}
try {
ActiveVoiceRoom room = JSON.parseObject(payload, ActiveVoiceRoom.class);
if (Objects.isNull(room) || !Objects.equals(room.getId(), expectedRoomId)) {
staleMembers.add(member);
return null;
}
return room;
} catch (RuntimeException e) {
// payload 对新旧 reader 都不可用删除后允许后续退房重新生成一致快照
staleMembers.add(member);
redisService.delete(payloadKey);
log.warn("清理无法解析的空房间快照, payloadKey={}", payloadKey, e);
return null;
}
}
private boolean matchesIndex(
ActiveVoiceRoom room,
String sysOrigin,
String region,
boolean allRegions
) {
return Objects.equals(sysOrigin, room.getSysOrigin())
&& (allRegions || Objects.equals(region, room.getRegion()));
}
private void normalizeEmptyRoom(ActiveVoiceRoom room) {
room.setOnlineQuantity(0L);
if (Objects.isNull(room.getWeights())) {
room.setWeights(0);
}
if (Objects.isNull(room.getFixedWeights())) {
room.setFixedWeights(0);
}
}
private Long parseRoomId(String member) {
try {
return Long.valueOf(member);
} catch (NumberFormatException e) {
log.warn("清理非法空房间索引成员, member={}", member);
return null;
}
}
}

View File

@ -1,11 +1,8 @@
package com.red.circle.other.infra.database.cache.service.user; package com.red.circle.other.infra.database.cache.service.user;
import com.red.circle.other.infra.database.cache.key.user.UserHashKey;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom; import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import java.util.Collection;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.concurrent.TimeUnit;
/** /**
* 区域房间缓存服务. * 区域房间缓存服务.
@ -37,6 +34,16 @@ public interface RegionRoomCacheService {
*/ */
void saveRegionsRoom(String jsonString, String key); void saveRegionsRoom(String jsonString, String key);
/**
* 使用指定 TTL 写入发现页缓存空结果使用短 TTL 负缓存避免所有请求持续回源
*
* @param jsonString 房间信息
* @param key 发现页缓存逻辑 key
* @param timeout 缓存时长
* @param timeUnit 时长单位
*/
void saveRegionsRoom(String jsonString, String key, long timeout, TimeUnit timeUnit);
/** /**
* 获取区域房间信息. * 获取区域房间信息.
* *
@ -44,4 +51,22 @@ public interface RegionRoomCacheService {
*/ */
String getRegionsRoom(String key); String getRegionsRoom(String key);
/**
* 竞争指定发现缓存 key 的跨实例重建权
*
* @param key 发现页缓存逻辑 key
* @param token 本次持有者的唯一 token
* @return 是否获得锁
*/
boolean tryAcquireRebuildLock(String key, String token);
/**
* 仅当 token 仍属于当前持有者时释放锁防止旧请求删除超时后新持有者的锁
*
* @param key 发现页缓存逻辑 key
* @param token 本次持有者的唯一 token
* @return 是否删除了当前锁
*/
boolean releaseRebuildLock(String key, String token);
} }

View File

@ -2,19 +2,16 @@ package com.red.circle.other.infra.database.cache.service.user.impl;
import com.google.gson.Gson; import com.google.gson.Gson;
import com.red.circle.component.redis.service.RedisService; import com.red.circle.component.redis.service.RedisService;
import com.red.circle.other.infra.database.cache.key.user.UserHashKey;
import com.red.circle.other.infra.database.cache.key.user.UserKey; import com.red.circle.other.infra.database.cache.key.user.UserKey;
import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService; import com.red.circle.other.infra.database.cache.service.user.RegionRoomCacheService;
import com.red.circle.other.infra.database.cache.service.user.UserRegionCacheService;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom; import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.tool.core.collection.CollectionUtils; import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.util.*;
import java.util.concurrent.TimeUnit;
/** /**
* 区域房间缓存服务. * 区域房间缓存服务.
* *
@ -25,9 +22,16 @@ import java.util.concurrent.TimeUnit;
@RequiredArgsConstructor @RequiredArgsConstructor
public class RegionRoomCacheServiceImpl implements RegionRoomCacheService { public class RegionRoomCacheServiceImpl implements RegionRoomCacheService {
private static final String REBUILD_LOCK_PREFIX = "discovery_cache_rebuild_lock:v2:";
private static final String RELEASE_LOCK_SCRIPT =
"if redis.call('GET', KEYS[1]) == ARGV[1] then "
+ "return redis.call('DEL', KEYS[1]) "
+ "end "
+ "return 0";
private static final long REBUILD_LOCK_SECONDS = 30L;
private static final long CACHE_TIME_MINUTES = 10L;
private final RedisService redisService; private final RedisService redisService;
private static final long CACHE_DAYS = 3;
private static final long CACHE_TIME = 10;
@Override @Override
public void remove(String key) { public void remove(String key) {
@ -39,12 +43,22 @@ public class RegionRoomCacheServiceImpl implements RegionRoomCacheService {
Gson gson = new Gson(); Gson gson = new Gson();
String jsonString = gson.toJson(activeVoiceRooms); String jsonString = gson.toJson(activeVoiceRooms);
log.warn("jsonString{}", jsonString); log.warn("jsonString{}", jsonString);
redisService.setIfAbsent(getRegionsRoomKey(key), jsonString, CACHE_TIME, TimeUnit.MINUTES); redisService.setIfAbsent(
getRegionsRoomKey(key),
jsonString,
CACHE_TIME_MINUTES,
TimeUnit.MINUTES
);
} }
@Override @Override
public void saveRegionsRoom(String jsonString, String key) { public void saveRegionsRoom(String jsonString, String key) {
redisService.setIfAbsent(getRegionsRoomKey(key), jsonString, CACHE_TIME, TimeUnit.MINUTES); saveRegionsRoom(jsonString, key, CACHE_TIME_MINUTES, TimeUnit.MINUTES);
}
@Override
public void saveRegionsRoom(String jsonString, String key, long timeout, TimeUnit timeUnit) {
redisService.setIfAbsent(getRegionsRoomKey(key), jsonString, timeout, timeUnit);
} }
@Override @Override
@ -52,8 +66,33 @@ public class RegionRoomCacheServiceImpl implements RegionRoomCacheService {
return redisService.getString(getRegionsRoomKey(key)); return redisService.getString(getRegionsRoomKey(key));
} }
@Override
public boolean tryAcquireRebuildLock(String key, String token) {
return redisService.setIfAbsent(
getRebuildLockKey(key),
token,
REBUILD_LOCK_SECONDS,
TimeUnit.SECONDS
);
}
@Override
public boolean releaseRebuildLock(String key, String token) {
Long removed = redisService.execute(
RELEASE_LOCK_SCRIPT,
Long.class,
Collections.singletonList(getRebuildLockKey(key)),
token
);
return Long.valueOf(1L).equals(removed);
}
private String getRegionsRoomKey(String userId) { private String getRegionsRoomKey(String userId) {
return UserKey.REGION_ROOM.getKey(userId); return UserKey.REGION_ROOM.getKey(userId);
} }
private String getRebuildLockKey(String key) {
return REBUILD_LOCK_PREFIX + key;
}
} }

View File

@ -0,0 +1,124 @@
package com.red.circle.other.infra.database.cache.service.other.impl;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyDouble;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.alibaba.fastjson.JSON;
import com.red.circle.component.redis.service.RedisService;
import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom;
import com.red.circle.other.inner.model.cache.EmptyRoomCacheKeys;
import java.util.Arrays;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
class EmptyRoomCacheReaderImplTest {
private RedisService redisService;
private EmptyRoomCacheReaderImpl reader;
@BeforeEach
void setUp() {
redisService = mock(RedisService.class);
reader = new EmptyRoomCacheReaderImpl(redisService);
}
@Test
void list_shouldReadRegionIndexNormalizeDefaultsAndRemoveStaleMembers() {
String indexKey = EmptyRoomCacheKeys.regionIndex("YOLO", "AE");
Set<Object> indexedMembers = new LinkedHashSet<>(
List.of("101", "102", "invalid-room-id", "103", "104"));
when(redisService.zsetGetReverseRange(indexKey, 0L, 49L))
.thenReturn(indexedMembers);
ActiveVoiceRoom validRoom = room(101L, "YOLO", "AE")
.setOnlineQuantity(99L)
.setWeights(null)
.setFixedWeights(null);
ActiveVoiceRoom wrongRegionRoom = room(104L, "YOLO", "SA");
List<String> payloadKeys = List.of(
EmptyRoomCacheKeys.payload(101L),
EmptyRoomCacheKeys.payload(102L),
EmptyRoomCacheKeys.payload(103L),
EmptyRoomCacheKeys.payload(104L)
);
// MGET 返回值与索引中的合法房间 ID 一一对齐null JSON 和归属错误都应淘汰
when(redisService.multiGetString(payloadKeys)).thenReturn(Arrays.asList(
JSON.toJSONString(validRoom),
null,
"{invalid-json",
JSON.toJSONString(wrongRegionRoom)
));
List<ActiveVoiceRoom> result = reader.list("YOLO", "AE", false, 50);
assertEquals(1, result.size());
assertEquals(validRoom.getId(), result.get(0).getId());
assertEquals(0L, result.get(0).getOnlineQuantity());
assertEquals(0, result.get(0).getWeights());
assertEquals(0, result.get(0).getFixedWeights());
verify(redisService).zsetRemoveByScore(eq(indexKey), eq(0D), anyDouble());
verify(redisService).multiGetString(payloadKeys);
verify(redisService).delete(EmptyRoomCacheKeys.payload(103L));
ArgumentCaptor<Object[]> staleMembersCaptor = ArgumentCaptor.forClass(Object[].class);
verify(redisService).zsetRemove(eq(indexKey), staleMembersCaptor.capture());
assertEquals(
Set.of("102", "invalid-room-id", "103", "104"),
Set.copyOf(Arrays.asList(staleMembersCaptor.getValue()))
);
}
@Test
void list_shouldUseAllRegionsIndexWithoutRejectingRoomRegion() {
String indexKey = EmptyRoomCacheKeys.allRegionsIndex("YOLO");
when(redisService.zsetGetReverseRange(indexKey, 0L, 49L))
.thenReturn(new LinkedHashSet<>(List.of("201")));
ActiveVoiceRoom saRoom = room(201L, "YOLO", "SA")
.setOnlineQuantity(7L)
.setWeights(8)
.setFixedWeights(9);
when(redisService.multiGetString(List.of(EmptyRoomCacheKeys.payload(201L))))
.thenReturn(List.of(JSON.toJSONString(saRoom)));
List<ActiveVoiceRoom> result = reader.list("YOLO", "AE", true, 50);
assertEquals(1, result.size());
assertEquals("SA", result.get(0).getRegion());
// 空房快照的在线人数必须归零已有排序权重则保持写入值
assertEquals(0L, result.get(0).getOnlineQuantity());
assertEquals(8, result.get(0).getWeights());
assertEquals(9, result.get(0).getFixedWeights());
verify(redisService).zsetRemoveByScore(eq(indexKey), eq(0D), anyDouble());
verify(redisService).zsetGetReverseRange(indexKey, 0L, 49L);
verify(redisService, never()).zsetRemove(eq(indexKey), org.mockito.ArgumentMatchers.any());
}
@Test
void list_shouldRemoveAllInvalidMembersWithoutEmptyMultiGet() {
String indexKey = EmptyRoomCacheKeys.regionIndex("YOLO", "AE");
when(redisService.zsetGetReverseRange(indexKey, 0L, 49L))
.thenReturn(new LinkedHashSet<>(List.of("invalid-room-id")));
assertEquals(List.of(), reader.list("YOLO", "AE", false, 50));
verify(redisService, never()).multiGetString(any());
verify(redisService).zsetRemove(indexKey, "invalid-room-id");
}
private ActiveVoiceRoom room(Long roomId, String sysOrigin, String region) {
return new ActiveVoiceRoom()
.setId(roomId)
.setSysOrigin(sysOrigin)
.setRegion(region);
}
}

View File

@ -0,0 +1,92 @@
package com.red.circle.other.infra.database.cache.service.user.impl;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import com.red.circle.component.redis.service.RedisService;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class RegionRoomCacheServiceImplTest {
private static final String RELEASE_LOCK_SCRIPT =
"if redis.call('GET', KEYS[1]) == ARGV[1] then "
+ "return redis.call('DEL', KEYS[1]) "
+ "end "
+ "return 0";
private static final String CACHE_KEY = "USER:REGION_ROOM:LIKEISA";
private static final String REBUILD_LOCK_KEY = "discovery_cache_rebuild_lock:v2:LIKEISA";
private RedisService redisService;
private RegionRoomCacheServiceImpl cacheService;
@BeforeEach
void setUp() {
redisService = mock(RedisService.class);
cacheService = new RegionRoomCacheServiceImpl(redisService);
}
@Test
void saveRegionsRoom_shouldPassThroughCustomCacheTtl() {
// 空结果使用调用方指定的短 TTL不能退回默认十分钟而持续放大回源窗口
cacheService.saveRegionsRoom("[]", "LIKEISA", 45L, TimeUnit.SECONDS);
verify(redisService).setIfAbsent(CACHE_KEY, "[]", 45L, TimeUnit.SECONDS);
}
@Test
void tryAcquireRebuildLock_shouldStoreOwnerTokenForThirtySeconds() {
when(redisService.setIfAbsent(REBUILD_LOCK_KEY, "owner-token", 30L, TimeUnit.SECONDS))
.thenReturn(true);
boolean acquired = cacheService.tryAcquireRebuildLock("LIKEISA", "owner-token");
assertTrue(acquired);
verify(redisService).setIfAbsent(
REBUILD_LOCK_KEY,
"owner-token",
30L,
TimeUnit.SECONDS
);
}
@Test
void releaseRebuildLock_shouldAtomicallyCompareOwnerTokenBeforeDelete() {
when(redisService.execute(
RELEASE_LOCK_SCRIPT,
Long.class,
Collections.singletonList(REBUILD_LOCK_KEY),
"owner-token"
)).thenReturn(1L);
boolean released = cacheService.releaseRebuildLock("LIKEISA", "owner-token");
assertTrue(released);
verify(redisService).execute(
RELEASE_LOCK_SCRIPT,
Long.class,
Collections.singletonList(REBUILD_LOCK_KEY),
"owner-token"
);
}
@Test
void releaseRebuildLock_shouldReportTokenMismatchWithoutClaimingRelease() {
// Lua 返回 0 代表锁已过期或 owner 已变化旧请求不得声称释放成功
when(redisService.execute(
RELEASE_LOCK_SCRIPT,
Long.class,
Collections.singletonList(REBUILD_LOCK_KEY),
"expired-owner-token"
)).thenReturn(0L);
boolean released = cacheService.releaseRebuildLock("LIKEISA", "expired-owner-token");
assertFalse(released);
}
}