diff --git a/rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/model/cache/EmptyRoomCacheKeys.java b/rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/model/cache/EmptyRoomCacheKeys.java new file mode 100644 index 00000000..f52059da --- /dev/null +++ b/rc-service/rc-inner-api/other-inner/other-inner-model/src/main/java/com/red/circle/other/inner/model/cache/EmptyRoomCacheKeys.java @@ -0,0 +1,35 @@ +package com.red.circle.other.inner.model.cache; + +/** + * 空房间缓存的跨服务 Redis key 约定。 + * + *

live-service 负责写入空房间快照，other-service 负责消费索引生成发现页。key 只能在这个 + * 共享模型中定义，避免两个服务独立拼接后发生不可见的数据分叉。

+ */ +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; + } +} diff --git a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java index dbf0be88..ea2e68ca 100644 --- a/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java +++ b/rc-service/rc-service-live/live-application/src/main/java/com/red/circle/live/app/command/user/LiveQuitRoomExe.java @@ -1,12 +1,12 @@ 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.component.redis.service.RedisService; -import com.red.circle.framework.dto.ResultResponse; -import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; -import com.red.circle.live.domain.live.LiveMicrophone; -import com.red.circle.live.domain.live.LiveMusicStatus; +import com.red.circle.framework.dto.ResultResponse; +import com.red.circle.live.domain.gateway.LiveMicrophoneGateway; +import com.red.circle.live.domain.live.LiveMicrophone; +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.LiveMusicHeartbeatService; 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 java.util.List; import java.util.Objects; -import java.util.concurrent.TimeUnit; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; @@ -37,10 +36,8 @@ public class LiveQuitRoomExe { private final ActiveVoiceRoomClient activeVoiceRoomClient; private final RedisService redisService; private final UserAgoraTokenCacheService userAgoraTokenCacheService; - - private static final String EMPTY_ROOM_CACHE_KEY = "empty_room_cache:"; - private static final int CACHE_EXPIRE_MINUTES = 10; - + private final EmptyRoomCacheWriter emptyRoomCacheWriter; + public void execute(AppRoomIdCmd cmd) { Long roomId = cmd.getRoomId(); Long userId = cmd.requiredReqUserId(); @@ -94,13 +91,8 @@ public class LiveQuitRoomExe { ResultResponse noBodyRoom = activeVoiceRoomClient.createNoBodyRoom(roomId); ActiveVoiceRoomCO body = noBodyRoom.getBody(); if (Objects.nonNull(body)) { - String cacheKey = EMPTY_ROOM_CACHE_KEY + roomId; - redisService.setIfAbsent( - cacheKey, - JSON.toJSONString(body), - CACHE_EXPIRE_MINUTES, - TimeUnit.MINUTES - ); + // 新 live 同时维护 legacy payload 与 ZSet，使它可以先于 other 发布且不影响旧 reader。 + emptyRoomCacheWriter.save(body); } } catch (Exception e) { log.error("quitRoom cacheEmptyRoom error roomId={}, userId={}", roomId, userId, e); diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/EmptyRoomCacheWriter.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/EmptyRoomCacheWriter.java new file mode 100644 index 00000000..0880fcbe --- /dev/null +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/EmptyRoomCacheWriter.java @@ -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); +} diff --git a/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/EmptyRoomCacheWriterImpl.java b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/EmptyRoomCacheWriterImpl.java new file mode 100644 index 00000000..cf95a8f9 --- /dev/null +++ b/rc-service/rc-service-live/live-infrastructure/src/main/java/com/red/circle/live/infra/database/cache/service/impl/EmptyRoomCacheWriterImpl.java @@ -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 到期索引”维护空房间。 + * + *

ZSet score 是快照实际过期毫秒。这样 Set/Hash 不具备的成员级 TTL 可以在读取时通过 + * ZREMRANGEBYSCORE 有界清理，同时发现页不再遍历整个 Redis DB。

+ */ +@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()); + } +} diff --git a/rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/EmptyRoomCacheWriterImplTest.java b/rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/EmptyRoomCacheWriterImplTest.java new file mode 100644 index 00000000..513821d0 --- /dev/null +++ b/rc-service/rc-service-live/live-infrastructure/src/test/java/com/red/circle/live/infra/database/cache/service/impl/EmptyRoomCacheWriterImplTest.java @@ -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 { + + private final List payloadWrites = new ArrayList<>(); + private final List indexWrites = new ArrayList<>(); + private final List indexCleanupCalls = new ArrayList<>(); + private final List expireCalls = new ArrayList<>(); + private final List events = new ArrayList<>(); + private final Queue remainingTtls = new ArrayDeque<>(); + private RuntimeException payloadWriteFailure; + private RuntimeException indexWriteFailure; + private int payloadWriteAttempts; + private int indexWriteAttempts; + private int getExpireCalls; + + @Override + @SuppressWarnings("unchecked") + public ValueOperations 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 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 proxy(Class 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); + } +} diff --git a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java index 5c5e9ca5..34b3cdb0 100644 --- a/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java +++ b/rc-service/rc-service-other/other-application/src/main/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExe.java @@ -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.rocket.RocketStatus; 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.mongo.entity.live.ActiveVoiceRoom; import com.red.circle.other.infra.database.mongo.entity.live.RoomProfileManager; @@ -26,18 +27,15 @@ import com.red.circle.other.infra.database.mongo.service.live.RoomProfileManager import com.red.circle.other.infra.database.mongo.service.live.RoomUserBlacklistService; 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.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 java.util.*; +import java.util.concurrent.TimeUnit; 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; /** @@ -57,22 +55,23 @@ public class RoomVoiceDiscoverQryExe { private final ActiveVoiceRoomService activeVoiceRoomService; private final RoomUserBlacklistService roomUserBlacklistService; private final LatestMobileDeviceService latestMobileDeviceService; - private final RoomProfileManagerService roomProfileManagerService; - private final RegionRoomCacheService regionRoomCacheService; - private final LiveMicClient liveMicClient; - private final RedisTemplate redisTemplate; + private final RoomProfileManagerService roomProfileManagerService; + private final RegionRoomCacheService regionRoomCacheService; + private final LiveMicClient liveMicClient; + private final EmptyRoomCacheReader emptyRoomCacheReader; private final RocketStatusCacheService rocketStatusCacheService; private final GameLudoService gameLudoService; private final UserProfileGateway userProfileGateway; 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 FIXED_WEIGHTS_EXT_KEY = "fixedWeights"; 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; + 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 execute(AppExtCommand cmd) { long totalStart = System.currentTimeMillis(); @@ -87,99 +86,160 @@ public class RoomVoiceDiscoverQryExe { stepStart = logDiscoveryStep("prepare", stepStart, reqUserId, sysOrigin, region); String key = discoveryCacheKey(sysOrigin, region, allRegionWhiteListUser); - String regionsRoom = regionRoomCacheService.getRegionsRoom(key); + List cachedRooms = readDiscoveryCache(key); stepStart = logDiscoveryStep("read-region-cache", stepStart, reqUserId, sysOrigin, region); - if (Objects.nonNull(regionsRoom)) { - List roomList = JSON.parseArray(regionsRoom, RoomVoiceProfileCO.class); - if (CollectionUtils.isEmpty(roomList)) { - logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, true, 0); + if (Objects.nonNull(cachedRooms)) { + stepStart = logDiscoveryStep("filter-cache", stepStart, reqUserId, sysOrigin, region); + logDiscoveryTotal( + totalStart, + reqUserId, + sysOrigin, + region, + allRegionWhiteListUser, + true, + cachedRooms.size() + ); + return cachedRooms; + } + + String rebuildToken = UUID.randomUUID().toString(); + 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 publishedRooms = waitForPublishedDiscoveryCache(key); + stepStart = logDiscoveryStep("wait-region-cache", stepStart, reqUserId, sysOrigin, region); + if (Objects.nonNull(publishedRooms)) { + logDiscoveryTotal( + totalStart, + reqUserId, + sysOrigin, + region, + allRegionWhiteListUser, + true, + publishedRooms.size() + ); + 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(); } - List availableRooms = roomVoiceProfileCommon - .filterAvailableRoomProfiles(roomList); - if (availableRooms.size() != roomList.size()) { - regionRoomCacheService.remove(key); - } - stepStart = logDiscoveryStep("filter-cache", stepStart, reqUserId, sysOrigin, region); - logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, true, availableRooms.size()); - return availableRooms; } - List activeVoiceRooms = allRegionWhiteListUser - ? activeVoiceRoomService.listDiscover( + try { + if (rebuildLockAcquired) { + // 获取锁后必须二次检查，覆盖“首次 miss 与 SETNX 之间已有 leader 发布”的竞态窗口。 + List doubleCheckedRooms = readDiscoveryCache(key); + stepStart = logDiscoveryStep( + "double-check-region-cache", + stepStart, + reqUserId, sysOrigin, - true, - region, - Optional.ofNullable(userProfile).map(UserProfile::getCountryCode).orElse(null), - DISCOVERY_LIMIT - ) - : roomRegionFilterCommon.filterActiveVoiceRoomsByRegion( - activeVoiceRoomService.listByRegion(sysOrigin, region, DISCOVERY_LIMIT), 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 emptyRooms = getEmptyRoomsFromCache(); - stepStart = logDiscoveryStep("scan-empty-room-cache", stepStart, reqUserId, sysOrigin, region); - if (CollectionUtils.isNotEmpty(emptyRooms)) { - List matchedEmptyRooms = allRegionWhiteListUser - ? emptyRooms.stream() - .filter(room -> Objects.equals(sysOrigin, room.getSysOrigin())) - .toList() + List activeVoiceRooms = allRegionWhiteListUser + ? activeVoiceRoomService.listDiscover( + sysOrigin, + true, + region, + Optional.ofNullable(userProfile).map(UserProfile::getCountryCode).orElse(null), + DISCOVERY_LIMIT + ) : roomRegionFilterCommon.filterActiveVoiceRoomsByRegion( - emptyRooms.stream() - .filter(room -> Objects.equals(sysOrigin, room.getSysOrigin())) - .toList(), + activeVoiceRoomService.listByRegion(sysOrigin, region, DISCOVERY_LIMIT), region ); - if (CollectionUtils.isNotEmpty(matchedEmptyRooms)) { - List allRooms = new ArrayList<>(activeVoiceRooms); - Set existingRoomIds = activeVoiceRooms.stream() - .map(ActiveVoiceRoom::getId) - .collect(Collectors.toSet()); - matchedEmptyRooms.stream() - .filter(room -> !existingRoomIds.contains(room.getId())) - .forEach(allRooms::add); - activeVoiceRooms = allRooms; - } - } - stepStart = logDiscoveryStep("merge-empty-rooms", stepStart, reqUserId, sysOrigin, region); + stepStart = logDiscoveryStep("query-active-rooms", stepStart, reqUserId, sysOrigin, region); - List roomList = roomVoiceProfileCommon.toListRoomVoiceProfileCO(activeVoiceRooms); - stepStart = logDiscoveryStep("build-profile-list", stepStart, reqUserId, sysOrigin, region); - if (roomList.size() < DISCOVERY_LIMIT) { - List fallbackRoomManagersSource = roomProfileManagerService.listSelectiveLimitByCountryCodes( + // 索引已经按系统/区域收敛，读取成本只与当前候选空房数量相关，不再遍历 DB0 全量 key。 + List emptyRooms = emptyRoomCacheReader.list( sysOrigin, - allRegionWhiteListUser ? List.of() : roomRegionFilterCommon.listCountryCodes(currentRegion), - roomList.stream().map(RoomVoiceProfileCO::getId).toList(), + region, + allRegionWhiteListUser, DISCOVERY_LIMIT ); - List fallbackRoomManagers = allRegionWhiteListUser - ? fallbackRoomManagersSource - : roomRegionFilterCommon.filterRoomProfileManagersByRegion( - fallbackRoomManagersSource, - region - ); - roomList = mergeRoomList(roomList, roomVoiceProfileCommon.listRoomVoiceProfilesV2(fallbackRoomManagers), - DISCOVERY_LIMIT); - } - stepStart = logDiscoveryStep("fallback-room-profiles", stepStart, reqUserId, sysOrigin, region); - roomList = sortByTopWeightAndMemberQuantityDesc(roomList); - stepStart = logDiscoveryStep("sort-rooms", stepStart, reqUserId, sysOrigin, region); + stepStart = logDiscoveryStep("read-empty-room-index", stepStart, reqUserId, sysOrigin, region); + if (CollectionUtils.isNotEmpty(emptyRooms)) { + List allRooms = new ArrayList<>(activeVoiceRooms); + Set existingRoomIds = activeVoiceRooms.stream() + .map(ActiveVoiceRoom::getId) + .collect(Collectors.toSet()); + emptyRooms.stream() + .filter(room -> !existingRoomIds.contains(room.getId())) + .forEach(allRooms::add); + activeVoiceRooms = allRooms; + } + stepStart = logDiscoveryStep("merge-empty-rooms", stepStart, reqUserId, sysOrigin, region); - if (CollectionUtils.isEmpty(roomList)) { - logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, 0); - return CollectionUtils.newArrayList(); - } - Map svipLevelEnumMap = userSVipGateway.checkSVipIdentity( - roomList.stream().map(RoomVoiceProfileCO::getUserId).collect( - Collectors.toSet())); - roomList = roomList.stream() - .peek(room -> { - room.setUserSVipLevel(svipLevelEnumMap.get(room.getUserId())); - }) - .toList(); + List roomList = roomVoiceProfileCommon.toListRoomVoiceProfileCO(activeVoiceRooms); + stepStart = logDiscoveryStep("build-profile-list", stepStart, reqUserId, sysOrigin, region); + if (roomList.size() < DISCOVERY_LIMIT) { + List fallbackRoomManagersSource = + roomProfileManagerService.listSelectiveLimitByCountryCodes( + sysOrigin, + allRegionWhiteListUser ? List.of() : roomRegionFilterCommon.listCountryCodes(currentRegion), + roomList.stream().map(RoomVoiceProfileCO::getId).toList(), + DISCOVERY_LIMIT + ); + List fallbackRoomManagers = allRegionWhiteListUser + ? fallbackRoomManagersSource + : roomRegionFilterCommon.filterRoomProfileManagersByRegion( + fallbackRoomManagersSource, + region + ); + roomList = mergeRoomList( + roomList, + roomVoiceProfileCommon.listRoomVoiceProfilesV2(fallbackRoomManagers), + DISCOVERY_LIMIT + ); + } + stepStart = logDiscoveryStep("fallback-room-profiles", stepStart, reqUserId, sysOrigin, region); + roomList = sortByTopWeightAndMemberQuantityDesc(roomList); + stepStart = logDiscoveryStep("sort-rooms", stepStart, reqUserId, sysOrigin, region); + + if (CollectionUtils.isEmpty(roomList)) { + // 短负缓存让空区域的 follower 也能结束等待，但不会把新开房间隐藏十分钟。 + saveDiscoveryCache(key, roomList, reqUserId, sysOrigin, region); + logDiscoveryStep("write-empty-region-cache", stepStart, reqUserId, sysOrigin, region); + logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, 0); + return CollectionUtils.newArrayList(); + } + Map svipLevelEnumMap = userSVipGateway.checkSVipIdentity( + roomList.stream().map(RoomVoiceProfileCO::getUserId).collect(Collectors.toSet())); + roomList = roomList.stream() + .peek(room -> room.setUserSVipLevel(svipLevelEnumMap.get(room.getUserId()))) + .toList(); stepStart = logDiscoveryStep("fill-svip", stepStart, reqUserId, sysOrigin, region); // 填充火箭状态 @@ -192,7 +252,12 @@ public class RoomVoiceDiscoverQryExe { logDiscoveryStep("write-region-cache", stepStart, reqUserId, sysOrigin, region); logDiscoveryTotal(totalStart, reqUserId, sysOrigin, region, allRegionWhiteListUser, false, roomList.size()); - return roomList; + return roomList; + } finally { + if (rebuildLockAcquired) { + releaseRebuildLock(key, rebuildToken, reqUserId); + } + } } private List sortByTopWeightAndMemberQuantityDesc( @@ -236,13 +301,105 @@ public class RoomVoiceDiscoverQryExe { return allRegionWhiteListUser ? sysOrigin + region + ":ALL_REGION" : sysOrigin + region; } - private void saveDiscoveryCache(String key, List roomList, Long userId, - String sysOrigin, String region) { - if (CollectionUtils.isEmpty(roomList)) { - return; + /** + * 读取并校验发现页缓存。返回 null 表示真正 miss；空 List 表示命中的负缓存。 + */ + private List readDiscoveryCache(String key) { + return readDiscoveryCache(key, true); + } + + private List 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 { - regionRoomCacheService.saveRegionsRoom(JSON.toJSONString(roomList), key); + List roomList = JSON.parseArray(regionsRoom, RoomVoiceProfileCO.class); + if (Objects.isNull(roomList)) { + removeDiscoveryCache(key); + return null; + } + if (CollectionUtils.isEmpty(roomList)) { + return CollectionUtils.newArrayList(); + } + + List 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 waitForPublishedDiscoveryCache(String key) { + for (int attempt = 0; attempt < REBUILD_WAIT_ATTEMPTS; attempt++) { + // follower 轮询期间抑制重复 GET 异常日志；初次读取和锁竞争仍会保留故障证据。 + List 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 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) { log.warn("写入房间发现缓存失败, key={}, userId={}, sysOrigin={}, region={}", key, userId, sysOrigin, region, e); @@ -410,88 +567,8 @@ public class RoomVoiceDiscoverQryExe { } } - /** - * 从缓存中获取没人的房间 - * - * @return - */ - private List getEmptyRoomsFromCache() { - try { - Set keys = scanEmptyRoomKeys(); - if (CollectionUtils.isEmpty(keys)) { - return new ArrayList<>(); - } - - List 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 scanEmptyRoomKeys() { - Set keys = new HashSet<>(); - ScanOptions options = ScanOptions.scanOptions() - .match(EMPTY_ROOM_CACHE_KEY) - .count(EMPTY_ROOM_SCAN_COUNT) - .build(); - try (var cursor = redisTemplate.scan(options)) { - if (Objects.isNull(cursor)) { - return keys; - } - while (cursor.hasNext()) { - keys.add(Objects.toString(cursor.next(), "")); - } - } - return keys; - } - /** - * 将ActiveVoiceRoomCO转换为ActiveVoiceRoom - * - * @param roomCO - * @return - */ - 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; - } - } - - /** - * 去除拉黑和被上锁的的房间 + * 去除拉黑和被上锁的的房间 * * @param roomIds * @return diff --git a/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExeTest.java b/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExeTest.java index 0d662565..6afd76fb 100644 --- a/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExeTest.java +++ b/rc-service/rc-service-other/other-application/src/test/java/com/red/circle/other/app/command/room/query/RoomVoiceDiscoverQryExeTest.java @@ -1,6 +1,7 @@ package com.red.circle.other.app.command.room.query; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.anySet; import static org.mockito.Mockito.mock; 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.ability.RegionConfig; 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.mongo.entity.live.ActiveVoiceRoom; 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 java.util.List; import java.util.Map; -import java.util.Set; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; -import org.springframework.data.redis.core.RedisTemplate; class RoomVoiceDiscoverQryExeTest { @@ -51,7 +52,7 @@ class RoomVoiceDiscoverQryExeTest { RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class); RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class); LiveMicClient liveMicClient = mock(LiveMicClient.class); - RedisTemplate redisTemplate = mock(RedisTemplate.class); + EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class); RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class); GameLudoService gameLudoService = mock(GameLudoService.class); UserProfileGateway userProfileGateway = mock(UserProfileGateway.class); @@ -67,7 +68,7 @@ class RoomVoiceDiscoverQryExeTest { roomProfileManagerService, regionRoomCacheService, liveMicClient, - redisTemplate, + emptyRoomCacheReader, rocketStatusCacheService, gameLudoService, userProfileGateway, @@ -94,10 +95,12 @@ class RoomVoiceDiscoverQryExeTest { when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion); when(userProfileGateway.getByUserId(1L)).thenReturn(userProfile); 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(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE")) .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( List.of(activeRoomCo) ); @@ -121,6 +124,8 @@ class RoomVoiceDiscoverQryExeTest { assertEquals(List.of(fallbackRoomCo, activeRoomCo), roomList); 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); } @@ -136,7 +141,7 @@ class RoomVoiceDiscoverQryExeTest { RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class); RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class); LiveMicClient liveMicClient = mock(LiveMicClient.class); - RedisTemplate redisTemplate = mock(RedisTemplate.class); + EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class); RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class); GameLudoService gameLudoService = mock(GameLudoService.class); UserProfileGateway userProfileGateway = mock(UserProfileGateway.class); @@ -152,7 +157,7 @@ class RoomVoiceDiscoverQryExeTest { roomProfileManagerService, regionRoomCacheService, liveMicClient, - redisTemplate, + emptyRoomCacheReader, rocketStatusCacheService, gameLudoService, userProfileGateway, @@ -179,9 +184,12 @@ class RoomVoiceDiscoverQryExeTest { when(roomRegionFilterCommon.requireRegionConfig(1L)).thenReturn(currentRegion); when(userProfileGateway.getByUserId(1L)).thenReturn(userProfile); 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(roomRegionFilterCommon.filterActiveVoiceRoomsByRegion(List.of(activeVoiceRoom), "AE")) .thenReturn(List.of(activeVoiceRoom)); + when(emptyRoomCacheReader.list("YOLO", "AE", false, 50)).thenReturn(List.of()); when(roomVoiceProfileCommon.toListRoomVoiceProfileCO(List.of(activeVoiceRoom))).thenReturn( List.of(topRoomCo) ); @@ -218,7 +226,7 @@ class RoomVoiceDiscoverQryExeTest { RoomProfileManagerService roomProfileManagerService = mock(RoomProfileManagerService.class); RegionRoomCacheService regionRoomCacheService = mock(RegionRoomCacheService.class); LiveMicClient liveMicClient = mock(LiveMicClient.class); - RedisTemplate redisTemplate = mock(RedisTemplate.class); + EmptyRoomCacheReader emptyRoomCacheReader = mock(EmptyRoomCacheReader.class); RocketStatusCacheService rocketStatusCacheService = mock(RocketStatusCacheService.class); GameLudoService gameLudoService = mock(GameLudoService.class); UserProfileGateway userProfileGateway = mock(UserProfileGateway.class); @@ -234,7 +242,7 @@ class RoomVoiceDiscoverQryExeTest { roomProfileManagerService, regionRoomCacheService, liveMicClient, - redisTemplate, + emptyRoomCacheReader, rocketStatusCacheService, gameLudoService, userProfileGateway, @@ -260,9 +268,11 @@ class RoomVoiceDiscoverQryExeTest { when(roomRegionFilterCommon.requireRegionConfig(ALL_REGION_WHITE_USER_ID)).thenReturn(currentRegion); when(userProfileGateway.getByUserId(ALL_REGION_WHITE_USER_ID)).thenReturn(userProfile); 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)) .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( List.of(activeRoomCo) ); @@ -282,7 +292,151 @@ class RoomVoiceDiscoverQryExeTest { assertEquals(List.of(activeRoomCo, fallbackRoomCo), roomList); 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()).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()); + } + } } diff --git a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/other/EmptyRoomCacheReader.java b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/other/EmptyRoomCacheReader.java new file mode 100644 index 00000000..fd9be954 --- /dev/null +++ b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/other/EmptyRoomCacheReader.java @@ -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 list(String sysOrigin, String region, boolean allRegions, int limit); +} diff --git a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/other/impl/EmptyRoomCacheReaderImpl.java b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/other/impl/EmptyRoomCacheReaderImpl.java new file mode 100644 index 00000000..ca66fb54 --- /dev/null +++ b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/other/impl/EmptyRoomCacheReaderImpl.java @@ -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 到期索引批量加载空房间快照。 + * + *

新 reader 不允许回退 legacy 全库 SCAN。滚动发布必须先全量升级 live writer，等待旧快照 + * 十分钟窗口收敛后再升级 other，确保任何在途 legacy-only payload 都已过期。

+ */ +@Slf4j +@Service +@RequiredArgsConstructor +public class EmptyRoomCacheReaderImpl implements EmptyRoomCacheReader { + + private final RedisService redisService; + + @Override + public List 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 indexedMembers = redisService.zsetGetReverseRange(indexKey, 0L, limit - 1L); + if (CollectionUtils.isEmpty(indexedMembers)) { + return new ArrayList<>(); + } + + List roomMembers = indexedMembers.stream() + .filter(Objects::nonNull) + .map(Objects::toString) + .filter(StringUtils::isNotBlank) + .distinct() + .toList(); + if (CollectionUtils.isEmpty(roomMembers)) { + return new ArrayList<>(); + } + + Set staleMembers = new LinkedHashSet<>(); + List validMembers = new ArrayList<>(); + List 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 payloadKeys = roomIds.stream() + .map(EmptyRoomCacheKeys::payload) + .toList(); + List payloads = Objects.requireNonNullElse( + redisService.multiGetString(payloadKeys), + List.of() + ); + List 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 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; + } + } +} diff --git a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/RegionRoomCacheService.java b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/RegionRoomCacheService.java index ed191298..efed4928 100644 --- a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/RegionRoomCacheService.java +++ b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/RegionRoomCacheService.java @@ -1,47 +1,72 @@ -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 java.util.Collection; -import java.util.List; -import java.util.Map; - -/** - * 区域房间缓存服务. - * - * @author lixiaofei on 2024/9/4 - */ -public interface RegionRoomCacheService { - - /** - * 移除缓存. - * - * @param key 用户id - */ - void remove(String key); - - /** - * 添加用户区域缓存. - * - * @param activeVoiceRooms 房间信息 - * @param key key - */ - void save(List activeVoiceRooms, String key); - - /** - * 添加用户区域缓存. - * - * @param jsonString 房间信息 - * @param key key - */ - void saveRegionsRoom(String jsonString, String key); - - /** - * 获取区域房间信息. - * - * @param key 用户id - */ - String getRegionsRoom(String key); - -} +package com.red.circle.other.infra.database.cache.service.user; + +import com.red.circle.other.infra.database.mongo.entity.live.ActiveVoiceRoom; +import java.util.List; +import java.util.concurrent.TimeUnit; + +/** + * 区域房间缓存服务. + * + * @author lixiaofei on 2024/9/4 + */ +public interface RegionRoomCacheService { + + /** + * 移除缓存. + * + * @param key 用户id + */ + void remove(String key); + + /** + * 添加用户区域缓存. + * + * @param activeVoiceRooms 房间信息 + * @param key key + */ + void save(List activeVoiceRooms, String key); + + /** + * 添加用户区域缓存. + * + * @param jsonString 房间信息 + * @param key 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); + + /** + * 获取区域房间信息. + * + * @param key 用户id + */ + 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); + +} diff --git a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImpl.java b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImpl.java index c7f85a77..248aba12 100644 --- a/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImpl.java +++ b/rc-service/rc-service-other/other-infrastructure/src/main/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImpl.java @@ -1,59 +1,98 @@ -package com.red.circle.other.infra.database.cache.service.user.impl; - -import com.google.gson.Gson; -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.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.tool.core.collection.CollectionUtils; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.stereotype.Service; - -import java.util.*; -import java.util.concurrent.TimeUnit; - -/** - * 区域房间缓存服务. - * - * @author lixiaofei on 2024/9/4 - */ -@Slf4j -@Service -@RequiredArgsConstructor -public class RegionRoomCacheServiceImpl implements RegionRoomCacheService { - - private final RedisService redisService; - private static final long CACHE_DAYS = 3; - private static final long CACHE_TIME = 10; - - @Override - public void remove(String key) { - redisService.delete(getRegionsRoomKey(key)); - } - - @Override - public void save(List activeVoiceRooms, String key) { - Gson gson = new Gson(); - String jsonString = gson.toJson(activeVoiceRooms); - log.warn("jsonString：{}", jsonString); - redisService.setIfAbsent(getRegionsRoomKey(key), jsonString, CACHE_TIME, TimeUnit.MINUTES); - } - - @Override - public void saveRegionsRoom(String jsonString, String key) { - redisService.setIfAbsent(getRegionsRoomKey(key), jsonString, CACHE_TIME, TimeUnit.MINUTES); - } - - @Override - public String getRegionsRoom(String key) { - return redisService.getString(getRegionsRoomKey(key)); - } - - private String getRegionsRoomKey(String userId) { - return UserKey.REGION_ROOM.getKey(userId); - } - -} +package com.red.circle.other.infra.database.cache.service.user.impl; + +import com.google.gson.Gson; +import com.red.circle.component.redis.service.RedisService; +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.mongo.entity.live.ActiveVoiceRoom; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.TimeUnit; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +/** + * 区域房间缓存服务. + * + * @author lixiaofei on 2024/9/4 + */ +@Slf4j +@Service +@RequiredArgsConstructor +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; + + @Override + public void remove(String key) { + redisService.delete(getRegionsRoomKey(key)); + } + + @Override + public void save(List activeVoiceRooms, String key) { + Gson gson = new Gson(); + String jsonString = gson.toJson(activeVoiceRooms); + log.warn("jsonString：{}", jsonString); + redisService.setIfAbsent( + getRegionsRoomKey(key), + jsonString, + CACHE_TIME_MINUTES, + TimeUnit.MINUTES + ); + } + + @Override + public void saveRegionsRoom(String jsonString, String key) { + 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 + public String getRegionsRoom(String 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) { + return UserKey.REGION_ROOM.getKey(userId); + } + + private String getRebuildLockKey(String key) { + return REBUILD_LOCK_PREFIX + key; + } + +} diff --git a/rc-service/rc-service-other/other-infrastructure/src/test/java/com/red/circle/other/infra/database/cache/service/other/impl/EmptyRoomCacheReaderImplTest.java b/rc-service/rc-service-other/other-infrastructure/src/test/java/com/red/circle/other/infra/database/cache/service/other/impl/EmptyRoomCacheReaderImplTest.java new file mode 100644 index 00000000..ba9297e8 --- /dev/null +++ b/rc-service/rc-service-other/other-infrastructure/src/test/java/com/red/circle/other/infra/database/cache/service/other/impl/EmptyRoomCacheReaderImplTest.java @@ -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 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 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 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 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 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); + } +} diff --git a/rc-service/rc-service-other/other-infrastructure/src/test/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImplTest.java b/rc-service/rc-service-other/other-infrastructure/src/test/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImplTest.java new file mode 100644 index 00000000..2d4489f8 --- /dev/null +++ b/rc-service/rc-service-other/other-infrastructure/src/test/java/com/red/circle/other/infra/database/cache/service/user/impl/RegionRoomCacheServiceImplTest.java @@ -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); + } +}