From 55675252da24902362c3ff122ad108f77a2074db Mon Sep 17 00:00:00 2001 From: LiggMax Date: Fri, 18 Sep 2026 17:17:42 +0800 Subject: [PATCH 01/19] feat: add anonymous user presence tracking --- .../com/ligg/common/constants/Constants.java | 12 ++ .../flow_bootstrap/config/RedisConfig.java | 19 ++ .../ligg/flowclient/config/WebMvcConfig.java | 5 +- .../controller/PresenceController.java | 67 +++++++ .../module/dto/PresenceHeartbeatDto.java | 36 ++++ .../flowclient/module/vo/OnlineCountVo.java | 15 ++ .../flowclient/module/vo/PresenceContext.java | 25 +++ .../flowclient/service/PresenceService.java | 16 ++ .../service/impl/PresenceServiceImpl.java | 176 ++++++++++++++++++ 9 files changed, 370 insertions(+), 1 deletion(-) create mode 100644 flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java create mode 100644 flow-client/src/main/java/com/ligg/flowclient/module/dto/PresenceHeartbeatDto.java create mode 100644 flow-client/src/main/java/com/ligg/flowclient/module/vo/OnlineCountVo.java create mode 100644 flow-client/src/main/java/com/ligg/flowclient/module/vo/PresenceContext.java create mode 100644 flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java create mode 100644 flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java diff --git a/common/src/main/java/com/ligg/common/constants/Constants.java b/common/src/main/java/com/ligg/common/constants/Constants.java index f925cb9..b428b03 100644 --- a/common/src/main/java/com/ligg/common/constants/Constants.java +++ b/common/src/main/java/com/ligg/common/constants/Constants.java @@ -124,6 +124,18 @@ public class Constants { */ public static final String AUTH_USER_SESSIONS_KEY = "animeflow:auth:user:sessions"; + /** 在线状态单设备记录，完整键为 animeflow:presence:{presenceId}。 */ + public static final String PRESENCE_KEY = "animeflow:presence"; + + /** 在线设备索引，成员为 presenceId，score 为最近心跳时间戳。 */ + public static final String PRESENCE_DEVICES_KEY = "animeflow:presence:devices"; + + /** 在线用户去重索引，成员为 user:{id} 或 visitor:{visitorId}。 */ + public static final String PRESENCE_USERS_KEY = "animeflow:presence:users"; + + /** 番剧在线设备索引，完整键为 animeflow:presence:subject:{subjectId}:devices。 */ + public static final String PRESENCE_SUBJECT_KEY = "animeflow:presence:subject"; + /** * user_oauth.platform 字段：Bangumi 第三方登录 */ diff --git a/flow-bootstrap/src/main/java/com/ligg/flow_bootstrap/config/RedisConfig.java b/flow-bootstrap/src/main/java/com/ligg/flow_bootstrap/config/RedisConfig.java index a9e0f63..5e61ef3 100644 --- a/flow-bootstrap/src/main/java/com/ligg/flow_bootstrap/config/RedisConfig.java +++ b/flow-bootstrap/src/main/java/com/ligg/flow_bootstrap/config/RedisConfig.java @@ -12,6 +12,7 @@ import org.springframework.core.io.ClassPathResource; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; @@ -48,6 +49,24 @@ public RedisTemplate redisTemplate(RedisConnectionFactory factor return template; } + /** + * Redis 字符串模板统一使用 UTF-8 字符串序列化。 + * + *

限流、Lua 脚本、队列和 Presence 索引都使用该模板；不要让这些 + * Redis 数据被对象 JSON 序列化器包装，否则会影响脚本参数和跨模块读取。

+ */ + @Bean + public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) { + StringRedisTemplate template = new StringRedisTemplate(factory); + StringRedisSerializer serializer = new StringRedisSerializer(); + template.setKeySerializer(serializer); + template.setValueSerializer(serializer); + template.setHashKeySerializer(serializer); + template.setHashValueSerializer(serializer); + template.afterPropertiesSet(); + return template; + } + @Bean public DefaultRedisScript ipRateLimitRedisScript() { DefaultRedisScript script = new DefaultRedisScript<>(); diff --git a/flow-client/src/main/java/com/ligg/flowclient/config/WebMvcConfig.java b/flow-client/src/main/java/com/ligg/flowclient/config/WebMvcConfig.java index bf90aad..bb1ceed 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/config/WebMvcConfig.java +++ b/flow-client/src/main/java/com/ligg/flowclient/config/WebMvcConfig.java @@ -56,7 +56,10 @@ public void addInterceptors(InterceptorRegistry registry) { .order(2); registry.addInterceptor(optionalAuthorizationInterceptor) - .addPathPatterns("/api/v1/bangumi/subjects/**", "/api/v1/bangumi/search/subjects") + .addPathPatterns( + "/api/v1/bangumi/subjects/**", + "/api/v1/bangumi/search/subjects", + "/api/v1/presence/**") .order(3); } diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java new file mode 100644 index 0000000..9931e43 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java @@ -0,0 +1,67 @@ +package com.ligg.flowclient.controller; + +import com.ligg.common.response.Result; +import com.ligg.common.statuenum.ResponseCode; +import com.ligg.common.exception.AuthorizationException; +import com.ligg.flowclient.annotation.IpEndpointRateLimit; +import com.ligg.flowclient.interceptor.AuthorizationInterceptor; +import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; +import com.ligg.flowclient.module.vo.OnlineCountVo; +import com.ligg.flowclient.service.PresenceService; +import jakarta.validation.Valid; +import lombok.RequiredArgsConstructor; +import org.springframework.http.HttpHeaders; +import org.springframework.http.ResponseEntity; +import org.springframework.util.StringUtils; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PutMapping; +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestAttribute; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequiredArgsConstructor +@RequestMapping("/api/v1/presence") +public class PresenceController { + + private final PresenceService presenceService; + + @PutMapping("/{presenceId}") + @IpEndpointRateLimit(keyPrefix = "animeflow:presence:heartbeat:ip:", seconds = 10, maxRequests = 2) + public ResponseEntity heartbeat( + @Valid @RequestBody PresenceHeartbeatDto body, + @PathVariable String presenceId, + @RequestAttribute( + value = AuthorizationInterceptor.ACCESS_TOKEN_REQUEST_ATTRIBUTE, + required = false) String accessToken, + @RequestHeader(value = HttpHeaders.AUTHORIZATION, required = false) + String authorization) { + boolean hasAuthorization = StringUtils.hasText(authorization); + if (hasAuthorization && !StringUtils.hasText(accessToken)) { + throw new AuthorizationException("缺少或无效的访问令牌"); + } + presenceService.heartbeat(presenceId, body, accessToken, hasAuthorization); + return ResponseEntity.noContent().build(); + } + + @DeleteMapping("/{presenceId}") + @IpEndpointRateLimit(keyPrefix = "animeflow:presence:offline:ip:", seconds = 10, maxRequests = 5) + public ResponseEntity offline(@PathVariable String presenceId) { + presenceService.offline(presenceId); + return ResponseEntity.noContent().build(); + } + + @GetMapping("/online-count") + public Result onlineCount() { + return Result.success(ResponseCode.SUCCESS, presenceService.onlineCount()); + } + + @GetMapping("/subjects/{subjectId}/online-count") + public Result subjectOnlineCount(@PathVariable int subjectId) { + return Result.success(ResponseCode.SUCCESS, presenceService.subjectOnlineCount(subjectId)); + } +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/module/dto/PresenceHeartbeatDto.java b/flow-client/src/main/java/com/ligg/flowclient/module/dto/PresenceHeartbeatDto.java new file mode 100644 index 0000000..49c38e3 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/module/dto/PresenceHeartbeatDto.java @@ -0,0 +1,36 @@ +package com.ligg.flowclient.module.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.Pattern; +import jakarta.validation.constraints.Positive; +import jakarta.validation.constraints.Size; +import lombok.Data; + +@Data +public class PresenceHeartbeatDto { + + @NotBlank + @Size(max = 64) + private String visitorId; + + @NotBlank + @Size(max = 32) + private String clientType; + + @Size(max = 32) + private String appVersion; + + @NotBlank + @Pattern(regexp = "online|watching|paused", message = "在线状态无效") + private String status = "online"; + + @Positive + private Integer subjectId; + + @Positive + private Integer episodeId; + + @Min(0) + private Integer positionSeconds; +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/module/vo/OnlineCountVo.java b/flow-client/src/main/java/com/ligg/flowclient/module/vo/OnlineCountVo.java new file mode 100644 index 0000000..60e444f --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/module/vo/OnlineCountVo.java @@ -0,0 +1,15 @@ +package com.ligg.flowclient.module.vo; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +@Data +@NoArgsConstructor +@AllArgsConstructor +public class OnlineCountVo { + private long onlineUsers; + private long onlineDevices; + private long anonymousUsers; + private long loggedInUsers; +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/module/vo/PresenceContext.java b/flow-client/src/main/java/com/ligg/flowclient/module/vo/PresenceContext.java new file mode 100644 index 0000000..50edf80 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/module/vo/PresenceContext.java @@ -0,0 +1,25 @@ +package com.ligg.flowclient.module.vo; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** + * 当前客户端 Presence 上下文，写入 Redis 后随单设备 TTL 过期。 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class PresenceContext { + private String presenceId; + private String visitorId; + private Long userId; + private String dedupeKey; + private String clientType; + private String appVersion; + private String status; + private Integer subjectId; + private Integer episodeId; + private Integer positionSeconds; + private long lastSeenAt; +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java new file mode 100644 index 0000000..89e4bc4 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java @@ -0,0 +1,16 @@ +package com.ligg.flowclient.service; + +import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; +import com.ligg.flowclient.module.vo.OnlineCountVo; + +public interface PresenceService { + + void heartbeat(String presenceId, PresenceHeartbeatDto dto, + String accessToken, boolean hasAuthorization); + + void offline(String presenceId); + + OnlineCountVo onlineCount(); + + OnlineCountVo subjectOnlineCount(int subjectId); +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java new file mode 100644 index 0000000..32d346b --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -0,0 +1,176 @@ +package com.ligg.flowclient.service.impl; + +import com.ligg.common.constants.Constants; +import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; +import com.ligg.flowclient.module.vo.OnlineCountVo; +import com.ligg.flowclient.module.vo.PresenceContext; +import com.ligg.flowclient.service.JwtTokenService; +import com.ligg.flowclient.service.PresenceService; +import lombok.RequiredArgsConstructor; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; + +import java.time.Instant; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +@Service +@RequiredArgsConstructor +public class PresenceServiceImpl implements PresenceService { + + private static final long TTL_SECONDS = 90L; + private static final long EXPIRED_BEFORE_SECONDS = TTL_SECONDS; + + private final RedisTemplate redisTemplate; + private final StringRedisTemplate stringRedisTemplate; + private final JwtTokenService jwtTokenService; + + @Override + public void heartbeat(String presenceId, PresenceHeartbeatDto dto, String accessToken, + boolean hasAuthorization) { + Long userId = hasAuthorization + ? jwtTokenService.validateAccessToken(accessToken) + : null; + long now = Instant.now().getEpochSecond(); + String dedupeKey = userId == null + ? "visitor:" + dto.getVisitorId() + : "user:" + userId; + String recordKey = recordKey(presenceId); + PresenceContext previous = readPreviousContext(recordKey); + PresenceContext context = new PresenceContext( + presenceId, + dto.getVisitorId(), + userId, + dedupeKey, + dto.getClientType(), + dto.getAppVersion(), + dto.getStatus(), + dto.getSubjectId(), + dto.getEpisodeId(), + dto.getPositionSeconds(), + now); + + // 单设备上下文使用 RedisConfig 中统一配置的对象序列化器，并随 TTL 过期。 + redisTemplate.opsForValue().set(recordKey, context, TTL_SECONDS, TimeUnit.SECONDS); + stringRedisTemplate.opsForZSet().add(Constants.PRESENCE_DEVICES_KEY, presenceId, now); + stringRedisTemplate.opsForZSet().add(Constants.PRESENCE_USERS_KEY, dedupeKey, now); + updateSubjectIndexes(previous, context, now); + + } + + @Override + public void offline(String presenceId) { + if (!StringUtils.hasText(presenceId)) { + return; + } + String key = recordKey(presenceId); + redisTemplate.delete(key); + stringRedisTemplate.opsForZSet().remove(Constants.PRESENCE_DEVICES_KEY, presenceId); + } + + @Override + public OnlineCountVo onlineCount() { + long now = Instant.now().getEpochSecond(); + long expiredBefore = now - EXPIRED_BEFORE_SECONDS; + stringRedisTemplate.opsForZSet().removeRangeByScore( + Constants.PRESENCE_DEVICES_KEY, Double.NEGATIVE_INFINITY, expiredBefore); + stringRedisTemplate.opsForZSet().removeRangeByScore( + Constants.PRESENCE_USERS_KEY, Double.NEGATIVE_INFINITY, expiredBefore); + + Set users = stringRedisTemplate.opsForZSet().rangeByScore( + Constants.PRESENCE_USERS_KEY, expiredBefore + 1, now); + long anonymous = 0; + long loggedIn = 0; + if (users != null) { + for (String user : users) { + if (user.startsWith("user:")) { + loggedIn++; + } else if (user.startsWith("visitor:")) { + anonymous++; + } + } + } + Long devices = stringRedisTemplate.opsForZSet().zCard(Constants.PRESENCE_DEVICES_KEY); + return new OnlineCountVo(anonymous + loggedIn, devices == null ? 0 : devices, anonymous, loggedIn); + } + + @Override + public OnlineCountVo subjectOnlineCount(int subjectId) { + if (subjectId <= 0) { + return new OnlineCountVo(0, 0, 0, 0); + } + long now = Instant.now().getEpochSecond(); + long expiredBefore = now - EXPIRED_BEFORE_SECONDS; + String devicesKey = subjectDevicesKey(subjectId); + String usersKey = subjectUsersKey(subjectId); + stringRedisTemplate.opsForZSet().removeRangeByScore( + devicesKey, Double.NEGATIVE_INFINITY, expiredBefore); + stringRedisTemplate.opsForZSet().removeRangeByScore( + usersKey, Double.NEGATIVE_INFINITY, expiredBefore); + + Set users = stringRedisTemplate.opsForZSet() + .rangeByScore(usersKey, expiredBefore + 1, now); + long anonymous = 0; + long loggedIn = 0; + if (users != null) { + for (String user : users) { + if (user.startsWith("user:")) { + loggedIn++; + } else if (user.startsWith("visitor:")) { + anonymous++; + } + } + } + Long devices = stringRedisTemplate.opsForZSet().zCard(devicesKey); + return new OnlineCountVo(anonymous + loggedIn, devices == null ? 0 : devices, anonymous, loggedIn); + } + + private void updateSubjectIndexes(PresenceContext previous, PresenceContext current, long now) { + if (previous != null && isWatching(previous) + && (!isWatching(current) || !sameSubject(previous, current))) { + stringRedisTemplate.opsForZSet().remove( + subjectDevicesKey(previous.getSubjectId()), current.getPresenceId()); + } + if (isWatching(current)) { + String devicesKey = subjectDevicesKey(current.getSubjectId()); + String usersKey = subjectUsersKey(current.getSubjectId()); + stringRedisTemplate.opsForZSet().add(devicesKey, current.getPresenceId(), now); + stringRedisTemplate.opsForZSet().add(usersKey, current.getDedupeKey(), now); + } + } + + private PresenceContext readPreviousContext(String recordKey) { + try { + Object value = redisTemplate.opsForValue().get(recordKey); + return value instanceof PresenceContext context ? context : null; + } catch (RuntimeException ignored) { + // 兼容从旧字符串占位记录迁移到对象序列化期间的短暂脏数据。 + return null; + } + } + + private static boolean isWatching(PresenceContext context) { + return "watching".equals(context.getStatus()) + && context.getSubjectId() != null + && context.getSubjectId() > 0; + } + + private static boolean sameSubject(PresenceContext first, PresenceContext second) { + return first.getSubjectId() != null + && first.getSubjectId().equals(second.getSubjectId()); + } + + private static String subjectDevicesKey(Integer subjectId) { + return Constants.PRESENCE_SUBJECT_KEY + ":" + subjectId + ":devices"; + } + + private static String subjectUsersKey(Integer subjectId) { + return Constants.PRESENCE_SUBJECT_KEY + ":" + subjectId + ":users"; + } + + private static String recordKey(String presenceId) { + return Constants.PRESENCE_KEY + ":" + presenceId; + } +} From d9ade3a5e1b7423c9b84991cd2329128aa2ec3c9 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Fri, 18 Sep 2026 18:02:11 +0800 Subject: [PATCH 02/19] feat: add watching subjects presence endpoint --- .../com/ligg/common/constants/Constants.java | 3 + .../controller/PresenceController.java | 6 ++ .../module/vo/WatchingSubjectVo.java | 19 ++++ .../flowclient/service/PresenceService.java | 5 ++ .../service/impl/PresenceServiceImpl.java | 88 +++++++++++++++++++ 5 files changed, 121 insertions(+) create mode 100644 flow-client/src/main/java/com/ligg/flowclient/module/vo/WatchingSubjectVo.java diff --git a/common/src/main/java/com/ligg/common/constants/Constants.java b/common/src/main/java/com/ligg/common/constants/Constants.java index b428b03..c15432c 100644 --- a/common/src/main/java/com/ligg/common/constants/Constants.java +++ b/common/src/main/java/com/ligg/common/constants/Constants.java @@ -136,6 +136,9 @@ public class Constants { /** 番剧在线设备索引，完整键为 animeflow:presence:subject:{subjectId}:devices。 */ public static final String PRESENCE_SUBJECT_KEY = "animeflow:presence:subject"; + /** 正在播放番剧索引，成员为 subjectId，score 为最近观看时间戳。 */ + public static final String PRESENCE_WATCHING_SUBJECTS_KEY = "animeflow:presence:watching-subjects"; + /** * user_oauth.platform 字段：Bangumi 第三方登录 */ diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java index 9931e43..997858b 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java @@ -7,6 +7,7 @@ import com.ligg.flowclient.interceptor.AuthorizationInterceptor; import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; import com.ligg.flowclient.module.vo.OnlineCountVo; +import com.ligg.flowclient.module.vo.WatchingSubjectVo; import com.ligg.flowclient.service.PresenceService; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; @@ -64,4 +65,9 @@ public Result onlineCount() { public Result subjectOnlineCount(@PathVariable int subjectId) { return Result.success(ResponseCode.SUCCESS, presenceService.subjectOnlineCount(subjectId)); } + + @GetMapping("/watching-subjects") + public Result> watchingSubjects() { + return Result.success(ResponseCode.SUCCESS, presenceService.watchingSubjects()); + } } diff --git a/flow-client/src/main/java/com/ligg/flowclient/module/vo/WatchingSubjectVo.java b/flow-client/src/main/java/com/ligg/flowclient/module/vo/WatchingSubjectVo.java new file mode 100644 index 0000000..f846468 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/module/vo/WatchingSubjectVo.java @@ -0,0 +1,19 @@ +package com.ligg.flowclient.module.vo; + +import com.ligg.common.model.CoverImages; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; + +/** 当前正在被用户播放的番剧。 */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class WatchingSubjectVo { + + private Integer subjectId; + private String name; + private String nameCn; + private CoverImages images; + private OnlineCountVo online; +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java index 89e4bc4..e6af7fc 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java @@ -2,6 +2,9 @@ import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; import com.ligg.flowclient.module.vo.OnlineCountVo; +import com.ligg.flowclient.module.vo.WatchingSubjectVo; + +import java.util.List; public interface PresenceService { @@ -13,4 +16,6 @@ void heartbeat(String presenceId, PresenceHeartbeatDto dto, OnlineCountVo onlineCount(); OnlineCountVo subjectOnlineCount(int subjectId); + + List watchingSubjects(); } diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 32d346b..ffe07d5 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -1,9 +1,15 @@ package com.ligg.flowclient.service.impl; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; import com.ligg.common.constants.Constants; +import com.ligg.common.entity.BangumiSubjectEntity; +import com.ligg.common.model.CoverImages; import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; import com.ligg.flowclient.module.vo.OnlineCountVo; import com.ligg.flowclient.module.vo.PresenceContext; +import com.ligg.flowclient.module.vo.WatchingSubjectVo; +import com.ligg.flowclient.mapper.BangumiSubjectMapper; import com.ligg.flowclient.service.JwtTokenService; import com.ligg.flowclient.service.PresenceService; import lombok.RequiredArgsConstructor; @@ -14,6 +20,8 @@ import java.time.Instant; import java.util.Set; +import java.util.Comparator; +import java.util.List; import java.util.concurrent.TimeUnit; @Service @@ -26,6 +34,8 @@ public class PresenceServiceImpl implements PresenceService { private final RedisTemplate redisTemplate; private final StringRedisTemplate stringRedisTemplate; private final JwtTokenService jwtTokenService; + private final BangumiSubjectMapper bangumiSubjectMapper; + private final ObjectMapper objectMapper; @Override public void heartbeat(String presenceId, PresenceHeartbeatDto dto, String accessToken, @@ -66,8 +76,14 @@ public void offline(String presenceId) { return; } String key = recordKey(presenceId); + PresenceContext previous = readPreviousContext(key); redisTemplate.delete(key); stringRedisTemplate.opsForZSet().remove(Constants.PRESENCE_DEVICES_KEY, presenceId); + if (previous != null && isWatching(previous)) { + stringRedisTemplate.opsForZSet().remove( + subjectDevicesKey(previous.getSubjectId()), presenceId); + removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); + } } @Override @@ -138,7 +154,48 @@ private void updateSubjectIndexes(PresenceContext previous, PresenceContext curr String usersKey = subjectUsersKey(current.getSubjectId()); stringRedisTemplate.opsForZSet().add(devicesKey, current.getPresenceId(), now); stringRedisTemplate.opsForZSet().add(usersKey, current.getDedupeKey(), now); + stringRedisTemplate.opsForZSet().add( + Constants.PRESENCE_WATCHING_SUBJECTS_KEY, + String.valueOf(current.getSubjectId()), now); + } + if (previous != null && isWatching(previous) + && (!isWatching(current) || !sameSubject(previous, current))) { + removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); + } + } + + @Override + public List watchingSubjects() { + long now = Instant.now().getEpochSecond(); + long expiredBefore = now - EXPIRED_BEFORE_SECONDS; + String indexKey = Constants.PRESENCE_WATCHING_SUBJECTS_KEY; + stringRedisTemplate.opsForZSet().removeRangeByScore( + indexKey, Double.NEGATIVE_INFINITY, expiredBefore); + Set subjectIds = stringRedisTemplate.opsForZSet() + .reverseRangeByScore(indexKey, now, expiredBefore + 1); + if (subjectIds == null || subjectIds.isEmpty()) { + return List.of(); } + + List ids = subjectIds.stream() + .map(this::parseSubjectId) + .filter(id -> id != null && id > 0) + .toList(); + if (ids.isEmpty()) { + return List.of(); + } + List subjects = bangumiSubjectMapper.selectBatchIds(ids); + return subjects.stream() + .map(subject -> new WatchingSubjectVo( + subject.getId(), + subject.getName(), + subject.getNameCn(), + parseImages(subject.getImages()), + subjectOnlineCount(subject.getId()))) + .filter(item -> item.getOnline().getOnlineDevices() > 0) + .sorted(Comparator.comparingLong( + item -> -item.getOnline().getOnlineUsers())) + .toList(); } private PresenceContext readPreviousContext(String recordKey) { @@ -162,6 +219,37 @@ private static boolean sameSubject(PresenceContext first, PresenceContext second && first.getSubjectId().equals(second.getSubjectId()); } + private void removeSubjectFromGlobalIndexIfOffline(Integer subjectId) { + if (subjectId == null || subjectId <= 0) { + return; + } + Long devices = stringRedisTemplate.opsForZSet() + .zCard(subjectDevicesKey(subjectId)); + if (devices == null || devices == 0) { + stringRedisTemplate.opsForZSet().remove( + Constants.PRESENCE_WATCHING_SUBJECTS_KEY, String.valueOf(subjectId)); + } + } + + private Integer parseSubjectId(String value) { + try { + return Integer.valueOf(value); + } catch (NumberFormatException ignored) { + return null; + } + } + + private CoverImages parseImages(String value) { + if (!StringUtils.hasText(value)) { + return null; + } + try { + return objectMapper.readValue(value, CoverImages.class); + } catch (JsonProcessingException ignored) { + return null; + } + } + private static String subjectDevicesKey(Integer subjectId) { return Constants.PRESENCE_SUBJECT_KEY + ":" + subjectId + ":devices"; } From 58e5c72595498e2558e69647dc0247dc70673e07 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Fri, 18 Sep 2026 18:08:55 +0800 Subject: [PATCH 03/19] fix: extend presence record ttl --- .../com/ligg/flowclient/service/impl/PresenceServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index ffe07d5..981594a 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -28,7 +28,7 @@ @RequiredArgsConstructor public class PresenceServiceImpl implements PresenceService { - private static final long TTL_SECONDS = 90L; + private static final long TTL_SECONDS = 180L; private static final long EXPIRED_BEFORE_SECONDS = TTL_SECONDS; private final RedisTemplate redisTemplate; From cf30da582fd62e3a50370a571759390cdcd3c338 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Fri, 18 Sep 2026 19:59:56 +0800 Subject: [PATCH 04/19] fix: correct watching subjects score range and playback presence --- .../service/impl/PresenceServiceImpl.java | 24 ++-- .../service/impl/PresenceServiceImplTest.java | 106 ++++++++++++++++++ 2 files changed, 120 insertions(+), 10 deletions(-) create mode 100644 flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 981594a..7b68381 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -79,7 +79,7 @@ public void offline(String presenceId) { PresenceContext previous = readPreviousContext(key); redisTemplate.delete(key); stringRedisTemplate.opsForZSet().remove(Constants.PRESENCE_DEVICES_KEY, presenceId); - if (previous != null && isWatching(previous)) { + if (previous != null && hasPlaybackContext(previous)) { stringRedisTemplate.opsForZSet().remove( subjectDevicesKey(previous.getSubjectId()), presenceId); removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); @@ -144,12 +144,12 @@ public OnlineCountVo subjectOnlineCount(int subjectId) { } private void updateSubjectIndexes(PresenceContext previous, PresenceContext current, long now) { - if (previous != null && isWatching(previous) - && (!isWatching(current) || !sameSubject(previous, current))) { + if (previous != null && hasPlaybackContext(previous) + && (!hasPlaybackContext(current) || !sameSubject(previous, current))) { stringRedisTemplate.opsForZSet().remove( subjectDevicesKey(previous.getSubjectId()), current.getPresenceId()); } - if (isWatching(current)) { + if (hasPlaybackContext(current)) { String devicesKey = subjectDevicesKey(current.getSubjectId()); String usersKey = subjectUsersKey(current.getSubjectId()); stringRedisTemplate.opsForZSet().add(devicesKey, current.getPresenceId(), now); @@ -158,8 +158,8 @@ private void updateSubjectIndexes(PresenceContext previous, PresenceContext curr Constants.PRESENCE_WATCHING_SUBJECTS_KEY, String.valueOf(current.getSubjectId()), now); } - if (previous != null && isWatching(previous) - && (!isWatching(current) || !sameSubject(previous, current))) { + if (previous != null && hasPlaybackContext(previous) + && (!hasPlaybackContext(current) || !sameSubject(previous, current))) { removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); } } @@ -172,7 +172,8 @@ public List watchingSubjects() { stringRedisTemplate.opsForZSet().removeRangeByScore( indexKey, Double.NEGATIVE_INFINITY, expiredBefore); Set subjectIds = stringRedisTemplate.opsForZSet() - .reverseRangeByScore(indexKey, now, expiredBefore + 1); + // Spring Data 的倒序查询仍然使用 min、max 参数顺序。 + .reverseRangeByScore(indexKey, expiredBefore + 1, now); if (subjectIds == null || subjectIds.isEmpty()) { return List.of(); } @@ -208,10 +209,13 @@ private PresenceContext readPreviousContext(String recordKey) { } } - private static boolean isWatching(PresenceContext context) { - return "watching".equals(context.getStatus()) + /** 播放页仍打开时，短暂暂停也属于有人正在观看该番剧。 */ + private static boolean hasPlaybackContext(PresenceContext context) { + return ("watching".equals(context.getStatus()) || "paused".equals(context.getStatus())) && context.getSubjectId() != null - && context.getSubjectId() > 0; + && context.getSubjectId() > 0 + && context.getEpisodeId() != null + && context.getEpisodeId() > 0; } private static boolean sameSubject(PresenceContext first, PresenceContext second) { diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java new file mode 100644 index 0000000..d5c4719 --- /dev/null +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -0,0 +1,106 @@ +package com.ligg.flowclient.service.impl; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.ligg.common.constants.Constants; +import com.ligg.common.entity.BangumiSubjectEntity; +import com.ligg.flowclient.mapper.BangumiSubjectMapper; +import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; +import com.ligg.flowclient.module.vo.PresenceContext; +import com.ligg.flowclient.service.JwtTokenService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.ValueOperations; +import org.springframework.data.redis.core.ZSetOperations; + +import java.util.List; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyDouble; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class PresenceServiceImplTest { + + private static final int SUBJECT_ID = 571784; + private static final String SUBJECT_DEVICES = Constants.PRESENCE_SUBJECT_KEY + ":" + SUBJECT_ID + ":devices"; + private static final String SUBJECT_USERS = Constants.PRESENCE_SUBJECT_KEY + ":" + SUBJECT_ID + ":users"; + + @Mock private RedisTemplate redisTemplate; + @Mock private StringRedisTemplate stringRedisTemplate; + @Mock private ValueOperations values; + @Mock private ZSetOperations zset; + @Mock private JwtTokenService jwtTokenService; + @Mock private BangumiSubjectMapper bangumiSubjectMapper; + + private PresenceServiceImpl service; + + @BeforeEach + void setUp() { + service = new PresenceServiceImpl(redisTemplate, stringRedisTemplate, + jwtTokenService, bangumiSubjectMapper, new ObjectMapper()); + when(redisTemplate.opsForValue()).thenReturn(values); + when(stringRedisTemplate.opsForZSet()).thenReturn(zset); + } + + @Test + void pausedPlaybackRemainsInWatchingSubjects() { + PresenceHeartbeatDto dto = new PresenceHeartbeatDto(); + dto.setVisitorId("visitor-1"); + dto.setClientType("ANDROID"); + dto.setStatus("paused"); + dto.setSubjectId(SUBJECT_ID); + dto.setEpisodeId(1699260); + + service.heartbeat("presence-1", dto, null, false); + + verify(zset).add(eq(SUBJECT_DEVICES), eq("presence-1"), anyDouble()); + verify(zset).add(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), eq(String.valueOf(SUBJECT_ID)), anyDouble()); + + BangumiSubjectEntity subject = new BangumiSubjectEntity(); + subject.setId(SUBJECT_ID); + subject.setName("Anime"); + when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), + anyDouble(), anyDouble())).thenAnswer(invocation -> { + double min = invocation.getArgument(1); + double max = invocation.getArgument(2); + // Spring Data takes min, max even for reverse queries. + assertEquals(179.0, max - min, "查询范围应覆盖最近 180 秒且下限小于上限"); + return Set.of(String.valueOf(SUBJECT_ID)); + }); + when(bangumiSubjectMapper.selectBatchIds(List.of(SUBJECT_ID))).thenReturn(List.of(subject)); + when(zset.rangeByScore(eq(SUBJECT_USERS), anyDouble(), anyDouble())) + .thenReturn(Set.of("visitor:visitor-1")); + when(zset.zCard(SUBJECT_DEVICES)).thenReturn(1L); + + var result = service.watchingSubjects(); + + assertEquals(1, result.size()); + assertEquals(SUBJECT_ID, result.get(0).getSubjectId()); + assertEquals(1, result.get(0).getOnline().getOnlineUsers()); + } + + @Test + void leavingPlaybackRemovesPausedSubject() { + PresenceContext previous = new PresenceContext("presence-1", "visitor-1", null, + "visitor:visitor-1", "ANDROID", null, "paused", SUBJECT_ID, + 1699260, 120, 1L); + when(values.get(Constants.PRESENCE_KEY + ":presence-1")).thenReturn(previous); + + PresenceHeartbeatDto dto = new PresenceHeartbeatDto(); + dto.setVisitorId("visitor-1"); + dto.setClientType("ANDROID"); + dto.setStatus("online"); + service.heartbeat("presence-1", dto, null, false); + + verify(zset).remove(SUBJECT_DEVICES, "presence-1"); + verify(zset).remove(Constants.PRESENCE_WATCHING_SUBJECTS_KEY, String.valueOf(SUBJECT_ID)); + } +} From 1e48695af6808826bdacd6fcd8421a509bc5a59d Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sat, 19 Sep 2026 11:52:37 +0800 Subject: [PATCH 05/19] update --- flow-bootstrap/src/main/resources/application.yaml | 2 +- .../java/com/ligg/flowclient/controller/PresenceController.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/flow-bootstrap/src/main/resources/application.yaml b/flow-bootstrap/src/main/resources/application.yaml index 4291ef3..0379d2b 100644 --- a/flow-bootstrap/src/main/resources/application.yaml +++ b/flow-bootstrap/src/main/resources/application.yaml @@ -99,7 +99,7 @@ anime-flow: ## 接口限流 rate-limit: enabled: true - max-requests: 60 #单个时间窗口内允许的最大请求次数。 + max-requests: 100 #单个时间窗口内允许的最大请求次数。 window-seconds: 60 #时间窗口长度（秒）。 path-patterns: #拦截路径 - /api/v1/** diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java index 997858b..6103a4a 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java @@ -32,7 +32,7 @@ public class PresenceController { private final PresenceService presenceService; @PutMapping("/{presenceId}") - @IpEndpointRateLimit(keyPrefix = "animeflow:presence:heartbeat:ip:", seconds = 10, maxRequests = 2) +// @IpEndpointRateLimit(keyPrefix = "animeflow:presence:heartbeat:ip:", seconds = 10, maxRequests = 2) public ResponseEntity heartbeat( @Valid @RequestBody PresenceHeartbeatDto body, @PathVariable String presenceId, From 49471d97015520c67ce03f0428e4841bf0a7aa41 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 18:39:33 +0800 Subject: [PATCH 06/19] refactor: remove deprecated sync status endpoint --- .../com/ligg/flowclient/controller/CollectionController.java | 1 - 1 file changed, 1 deletion(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/CollectionController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/CollectionController.java index a13f801..d9cc12c 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/controller/CollectionController.java +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/CollectionController.java @@ -94,7 +94,6 @@ public Result syncCollections( /** * 查询 Bangumi 收藏同步任务状态。 */ - @Deprecated @GetMapping("/sync") public Result getCollectionSyncStatus( @RequestAttribute(AuthorizationInterceptor.ACCESS_TOKEN_REQUEST_ATTRIBUTE) String accessToken) { From 2f8dc5582d2431df44983e75f556ebddce2efc29 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 21:24:05 +0800 Subject: [PATCH 07/19] fix: replace deprecated presence subject query --- .../ligg/flowclient/controller/PresenceController.java | 4 +++- .../flowclient/service/impl/PresenceServiceImpl.java | 10 ++++------ .../service/impl/PresenceServiceImplTest.java | 2 +- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java index 6103a4a..891c8e5 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java @@ -24,6 +24,8 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import java.util.List; + @RestController @RequiredArgsConstructor @RequestMapping("/api/v1/presence") @@ -67,7 +69,7 @@ public Result subjectOnlineCount(@PathVariable int subjectId) { } @GetMapping("/watching-subjects") - public Result> watchingSubjects() { + public Result> watchingSubjects() { return Result.success(ResponseCode.SUCCESS, presenceService.watchingSubjects()); } } diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 7b68381..e89ed67 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -29,7 +29,6 @@ public class PresenceServiceImpl implements PresenceService { private static final long TTL_SECONDS = 180L; - private static final long EXPIRED_BEFORE_SECONDS = TTL_SECONDS; private final RedisTemplate redisTemplate; private final StringRedisTemplate stringRedisTemplate; @@ -62,7 +61,6 @@ public void heartbeat(String presenceId, PresenceHeartbeatDto dto, String access dto.getPositionSeconds(), now); - // 单设备上下文使用 RedisConfig 中统一配置的对象序列化器，并随 TTL 过期。 redisTemplate.opsForValue().set(recordKey, context, TTL_SECONDS, TimeUnit.SECONDS); stringRedisTemplate.opsForZSet().add(Constants.PRESENCE_DEVICES_KEY, presenceId, now); stringRedisTemplate.opsForZSet().add(Constants.PRESENCE_USERS_KEY, dedupeKey, now); @@ -89,7 +87,7 @@ public void offline(String presenceId) { @Override public OnlineCountVo onlineCount() { long now = Instant.now().getEpochSecond(); - long expiredBefore = now - EXPIRED_BEFORE_SECONDS; + long expiredBefore = now - TTL_SECONDS; stringRedisTemplate.opsForZSet().removeRangeByScore( Constants.PRESENCE_DEVICES_KEY, Double.NEGATIVE_INFINITY, expiredBefore); stringRedisTemplate.opsForZSet().removeRangeByScore( @@ -118,7 +116,7 @@ public OnlineCountVo subjectOnlineCount(int subjectId) { return new OnlineCountVo(0, 0, 0, 0); } long now = Instant.now().getEpochSecond(); - long expiredBefore = now - EXPIRED_BEFORE_SECONDS; + long expiredBefore = now - TTL_SECONDS; String devicesKey = subjectDevicesKey(subjectId); String usersKey = subjectUsersKey(subjectId); stringRedisTemplate.opsForZSet().removeRangeByScore( @@ -167,7 +165,7 @@ private void updateSubjectIndexes(PresenceContext previous, PresenceContext curr @Override public List watchingSubjects() { long now = Instant.now().getEpochSecond(); - long expiredBefore = now - EXPIRED_BEFORE_SECONDS; + long expiredBefore = now - TTL_SECONDS; String indexKey = Constants.PRESENCE_WATCHING_SUBJECTS_KEY; stringRedisTemplate.opsForZSet().removeRangeByScore( indexKey, Double.NEGATIVE_INFINITY, expiredBefore); @@ -185,7 +183,7 @@ public List watchingSubjects() { if (ids.isEmpty()) { return List.of(); } - List subjects = bangumiSubjectMapper.selectBatchIds(ids); + List subjects = bangumiSubjectMapper.selectByIds(ids); return subjects.stream() .map(subject -> new WatchingSubjectVo( subject.getId(), diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index d5c4719..368516c 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -75,7 +75,7 @@ void pausedPlaybackRemainsInWatchingSubjects() { assertEquals(179.0, max - min, "查询范围应覆盖最近 180 秒且下限小于上限"); return Set.of(String.valueOf(SUBJECT_ID)); }); - when(bangumiSubjectMapper.selectBatchIds(List.of(SUBJECT_ID))).thenReturn(List.of(subject)); + when(bangumiSubjectMapper.selectByIds(List.of(SUBJECT_ID))).thenReturn(List.of(subject)); when(zset.rangeByScore(eq(SUBJECT_USERS), anyDouble(), anyDouble())) .thenReturn(Set.of("visitor:visitor-1")); when(zset.zCard(SUBJECT_DEVICES)).thenReturn(1L); From ddb6c715840c363358e2101d107576695caebe10 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 21:40:11 +0800 Subject: [PATCH 08/19] feat: expose presence endpoints as SSE --- .../controller/PresenceController.java | 28 +++-- .../service/PresenceSseService.java | 108 ++++++++++++++++++ 2 files changed, 127 insertions(+), 9 deletions(-) create mode 100644 flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java index 891c8e5..8247e68 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java @@ -1,17 +1,18 @@ package com.ligg.flowclient.controller; +import com.ligg.common.exception.AuthorizationException; import com.ligg.common.response.Result; import com.ligg.common.statuenum.ResponseCode; -import com.ligg.common.exception.AuthorizationException; import com.ligg.flowclient.annotation.IpEndpointRateLimit; import com.ligg.flowclient.interceptor.AuthorizationInterceptor; import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; import com.ligg.flowclient.module.vo.OnlineCountVo; -import com.ligg.flowclient.module.vo.WatchingSubjectVo; import com.ligg.flowclient.service.PresenceService; +import com.ligg.flowclient.service.PresenceSseService; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.util.StringUtils; import org.springframework.web.bind.annotation.GetMapping; @@ -24,7 +25,7 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; -import java.util.List; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @RestController @RequiredArgsConstructor @@ -32,6 +33,7 @@ public class PresenceController { private final PresenceService presenceService; + private final PresenceSseService presenceSseService; @PutMapping("/{presenceId}") // @IpEndpointRateLimit(keyPrefix = "animeflow:presence:heartbeat:ip:", seconds = 10, maxRequests = 2) @@ -58,9 +60,9 @@ public ResponseEntity offline(@PathVariable String presenceId) { return ResponseEntity.noContent().build(); } - @GetMapping("/online-count") - public Result onlineCount() { - return Result.success(ResponseCode.SUCCESS, presenceService.onlineCount()); + @GetMapping(value = "/online-count", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity onlineCount() { + return sseResponse(presenceSseService.subscribeOnlineCount()); } @GetMapping("/subjects/{subjectId}/online-count") @@ -68,8 +70,16 @@ public Result subjectOnlineCount(@PathVariable int subjectId) { return Result.success(ResponseCode.SUCCESS, presenceService.subjectOnlineCount(subjectId)); } - @GetMapping("/watching-subjects") - public Result> watchingSubjects() { - return Result.success(ResponseCode.SUCCESS, presenceService.watchingSubjects()); + @GetMapping(value = "/watching-subjects", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity watchingSubjects() { + return sseResponse(presenceSseService.subscribeWatchingSubjects()); + } + + private static ResponseEntity sseResponse(SseEmitter emitter) { + return ResponseEntity.ok() + .contentType(MediaType.TEXT_EVENT_STREAM) + .header("Cache-Control", "no-cache, no-transform") + .header("X-Accel-Buffering", "no") + .body(emitter); } } diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java new file mode 100644 index 0000000..c8a9b52 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java @@ -0,0 +1,108 @@ +package com.ligg.flowclient.service; + +import com.ligg.flowclient.module.vo.OnlineCountVo; +import com.ligg.flowclient.module.vo.WatchingSubjectVo; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.util.List; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** 在线状态 SSE；客户端断线后会重新获取完整快照。 */ +@Service +@RequiredArgsConstructor +public class PresenceSseService { + + private static final long CONNECTION_TIMEOUT_MS = 300_000L; + private static final long PUBLISH_INTERVAL_SECONDS = 5L; + + private final PresenceService presenceService; + private final Set onlineCountEmitters = ConcurrentHashMap.newKeySet(); + private final Set watchingSubjectEmitters = ConcurrentHashMap.newKeySet(); + private final ScheduledExecutorService scheduler = + Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "presence-sse-publisher"); + thread.setDaemon(true); + return thread; + }); + + @PostConstruct + void start() { + scheduler.scheduleWithFixedDelay( + this::publish, + PUBLISH_INTERVAL_SECONDS, + PUBLISH_INTERVAL_SECONDS, + TimeUnit.SECONDS); + } + + public SseEmitter subscribeOnlineCount() { + SseEmitter emitter = register(onlineCountEmitters); + sendOnlineCount(emitter); + return emitter; + } + + public SseEmitter subscribeWatchingSubjects() { + SseEmitter emitter = register(watchingSubjectEmitters); + sendWatchingSubjects(emitter); + return emitter; + } + + private SseEmitter register(Set emitters) { + SseEmitter emitter = new SseEmitter(CONNECTION_TIMEOUT_MS); + emitters.add(emitter); + Runnable remove = () -> emitters.remove(emitter); + emitter.onCompletion(remove); + emitter.onTimeout(remove); + emitter.onError(error -> remove.run()); + return emitter; + } + + private void publish() { + if (!onlineCountEmitters.isEmpty()) { + onlineCountEmitters.forEach(this::sendOnlineCount); + } + if (!watchingSubjectEmitters.isEmpty()) { + watchingSubjectEmitters.forEach(this::sendWatchingSubjects); + } + } + + private void sendOnlineCount(SseEmitter emitter) { + try { + emitter.send(SseEmitter.event() + .name("online-count") + .data(presenceService.onlineCount())); + } catch (Exception error) { + close(onlineCountEmitters, emitter); + } + } + + private void sendWatchingSubjects(SseEmitter emitter) { + try { + List subjects = presenceService.watchingSubjects(); + emitter.send(SseEmitter.event() + .name("watching-subjects") + .data(subjects)); + } catch (Exception error) { + close(watchingSubjectEmitters, emitter); + } + } + + private void close(Set emitters, SseEmitter emitter) { + emitters.remove(emitter); + emitter.completeWithError(new IllegalStateException("presence SSE closed")); + } + + @PreDestroy + void stop() { + onlineCountEmitters.forEach(SseEmitter::complete); + watchingSubjectEmitters.forEach(SseEmitter::complete); + scheduler.shutdownNow(); + } +} From 18939b321d85db8615090851ed260b0a3b0086b7 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 21:52:40 +0800 Subject: [PATCH 09/19] fix: convert presence subject images to CDN --- .../service/impl/PresenceServiceImpl.java | 17 +++++++++++------ .../service/impl/PresenceServiceImplTest.java | 3 +++ 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index e89ed67..61db212 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -5,6 +5,7 @@ import com.ligg.common.constants.Constants; import com.ligg.common.entity.BangumiSubjectEntity; import com.ligg.common.model.CoverImages; +import com.ligg.common.utils.Utils; import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; import com.ligg.flowclient.module.vo.OnlineCountVo; import com.ligg.flowclient.module.vo.PresenceContext; @@ -185,12 +186,16 @@ public List watchingSubjects() { } List subjects = bangumiSubjectMapper.selectByIds(ids); return subjects.stream() - .map(subject -> new WatchingSubjectVo( - subject.getId(), - subject.getName(), - subject.getNameCn(), - parseImages(subject.getImages()), - subjectOnlineCount(subject.getId()))) + .map(subject -> { + CoverImages images = parseImages(subject.getImages()); + Utils.applyWsrvCdnInPlace(images); + return new WatchingSubjectVo( + subject.getId(), + subject.getName(), + subject.getNameCn(), + images, + subjectOnlineCount(subject.getId())); + }) .filter(item -> item.getOnline().getOnlineDevices() > 0) .sorted(Comparator.comparingLong( item -> -item.getOnline().getOnlineUsers())) diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index 368516c..57a383f 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -67,6 +67,7 @@ void pausedPlaybackRemainsInWatchingSubjects() { BangumiSubjectEntity subject = new BangumiSubjectEntity(); subject.setId(SUBJECT_ID); subject.setName("Anime"); + subject.setImages("{\"large\":\"https://example.com/cover.jpg\"}"); when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), anyDouble(), anyDouble())).thenAnswer(invocation -> { double min = invocation.getArgument(1); @@ -85,6 +86,8 @@ void pausedPlaybackRemainsInWatchingSubjects() { assertEquals(1, result.size()); assertEquals(SUBJECT_ID, result.get(0).getSubjectId()); assertEquals(1, result.get(0).getOnline().getOnlineUsers()); + assertEquals("https://wsrv.nl/?url=https://example.com/cover.jpg", + result.get(0).getImages().getLarge()); } @Test From 04c1dfc77f87cb597d2027a5c1bbac27876383e7 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 23:15:09 +0800 Subject: [PATCH 10/19] feat: optimize presence SSE updates --- .../com/ligg/common/constants/Constants.java | 3 + .../config/PresenceSseConfiguration.java | 25 ++++++++ .../service/PresenceSseService.java | 61 +++++++++++++++--- .../service/impl/PresenceServiceImpl.java | 47 +++++++++++++- .../service/impl/PresenceServiceImplTest.java | 63 +++++++++++++++++++ 5 files changed, 189 insertions(+), 10 deletions(-) create mode 100644 flow-client/src/main/java/com/ligg/flowclient/config/PresenceSseConfiguration.java diff --git a/common/src/main/java/com/ligg/common/constants/Constants.java b/common/src/main/java/com/ligg/common/constants/Constants.java index c15432c..aeca165 100644 --- a/common/src/main/java/com/ligg/common/constants/Constants.java +++ b/common/src/main/java/com/ligg/common/constants/Constants.java @@ -139,6 +139,9 @@ public class Constants { /** 正在播放番剧索引，成员为 subjectId，score 为最近观看时间戳。 */ public static final String PRESENCE_WATCHING_SUBJECTS_KEY = "animeflow:presence:watching-subjects"; + /** Presence 状态变化通知频道。 */ + public static final String PRESENCE_CHANGED_CHANNEL = "animeflow:presence:changed"; + /** * user_oauth.platform 字段：Bangumi 第三方登录 */ diff --git a/flow-client/src/main/java/com/ligg/flowclient/config/PresenceSseConfiguration.java b/flow-client/src/main/java/com/ligg/flowclient/config/PresenceSseConfiguration.java new file mode 100644 index 0000000..b012443 --- /dev/null +++ b/flow-client/src/main/java/com/ligg/flowclient/config/PresenceSseConfiguration.java @@ -0,0 +1,25 @@ +package com.ligg.flowclient.config; + +import com.ligg.flowclient.service.PresenceSseService; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; + +/** Presence SSE 的 Redis Pub/Sub 监听配置。 */ +@Configuration +public class PresenceSseConfiguration { + + @Bean + public RedisMessageListenerContainer presenceSseListener( + RedisConnectionFactory factory, + PresenceSseService presenceSseService) { + var container = new RedisMessageListenerContainer(); + container.setConnectionFactory(factory); + container.addMessageListener( + presenceSseService, + new ChannelTopic(PresenceSseService.CHANNEL)); + return container; + } +} diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java index c8a9b52..639f210 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java @@ -1,10 +1,12 @@ package com.ligg.flowclient.service; -import com.ligg.flowclient.module.vo.OnlineCountVo; +import com.ligg.common.constants.Constants; import com.ligg.flowclient.module.vo.WatchingSubjectVo; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; +import org.springframework.data.redis.connection.Message; +import org.springframework.data.redis.connection.MessageListener; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -14,18 +16,24 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; -/** 在线状态 SSE；客户端断线后会重新获取完整快照。 */ +/** 在线状态 SSE；状态变化时推送，客户端断线后会重新获取完整快照。 */ @Service @RequiredArgsConstructor -public class PresenceSseService { +public class PresenceSseService implements MessageListener { + + public static final String CHANNEL = Constants.PRESENCE_CHANGED_CHANNEL; private static final long CONNECTION_TIMEOUT_MS = 300_000L; - private static final long PUBLISH_INTERVAL_SECONDS = 5L; + private static final long FLUSH_INTERVAL_MILLIS = 250L; + private static final long FALLBACK_REFRESH_SECONDS = 30L; + private static final long HEARTBEAT_SECONDS = 20L; private final PresenceService presenceService; private final Set onlineCountEmitters = ConcurrentHashMap.newKeySet(); private final Set watchingSubjectEmitters = ConcurrentHashMap.newKeySet(); + private final AtomicBoolean snapshotDirty = new AtomicBoolean(); private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { Thread thread = new Thread(runnable, "presence-sse-publisher"); @@ -36,9 +44,19 @@ public class PresenceSseService { @PostConstruct void start() { scheduler.scheduleWithFixedDelay( - this::publish, - PUBLISH_INTERVAL_SECONDS, - PUBLISH_INTERVAL_SECONDS, + this::flushDirtySnapshot, + FLUSH_INTERVAL_MILLIS, + FLUSH_INTERVAL_MILLIS, + TimeUnit.MILLISECONDS); + scheduler.scheduleWithFixedDelay( + this::markSnapshotDirty, + FALLBACK_REFRESH_SECONDS, + FALLBACK_REFRESH_SECONDS, + TimeUnit.SECONDS); + scheduler.scheduleWithFixedDelay( + this::sendHeartbeats, + HEARTBEAT_SECONDS, + HEARTBEAT_SECONDS, TimeUnit.SECONDS); } @@ -54,6 +72,11 @@ public SseEmitter subscribeWatchingSubjects() { return emitter; } + @Override + public void onMessage(Message message, byte[] pattern) { + snapshotDirty.set(true); + } + private SseEmitter register(Set emitters) { SseEmitter emitter = new SseEmitter(CONNECTION_TIMEOUT_MS); emitters.add(emitter); @@ -64,7 +87,16 @@ private SseEmitter register(Set emitters) { return emitter; } - private void publish() { + private void markSnapshotDirty() { + if (!onlineCountEmitters.isEmpty() || !watchingSubjectEmitters.isEmpty()) { + snapshotDirty.set(true); + } + } + + private void flushDirtySnapshot() { + if (!snapshotDirty.compareAndSet(true, false)) { + return; + } if (!onlineCountEmitters.isEmpty()) { onlineCountEmitters.forEach(this::sendOnlineCount); } @@ -73,6 +105,11 @@ private void publish() { } } + private void sendHeartbeats() { + onlineCountEmitters.forEach(emitter -> sendHeartbeat(onlineCountEmitters, emitter)); + watchingSubjectEmitters.forEach(emitter -> sendHeartbeat(watchingSubjectEmitters, emitter)); + } + private void sendOnlineCount(SseEmitter emitter) { try { emitter.send(SseEmitter.event() @@ -94,6 +131,14 @@ private void sendWatchingSubjects(SseEmitter emitter) { } } + private void sendHeartbeat(Set emitters, SseEmitter emitter) { + try { + emitter.send(SseEmitter.event().comment("heartbeat")); + } catch (Exception error) { + close(emitters, emitter); + } + } + private void close(Set emitters, SseEmitter emitter) { emitters.remove(emitter); emitter.completeWithError(new IllegalStateException("presence SSE closed")); diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 61db212..dca70b0 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -14,12 +14,14 @@ import com.ligg.flowclient.service.JwtTokenService; import com.ligg.flowclient.service.PresenceService; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import java.time.Instant; +import java.util.Objects; import java.util.Set; import java.util.Comparator; import java.util.List; @@ -27,9 +29,11 @@ @Service @RequiredArgsConstructor +@Slf4j public class PresenceServiceImpl implements PresenceService { private static final long TTL_SECONDS = 180L; + private static final int MAX_WATCHING_SUBJECTS = 100; private final RedisTemplate redisTemplate; private final StringRedisTemplate stringRedisTemplate; @@ -66,7 +70,9 @@ public void heartbeat(String presenceId, PresenceHeartbeatDto dto, String access stringRedisTemplate.opsForZSet().add(Constants.PRESENCE_DEVICES_KEY, presenceId, now); stringRedisTemplate.opsForZSet().add(Constants.PRESENCE_USERS_KEY, dedupeKey, now); updateSubjectIndexes(previous, context, now); - + if (presenceChanged(previous, context, now)) { + publishPresenceChanged(); + } } @Override @@ -83,6 +89,9 @@ public void offline(String presenceId) { subjectDevicesKey(previous.getSubjectId()), presenceId); removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); } + if (previous != null && isActive(previous, Instant.now().getEpochSecond())) { + publishPresenceChanged(); + } } @Override @@ -198,7 +207,10 @@ public List watchingSubjects() { }) .filter(item -> item.getOnline().getOnlineDevices() > 0) .sorted(Comparator.comparingLong( - item -> -item.getOnline().getOnlineUsers())) + (WatchingSubjectVo item) -> item.getOnline().getOnlineUsers()) + .reversed() + .thenComparingInt(WatchingSubjectVo::getSubjectId)) + .limit(MAX_WATCHING_SUBJECTS) .toList(); } @@ -226,6 +238,37 @@ private static boolean sameSubject(PresenceContext first, PresenceContext second && first.getSubjectId().equals(second.getSubjectId()); } + private static boolean presenceChanged(PresenceContext previous, + PresenceContext current, + long now) { + if (previous == null || !isActive(previous, now)) { + return true; + } + boolean onlineUsersChanged = !Objects.equals( + previous.getDedupeKey(), current.getDedupeKey()); + boolean watchingSubjectsChanged = !samePlaybackContext(previous, current); + return onlineUsersChanged || watchingSubjectsChanged; + } + + private static boolean samePlaybackContext(PresenceContext first, PresenceContext second) { + return hasPlaybackContext(first) + && hasPlaybackContext(second) + && sameSubject(first, second); + } + + private static boolean isActive(PresenceContext context, long now) { + return context.getLastSeenAt() > now - TTL_SECONDS; + } + + private void publishPresenceChanged() { + try { + stringRedisTemplate.convertAndSend(Constants.PRESENCE_CHANGED_CHANNEL, "changed"); + } catch (RuntimeException error) { + // SSE 有低频兜底刷新，通知失败不应影响在线状态写入。 + log.warn("Presence SSE change notification unavailable", error); + } + } + private void removeSubjectFromGlobalIndexIfOffline(Integer subjectId) { if (subjectId == null || subjectId <= 0) { return; diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index 57a383f..723a450 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -6,6 +6,7 @@ import com.ligg.flowclient.mapper.BangumiSubjectMapper; import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; import com.ligg.flowclient.module.vo.PresenceContext; +import com.ligg.flowclient.module.vo.WatchingSubjectVo; import com.ligg.flowclient.service.JwtTokenService; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -18,11 +19,17 @@ import org.springframework.data.redis.core.ZSetOperations; import java.util.List; +import java.util.HashSet; import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.anyDouble; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -63,6 +70,8 @@ void pausedPlaybackRemainsInWatchingSubjects() { verify(zset).add(eq(SUBJECT_DEVICES), eq("presence-1"), anyDouble()); verify(zset).add(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), eq(String.valueOf(SUBJECT_ID)), anyDouble()); + verify(stringRedisTemplate).convertAndSend( + Constants.PRESENCE_CHANGED_CHANNEL, "changed"); BangumiSubjectEntity subject = new BangumiSubjectEntity(); subject.setId(SUBJECT_ID); @@ -90,6 +99,60 @@ void pausedPlaybackRemainsInWatchingSubjects() { result.get(0).getImages().getLarge()); } + @Test + void unchangedHeartbeatDoesNotBroadcastAnotherSnapshot() { + long now = java.time.Instant.now().getEpochSecond(); + PresenceContext previous = new PresenceContext( + "presence-1", "visitor-1", null, "visitor:visitor-1", "ANDROID", null, + "watching", SUBJECT_ID, 1699260, 120, now - 1); + when(values.get(Constants.PRESENCE_KEY + ":presence-1")).thenReturn(previous); + + PresenceHeartbeatDto dto = new PresenceHeartbeatDto(); + dto.setVisitorId("visitor-1"); + dto.setClientType("ANDROID"); + dto.setStatus("watching"); + dto.setSubjectId(SUBJECT_ID); + dto.setEpisodeId(1699260); + dto.setPositionSeconds(121); + + service.heartbeat("presence-1", dto, null, false); + + verify(stringRedisTemplate, never()).convertAndSend( + Constants.PRESENCE_CHANGED_CHANNEL, "changed"); + } + + @Test + void watchingSubjectsReturnsTopOneHundredByOnlineUsers() { + List ids = IntStream.rangeClosed(1, 101).boxed().toList(); + List subjects = ids.stream().map(id -> { + BangumiSubjectEntity subject = new BangumiSubjectEntity(); + subject.setId(id); + subject.setName("Anime " + id); + return subject; + }).toList(); + + when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), + anyDouble(), anyDouble())) + .thenReturn(ids.stream().map(String::valueOf).collect(Collectors.toSet())); + when(bangumiSubjectMapper.selectByIds(anyList())).thenReturn(subjects); + when(zset.rangeByScore(anyString(), anyDouble(), anyDouble())).thenAnswer(invocation -> { + String key = invocation.getArgument(0); + int subjectId = Integer.parseInt(key.substring(key.lastIndexOf(':') + 1)); + Set users = new HashSet<>(); + for (int index = 0; index < 102 - subjectId; index++) { + users.add("visitor:" + subjectId + ":" + index); + } + return users; + }); + when(zset.zCard(anyString())).thenReturn(1L); + + List result = service.watchingSubjects(); + + assertEquals(100, result.size()); + assertEquals(1, result.get(0).getSubjectId()); + assertEquals(100, result.get(99).getSubjectId()); + } + @Test void leavingPlaybackRemovesPausedSubject() { PresenceContext previous = new PresenceContext("presence-1", "visitor-1", null, From d55963b74befdc57d78e7403f457694ef6a90880 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 23:24:08 +0800 Subject: [PATCH 11/19] perf: bound presence subject snapshot work --- .../service/PresenceSseService.java | 114 ++++++++++++++++-- .../service/impl/PresenceServiceImpl.java | 73 +++++++---- .../service/impl/PresenceServiceImplTest.java | 5 +- 3 files changed, 158 insertions(+), 34 deletions(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java index 639f210..a8b7507 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java @@ -1,6 +1,7 @@ package com.ligg.flowclient.service; import com.ligg.common.constants.Constants; +import com.ligg.flowclient.module.vo.OnlineCountVo; import com.ligg.flowclient.module.vo.WatchingSubjectVo; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; @@ -29,11 +30,20 @@ public class PresenceSseService implements MessageListener { private static final long FLUSH_INTERVAL_MILLIS = 250L; private static final long FALLBACK_REFRESH_SECONDS = 30L; private static final long HEARTBEAT_SECONDS = 20L; + private static final long SNAPSHOT_CACHE_MILLIS = 1_000L; private final PresenceService presenceService; private final Set onlineCountEmitters = ConcurrentHashMap.newKeySet(); private final Set watchingSubjectEmitters = ConcurrentHashMap.newKeySet(); private final AtomicBoolean snapshotDirty = new AtomicBoolean(); + private final AtomicBoolean onlineCountSnapshotDirty = new AtomicBoolean(true); + private final AtomicBoolean watchingSubjectsSnapshotDirty = new AtomicBoolean(true); + private final Object onlineCountSnapshotLock = new Object(); + private final Object watchingSubjectsSnapshotLock = new Object(); + private volatile OnlineCountVo onlineCountSnapshot; + private volatile long onlineCountSnapshotAt; + private volatile List watchingSubjectsSnapshot; + private volatile long watchingSubjectsSnapshotAt; private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(runnable -> { Thread thread = new Thread(runnable, "presence-sse-publisher"); @@ -74,7 +84,7 @@ public SseEmitter subscribeWatchingSubjects() { @Override public void onMessage(Message message, byte[] pattern) { - snapshotDirty.set(true); + invalidateSnapshots(); } private SseEmitter register(Set emitters) { @@ -89,7 +99,7 @@ private SseEmitter register(Set emitters) { private void markSnapshotDirty() { if (!onlineCountEmitters.isEmpty() || !watchingSubjectEmitters.isEmpty()) { - snapshotDirty.set(true); + invalidateSnapshots(); } } @@ -98,23 +108,53 @@ private void flushDirtySnapshot() { return; } if (!onlineCountEmitters.isEmpty()) { - onlineCountEmitters.forEach(this::sendOnlineCount); + sendOnlineCountSnapshot(); } if (!watchingSubjectEmitters.isEmpty()) { - watchingSubjectEmitters.forEach(this::sendWatchingSubjects); + sendWatchingSubjectsSnapshot(); } } + private void sendOnlineCountSnapshot() { + final OnlineCountVo snapshot; + try { + snapshot = getOnlineCountSnapshot(); + } catch (Exception error) { + snapshotDirty.set(true); + return; + } + onlineCountEmitters.forEach(emitter -> sendOnlineCount(emitter, snapshot)); + } + + private void sendWatchingSubjectsSnapshot() { + final List snapshot; + try { + snapshot = getWatchingSubjectsSnapshot(); + } catch (Exception error) { + snapshotDirty.set(true); + return; + } + watchingSubjectEmitters.forEach(emitter -> sendWatchingSubjects(emitter, snapshot)); + } + private void sendHeartbeats() { onlineCountEmitters.forEach(emitter -> sendHeartbeat(onlineCountEmitters, emitter)); watchingSubjectEmitters.forEach(emitter -> sendHeartbeat(watchingSubjectEmitters, emitter)); } private void sendOnlineCount(SseEmitter emitter) { + try { + sendOnlineCount(emitter, getOnlineCountSnapshot()); + } catch (Exception error) { + close(onlineCountEmitters, emitter); + } + } + + private void sendOnlineCount(SseEmitter emitter, OnlineCountVo snapshot) { try { emitter.send(SseEmitter.event() .name("online-count") - .data(presenceService.onlineCount())); + .data(snapshot)); } catch (Exception error) { close(onlineCountEmitters, emitter); } @@ -122,10 +162,17 @@ private void sendOnlineCount(SseEmitter emitter) { private void sendWatchingSubjects(SseEmitter emitter) { try { - List subjects = presenceService.watchingSubjects(); + sendWatchingSubjects(emitter, getWatchingSubjectsSnapshot()); + } catch (Exception error) { + close(watchingSubjectEmitters, emitter); + } + } + + private void sendWatchingSubjects(SseEmitter emitter, List snapshot) { + try { emitter.send(SseEmitter.event() .name("watching-subjects") - .data(subjects)); + .data(snapshot)); } catch (Exception error) { close(watchingSubjectEmitters, emitter); } @@ -144,6 +191,59 @@ private void close(Set emitters, SseEmitter emitter) { emitter.completeWithError(new IllegalStateException("presence SSE closed")); } + private void invalidateSnapshots() { + snapshotDirty.set(true); + onlineCountSnapshotDirty.set(true); + watchingSubjectsSnapshotDirty.set(true); + } + + private OnlineCountVo getOnlineCountSnapshot() { + long now = System.currentTimeMillis(); + if (isFresh(onlineCountSnapshot, onlineCountSnapshotAt, + onlineCountSnapshotDirty.get(), now)) { + return onlineCountSnapshot; + } + synchronized (onlineCountSnapshotLock) { + now = System.currentTimeMillis(); + if (isFresh(onlineCountSnapshot, onlineCountSnapshotAt, + onlineCountSnapshotDirty.get(), now)) { + return onlineCountSnapshot; + } + OnlineCountVo snapshot = presenceService.onlineCount(); + onlineCountSnapshot = snapshot; + onlineCountSnapshotAt = now; + onlineCountSnapshotDirty.set(false); + return snapshot; + } + } + + private List getWatchingSubjectsSnapshot() { + long now = System.currentTimeMillis(); + if (isFresh(watchingSubjectsSnapshot, watchingSubjectsSnapshotAt, + watchingSubjectsSnapshotDirty.get(), now)) { + return watchingSubjectsSnapshot; + } + synchronized (watchingSubjectsSnapshotLock) { + now = System.currentTimeMillis(); + if (isFresh(watchingSubjectsSnapshot, watchingSubjectsSnapshotAt, + watchingSubjectsSnapshotDirty.get(), now)) { + return watchingSubjectsSnapshot; + } + List snapshot = List.copyOf(presenceService.watchingSubjects()); + watchingSubjectsSnapshot = snapshot; + watchingSubjectsSnapshotAt = now; + watchingSubjectsSnapshotDirty.set(false); + return snapshot; + } + } + + private static boolean isFresh(Object snapshot, long snapshotAt, + boolean dirty, long now) { + return snapshot != null + && !dirty + && now - snapshotAt < SNAPSHOT_CACHE_MILLIS; + } + @PreDestroy void stop() { onlineCountEmitters.forEach(SseEmitter::complete); diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index dca70b0..766a0e4 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -22,6 +22,7 @@ import java.time.Instant; import java.util.Objects; +import java.util.PriorityQueue; import java.util.Set; import java.util.Comparator; import java.util.List; @@ -34,6 +35,7 @@ public class PresenceServiceImpl implements PresenceService { private static final long TTL_SECONDS = 180L; private static final int MAX_WATCHING_SUBJECTS = 100; + private static final int WATCHING_SUBJECTS_PAGE_SIZE = 200; private final RedisTemplate redisTemplate; private final StringRedisTemplate stringRedisTemplate; @@ -179,39 +181,60 @@ public List watchingSubjects() { String indexKey = Constants.PRESENCE_WATCHING_SUBJECTS_KEY; stringRedisTemplate.opsForZSet().removeRangeByScore( indexKey, Double.NEGATIVE_INFINITY, expiredBefore); - Set subjectIds = stringRedisTemplate.opsForZSet() - // Spring Data 的倒序查询仍然使用 min、max 参数顺序。 - .reverseRangeByScore(indexKey, expiredBefore + 1, now); - if (subjectIds == null || subjectIds.isEmpty()) { - return List.of(); - } + Comparator bestFirst = Comparator.comparingLong( + (WatchingSubjectVo item) -> item.getOnline().getOnlineUsers()) + .reversed() + .thenComparingInt(WatchingSubjectVo::getSubjectId); + Comparator worstFirst = (left, right) -> { + int onlineUsers = Long.compare( + left.getOnline().getOnlineUsers(), right.getOnline().getOnlineUsers()); + return onlineUsers != 0 + ? onlineUsers + : Integer.compare(right.getSubjectId(), left.getSubjectId()); + }; + PriorityQueue topSubjects = + new PriorityQueue<>(MAX_WATCHING_SUBJECTS + 1, worstFirst); - List ids = subjectIds.stream() - .map(this::parseSubjectId) - .filter(id -> id != null && id > 0) - .toList(); - if (ids.isEmpty()) { - return List.of(); - } - List subjects = bangumiSubjectMapper.selectByIds(ids); - return subjects.stream() - .map(subject -> { + long offset = 0; + while (true) { + Set subjectIds = stringRedisTemplate.opsForZSet() + // Spring Data 的倒序查询仍然使用 min、max 参数顺序。 + .reverseRangeByScore(indexKey, expiredBefore + 1, now, + offset, WATCHING_SUBJECTS_PAGE_SIZE); + if (subjectIds == null || subjectIds.isEmpty()) { + break; + } + + List ids = subjectIds.stream() + .map(this::parseSubjectId) + .filter(id -> id != null && id > 0) + .toList(); + if (!ids.isEmpty()) { + List subjects = bangumiSubjectMapper.selectByIds(ids); + for (BangumiSubjectEntity subject : subjects) { CoverImages images = parseImages(subject.getImages()); Utils.applyWsrvCdnInPlace(images); - return new WatchingSubjectVo( + WatchingSubjectVo item = new WatchingSubjectVo( subject.getId(), subject.getName(), subject.getNameCn(), images, subjectOnlineCount(subject.getId())); - }) - .filter(item -> item.getOnline().getOnlineDevices() > 0) - .sorted(Comparator.comparingLong( - (WatchingSubjectVo item) -> item.getOnline().getOnlineUsers()) - .reversed() - .thenComparingInt(WatchingSubjectVo::getSubjectId)) - .limit(MAX_WATCHING_SUBJECTS) - .toList(); + if (item.getOnline().getOnlineDevices() <= 0) { + continue; + } + topSubjects.offer(item); + if (topSubjects.size() > MAX_WATCHING_SUBJECTS) { + topSubjects.poll(); + } + } + } + if (subjectIds.size() < WATCHING_SUBJECTS_PAGE_SIZE) { + break; + } + offset += subjectIds.size(); + } + return topSubjects.stream().sorted(bestFirst).toList(); } private PresenceContext readPreviousContext(String recordKey) { diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index 723a450..4e37efa 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -27,6 +27,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.anyDouble; import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.never; @@ -78,7 +79,7 @@ void pausedPlaybackRemainsInWatchingSubjects() { subject.setName("Anime"); subject.setImages("{\"large\":\"https://example.com/cover.jpg\"}"); when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), - anyDouble(), anyDouble())).thenAnswer(invocation -> { + anyDouble(), anyDouble(), anyLong(), anyLong())).thenAnswer(invocation -> { double min = invocation.getArgument(1); double max = invocation.getArgument(2); // Spring Data takes min, max even for reverse queries. @@ -132,7 +133,7 @@ void watchingSubjectsReturnsTopOneHundredByOnlineUsers() { }).toList(); when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), - anyDouble(), anyDouble())) + anyDouble(), anyDouble(), anyLong(), anyLong())) .thenReturn(ids.stream().map(String::valueOf).collect(Collectors.toSet())); when(bangumiSubjectMapper.selectByIds(anyList())).thenReturn(subjects); when(zset.rangeByScore(anyString(), anyDouble(), anyDouble())).thenAnswer(invocation -> { From e987e996884ea6dab51ad40fc0218a87dcaded9c Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 23:46:28 +0800 Subject: [PATCH 12/19] perf: limit presence subject snapshot work --- .../service/impl/PresenceServiceImpl.java | 65 +++++++++++++------ .../service/impl/PresenceServiceImplTest.java | 13 +++- 2 files changed, 56 insertions(+), 22 deletions(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 766a0e4..6d1aac4 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -21,6 +21,8 @@ import org.springframework.util.StringUtils; import java.time.Instant; +import java.util.HashMap; +import java.util.Map; import java.util.Objects; import java.util.PriorityQueue; import java.util.Set; @@ -181,18 +183,18 @@ public List watchingSubjects() { String indexKey = Constants.PRESENCE_WATCHING_SUBJECTS_KEY; stringRedisTemplate.opsForZSet().removeRangeByScore( indexKey, Double.NEGATIVE_INFINITY, expiredBefore); - Comparator bestFirst = Comparator.comparingLong( - (WatchingSubjectVo item) -> item.getOnline().getOnlineUsers()) + Comparator bestFirst = Comparator.comparingLong( + (SubjectPresence item) -> item.online().getOnlineUsers()) .reversed() - .thenComparingInt(WatchingSubjectVo::getSubjectId); - Comparator worstFirst = (left, right) -> { + .thenComparingInt(SubjectPresence::subjectId); + Comparator worstFirst = (left, right) -> { int onlineUsers = Long.compare( - left.getOnline().getOnlineUsers(), right.getOnline().getOnlineUsers()); + left.online().getOnlineUsers(), right.online().getOnlineUsers()); return onlineUsers != 0 ? onlineUsers - : Integer.compare(right.getSubjectId(), left.getSubjectId()); + : Integer.compare(right.subjectId(), left.subjectId()); }; - PriorityQueue topSubjects = + PriorityQueue topSubjects = new PriorityQueue<>(MAX_WATCHING_SUBJECTS + 1, worstFirst); long offset = 0; @@ -210,20 +212,12 @@ public List watchingSubjects() { .filter(id -> id != null && id > 0) .toList(); if (!ids.isEmpty()) { - List subjects = bangumiSubjectMapper.selectByIds(ids); - for (BangumiSubjectEntity subject : subjects) { - CoverImages images = parseImages(subject.getImages()); - Utils.applyWsrvCdnInPlace(images); - WatchingSubjectVo item = new WatchingSubjectVo( - subject.getId(), - subject.getName(), - subject.getNameCn(), - images, - subjectOnlineCount(subject.getId())); - if (item.getOnline().getOnlineDevices() <= 0) { + for (Integer subjectId : ids) { + OnlineCountVo online = subjectOnlineCount(subjectId); + if (online.getOnlineDevices() <= 0) { continue; } - topSubjects.offer(item); + topSubjects.offer(new SubjectPresence(subjectId, online)); if (topSubjects.size() > MAX_WATCHING_SUBJECTS) { topSubjects.poll(); } @@ -234,7 +228,38 @@ public List watchingSubjects() { } offset += subjectIds.size(); } - return topSubjects.stream().sorted(bestFirst).toList(); + List rankedSubjects = topSubjects.stream().sorted(bestFirst).toList(); + if (rankedSubjects.isEmpty()) { + return List.of(); + } + + List topSubjectIds = rankedSubjects.stream() + .map(SubjectPresence::subjectId) + .toList(); + Map subjectsById = new HashMap<>(); + for (BangumiSubjectEntity subject : bangumiSubjectMapper.selectByIds(topSubjectIds)) { + subjectsById.put(subject.getId(), subject); + } + return rankedSubjects.stream() + .map(item -> { + BangumiSubjectEntity subject = subjectsById.get(item.subjectId()); + if (subject == null) { + return null; + } + CoverImages images = parseImages(subject.getImages()); + Utils.applyWsrvCdnInPlace(images); + return new WatchingSubjectVo( + subject.getId(), + subject.getName(), + subject.getNameCn(), + images, + item.online()); + }) + .filter(Objects::nonNull) + .toList(); + } + + private record SubjectPresence(int subjectId, OnlineCountVo online) { } private PresenceContext readPreviousContext(String recordKey) { diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index 4e37efa..ea22a5d 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -11,6 +11,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.data.redis.core.RedisTemplate; @@ -30,6 +31,7 @@ import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -54,7 +56,7 @@ class PresenceServiceImplTest { void setUp() { service = new PresenceServiceImpl(redisTemplate, stringRedisTemplate, jwtTokenService, bangumiSubjectMapper, new ObjectMapper()); - when(redisTemplate.opsForValue()).thenReturn(values); + lenient().when(redisTemplate.opsForValue()).thenReturn(values); when(stringRedisTemplate.opsForZSet()).thenReturn(zset); } @@ -138,7 +140,8 @@ void watchingSubjectsReturnsTopOneHundredByOnlineUsers() { when(bangumiSubjectMapper.selectByIds(anyList())).thenReturn(subjects); when(zset.rangeByScore(anyString(), anyDouble(), anyDouble())).thenAnswer(invocation -> { String key = invocation.getArgument(0); - int subjectId = Integer.parseInt(key.substring(key.lastIndexOf(':') + 1)); + String[] keyParts = key.split(":"); + int subjectId = Integer.parseInt(keyParts[keyParts.length - 2]); Set users = new HashSet<>(); for (int index = 0; index < 102 - subjectId; index++) { users.add("visitor:" + subjectId + ":" + index); @@ -152,6 +155,12 @@ void watchingSubjectsReturnsTopOneHundredByOnlineUsers() { assertEquals(100, result.size()); assertEquals(1, result.get(0).getSubjectId()); assertEquals(100, result.get(99).getSubjectId()); + + ArgumentCaptor> idsCaptor = ArgumentCaptor.forClass(List.class); + verify(bangumiSubjectMapper).selectByIds(idsCaptor.capture()); + assertEquals(100, idsCaptor.getValue().size()); + assertEquals(1, idsCaptor.getValue().get(0)); + assertEquals(100, idsCaptor.getValue().get(99)); } @Test From 17447f9099c0fe0eb7f6c6c58b3f267eabc1ff18 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Sun, 20 Sep 2026 23:49:39 +0800 Subject: [PATCH 13/19] fix: clean up stale presence user indexes --- .../service/impl/PresenceServiceImpl.java | 67 +++++++++++++++++-- .../service/impl/PresenceServiceImplTest.java | 42 ++++++++++++ 2 files changed, 103 insertions(+), 6 deletions(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 6d1aac4..098c805 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -86,14 +86,17 @@ public void offline(String presenceId) { } String key = recordKey(presenceId); PresenceContext previous = readPreviousContext(key); + long now = Instant.now().getEpochSecond(); redisTemplate.delete(key); stringRedisTemplate.opsForZSet().remove(Constants.PRESENCE_DEVICES_KEY, presenceId); + if (previous != null) { + removeGlobalUserIndexIfUnused(previous, now); + } if (previous != null && hasPlaybackContext(previous)) { - stringRedisTemplate.opsForZSet().remove( - subjectDevicesKey(previous.getSubjectId()), presenceId); + removeSubjectPresenceIndex(previous, now); removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); } - if (previous != null && isActive(previous, Instant.now().getEpochSecond())) { + if (previous != null && isActive(previous, now)) { publishPresenceChanged(); } } @@ -157,9 +160,10 @@ public OnlineCountVo subjectOnlineCount(int subjectId) { private void updateSubjectIndexes(PresenceContext previous, PresenceContext current, long now) { if (previous != null && hasPlaybackContext(previous) - && (!hasPlaybackContext(current) || !sameSubject(previous, current))) { - stringRedisTemplate.opsForZSet().remove( - subjectDevicesKey(previous.getSubjectId()), current.getPresenceId()); + && (!hasPlaybackContext(current) + || !sameSubject(previous, current) + || !Objects.equals(previous.getDedupeKey(), current.getDedupeKey()))) { + removeSubjectPresenceIndex(previous, now); } if (hasPlaybackContext(current)) { String devicesKey = subjectDevicesKey(current.getSubjectId()); @@ -174,6 +178,10 @@ private void updateSubjectIndexes(PresenceContext previous, PresenceContext curr && (!hasPlaybackContext(current) || !sameSubject(previous, current))) { removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); } + if (previous != null + && !Objects.equals(previous.getDedupeKey(), current.getDedupeKey())) { + removeGlobalUserIndexIfUnused(previous, now); + } } @Override @@ -329,6 +337,53 @@ private void removeSubjectFromGlobalIndexIfOffline(Integer subjectId) { } } + private void removeSubjectPresenceIndex(PresenceContext previous, long now) { + String devicesKey = subjectDevicesKey(previous.getSubjectId()); + stringRedisTemplate.opsForZSet().remove(devicesKey, previous.getPresenceId()); + removeDedupeIndexIfUnused( + devicesKey, + subjectUsersKey(previous.getSubjectId()), + previous.getDedupeKey(), + previous.getPresenceId(), + now); + } + + private void removeGlobalUserIndexIfUnused(PresenceContext previous, long now) { + removeDedupeIndexIfUnused( + Constants.PRESENCE_DEVICES_KEY, + Constants.PRESENCE_USERS_KEY, + previous.getDedupeKey(), + previous.getPresenceId(), + now); + } + + private void removeDedupeIndexIfUnused( + String devicesKey, + String usersKey, + String dedupeKey, + String excludedPresenceId, + long now) { + if (!StringUtils.hasText(dedupeKey)) { + return; + } + Set activePresenceIds = stringRedisTemplate.opsForZSet() + .rangeByScore(devicesKey, now - TTL_SECONDS + 1, now); + if (activePresenceIds != null) { + for (String presenceId : activePresenceIds) { + if (Objects.equals(presenceId, excludedPresenceId)) { + continue; + } + PresenceContext context = readPreviousContext(recordKey(presenceId)); + if (context != null + && isActive(context, now) + && Objects.equals(dedupeKey, context.getDedupeKey())) { + return; + } + } + } + stringRedisTemplate.opsForZSet().remove(usersKey, dedupeKey); + } + private Integer parseSubjectId(String value) { try { return Integer.valueOf(value); diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index ea22a5d..3c823b4 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -179,4 +179,46 @@ void leavingPlaybackRemovesPausedSubject() { verify(zset).remove(SUBJECT_DEVICES, "presence-1"); verify(zset).remove(Constants.PRESENCE_WATCHING_SUBJECTS_KEY, String.valueOf(SUBJECT_ID)); } + + @Test + void offlineRemovesSubjectAndGlobalUserIndexes() { + long now = java.time.Instant.now().getEpochSecond(); + PresenceContext previous = new PresenceContext( + "presence-1", "visitor-1", null, "visitor:visitor-1", "ANDROID", null, + "watching", SUBJECT_ID, 1699260, 120, now); + when(values.get(Constants.PRESENCE_KEY + ":presence-1")).thenReturn(previous); + + service.offline("presence-1"); + + verify(zset).remove(Constants.PRESENCE_DEVICES_KEY, "presence-1"); + verify(zset).remove(Constants.PRESENCE_USERS_KEY, "visitor:visitor-1"); + verify(zset).remove(SUBJECT_DEVICES, "presence-1"); + verify(zset).remove(SUBJECT_USERS, "visitor:visitor-1"); + } + + @Test + void switchingSubjectRemovesPreviousSubjectUserIndex() { + long now = java.time.Instant.now().getEpochSecond(); + int nextSubjectId = SUBJECT_ID + 1; + String nextSubjectDevices = Constants.PRESENCE_SUBJECT_KEY + ":" + nextSubjectId + ":devices"; + String nextSubjectUsers = Constants.PRESENCE_SUBJECT_KEY + ":" + nextSubjectId + ":users"; + PresenceContext previous = new PresenceContext( + "presence-1", "visitor-1", null, "visitor:visitor-1", "ANDROID", null, + "watching", SUBJECT_ID, 1699260, 120, now); + when(values.get(Constants.PRESENCE_KEY + ":presence-1")).thenReturn(previous); + + PresenceHeartbeatDto dto = new PresenceHeartbeatDto(); + dto.setVisitorId("visitor-1"); + dto.setClientType("ANDROID"); + dto.setStatus("watching"); + dto.setSubjectId(nextSubjectId); + dto.setEpisodeId(1699261); + + service.heartbeat("presence-1", dto, null, false); + + verify(zset).remove(SUBJECT_DEVICES, "presence-1"); + verify(zset).remove(SUBJECT_USERS, "visitor:visitor-1"); + verify(zset).add(eq(nextSubjectDevices), eq("presence-1"), anyDouble()); + verify(zset).add(eq(nextSubjectUsers), eq("visitor:visitor-1"), anyDouble()); + } } From 1364e46d4de32f47b740d959e20faa430a0538d6 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Mon, 21 Sep 2026 00:01:11 +0800 Subject: [PATCH 14/19] perf: rank active subjects with redis lua --- .../com/ligg/common/constants/Constants.java | 4 + .../service/impl/PresenceServiceImpl.java | 116 +++++++++++------- .../resources/lua/presence_top_subjects.lua | 36 ++++++ .../service/impl/PresenceServiceImplTest.java | 21 ++-- 4 files changed, 123 insertions(+), 54 deletions(-) create mode 100644 flow-client/src/main/resources/lua/presence_top_subjects.lua diff --git a/common/src/main/java/com/ligg/common/constants/Constants.java b/common/src/main/java/com/ligg/common/constants/Constants.java index aeca165..347dddc 100644 --- a/common/src/main/java/com/ligg/common/constants/Constants.java +++ b/common/src/main/java/com/ligg/common/constants/Constants.java @@ -139,6 +139,10 @@ public class Constants { /** 正在播放番剧索引，成员为 subjectId，score 为最近观看时间戳。 */ public static final String PRESENCE_WATCHING_SUBJECTS_KEY = "animeflow:presence:watching-subjects"; + /** 正在播放番剧排名索引，成员为 subjectId，score 为去重后的在线用户数。 */ + public static final String PRESENCE_WATCHING_SUBJECTS_RANK_KEY = + "animeflow:presence:watching-subjects:rank"; + /** Presence 状态变化通知频道。 */ public static final String PRESENCE_CHANGED_CHANNEL = "animeflow:presence:changed"; diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 098c805..5aa93c5 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -15,8 +15,10 @@ import com.ligg.flowclient.service.PresenceService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.core.io.ClassPathResource; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; @@ -24,7 +26,6 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; -import java.util.PriorityQueue; import java.util.Set; import java.util.Comparator; import java.util.List; @@ -37,7 +38,14 @@ public class PresenceServiceImpl implements PresenceService { private static final long TTL_SECONDS = 180L; private static final int MAX_WATCHING_SUBJECTS = 100; - private static final int WATCHING_SUBJECTS_PAGE_SIZE = 200; + private static final DefaultRedisScript TOP_WATCHING_SUBJECTS_SCRIPT = + new DefaultRedisScript<>(); + + static { + TOP_WATCHING_SUBJECTS_SCRIPT.setLocation( + new ClassPathResource("lua/presence_top_subjects.lua")); + TOP_WATCHING_SUBJECTS_SCRIPT.setResultType(List.class); + } private final RedisTemplate redisTemplate; private final StringRedisTemplate stringRedisTemplate; @@ -94,6 +102,7 @@ public void offline(String presenceId) { } if (previous != null && hasPlaybackContext(previous)) { removeSubjectPresenceIndex(previous, now); + refreshSubjectRanking(previous.getSubjectId(), now); removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); } if (previous != null && isActive(previous, now)) { @@ -178,6 +187,15 @@ private void updateSubjectIndexes(PresenceContext previous, PresenceContext curr && (!hasPlaybackContext(current) || !sameSubject(previous, current))) { removeSubjectFromGlobalIndexIfOffline(previous.getSubjectId()); } + if (previous != null && hasPlaybackContext(previous) + && (!hasPlaybackContext(current) + || !sameSubject(previous, current) + || !Objects.equals(previous.getDedupeKey(), current.getDedupeKey()))) { + refreshSubjectRanking(previous.getSubjectId(), now); + } + if (hasPlaybackContext(current)) { + refreshSubjectRanking(current.getSubjectId(), now); + } if (previous != null && !Objects.equals(previous.getDedupeKey(), current.getDedupeKey())) { removeGlobalUserIndexIfUnused(previous, now); @@ -195,48 +213,23 @@ public List watchingSubjects() { (SubjectPresence item) -> item.online().getOnlineUsers()) .reversed() .thenComparingInt(SubjectPresence::subjectId); - Comparator worstFirst = (left, right) -> { - int onlineUsers = Long.compare( - left.online().getOnlineUsers(), right.online().getOnlineUsers()); - return onlineUsers != 0 - ? onlineUsers - : Integer.compare(right.subjectId(), left.subjectId()); - }; - PriorityQueue topSubjects = - new PriorityQueue<>(MAX_WATCHING_SUBJECTS + 1, worstFirst); - - long offset = 0; - while (true) { - Set subjectIds = stringRedisTemplate.opsForZSet() - // Spring Data 的倒序查询仍然使用 min、max 参数顺序。 - .reverseRangeByScore(indexKey, expiredBefore + 1, now, - offset, WATCHING_SUBJECTS_PAGE_SIZE); - if (subjectIds == null || subjectIds.isEmpty()) { - break; - } + List candidateIds = topWatchingSubjectIds(expiredBefore); + if (candidateIds.isEmpty()) { + return List.of(); + } - List ids = subjectIds.stream() - .map(this::parseSubjectId) - .filter(id -> id != null && id > 0) - .toList(); - if (!ids.isEmpty()) { - for (Integer subjectId : ids) { - OnlineCountVo online = subjectOnlineCount(subjectId); - if (online.getOnlineDevices() <= 0) { - continue; - } - topSubjects.offer(new SubjectPresence(subjectId, online)); - if (topSubjects.size() > MAX_WATCHING_SUBJECTS) { - topSubjects.poll(); - } - } - } - if (subjectIds.size() < WATCHING_SUBJECTS_PAGE_SIZE) { - break; + List rankedSubjects = new java.util.ArrayList<>(candidateIds.size()); + for (Integer subjectId : candidateIds) { + OnlineCountVo online = subjectOnlineCount(subjectId); + if (online.getOnlineDevices() <= 0 || online.getOnlineUsers() <= 0) { + stringRedisTemplate.opsForZSet().remove( + Constants.PRESENCE_WATCHING_SUBJECTS_RANK_KEY, + String.valueOf(subjectId)); + continue; } - offset += subjectIds.size(); + rankedSubjects.add(new SubjectPresence(subjectId, online)); } - List rankedSubjects = topSubjects.stream().sorted(bestFirst).toList(); + rankedSubjects = rankedSubjects.stream().sorted(bestFirst).toList(); if (rankedSubjects.isEmpty()) { return List.of(); } @@ -270,6 +263,25 @@ public List watchingSubjects() { private record SubjectPresence(int subjectId, OnlineCountVo online) { } + private List topWatchingSubjectIds(long expiredBefore) { + List values = stringRedisTemplate.execute( + TOP_WATCHING_SUBJECTS_SCRIPT, + List.of( + Constants.PRESENCE_WATCHING_SUBJECTS_RANK_KEY, + Constants.PRESENCE_WATCHING_SUBJECTS_KEY), + String.valueOf(expiredBefore), + String.valueOf(MAX_WATCHING_SUBJECTS)); + if (values == null || values.isEmpty()) { + return List.of(); + } + return values.stream() + .map(Object::toString) + .map(this::parseSubjectId) + .filter(id -> id != null && id > 0) + .limit(MAX_WATCHING_SUBJECTS) + .toList(); + } + private PresenceContext readPreviousContext(String recordKey) { try { Object value = redisTemplate.opsForValue().get(recordKey); @@ -334,6 +346,8 @@ private void removeSubjectFromGlobalIndexIfOffline(Integer subjectId) { if (devices == null || devices == 0) { stringRedisTemplate.opsForZSet().remove( Constants.PRESENCE_WATCHING_SUBJECTS_KEY, String.valueOf(subjectId)); + stringRedisTemplate.opsForZSet().remove( + Constants.PRESENCE_WATCHING_SUBJECTS_RANK_KEY, String.valueOf(subjectId)); } } @@ -348,6 +362,26 @@ private void removeSubjectPresenceIndex(PresenceContext previous, long now) { now); } + private void refreshSubjectRanking(Integer subjectId, long now) { + if (subjectId == null || subjectId <= 0) { + return; + } + String usersKey = subjectUsersKey(subjectId); + stringRedisTemplate.opsForZSet().removeRangeByScore( + usersKey, Double.NEGATIVE_INFINITY, now - TTL_SECONDS); + Long onlineUsers = stringRedisTemplate.opsForZSet().zCard(usersKey); + if (onlineUsers == null || onlineUsers <= 0) { + stringRedisTemplate.opsForZSet().remove( + Constants.PRESENCE_WATCHING_SUBJECTS_RANK_KEY, + String.valueOf(subjectId)); + return; + } + stringRedisTemplate.opsForZSet().add( + Constants.PRESENCE_WATCHING_SUBJECTS_RANK_KEY, + String.valueOf(subjectId), + onlineUsers.doubleValue()); + } + private void removeGlobalUserIndexIfUnused(PresenceContext previous, long now) { removeDedupeIndexIfUnused( Constants.PRESENCE_DEVICES_KEY, diff --git a/flow-client/src/main/resources/lua/presence_top_subjects.lua b/flow-client/src/main/resources/lua/presence_top_subjects.lua new file mode 100644 index 0000000..0b52825 --- /dev/null +++ b/flow-client/src/main/resources/lua/presence_top_subjects.lua @@ -0,0 +1,36 @@ +local rank_key = KEYS[1] +local active_key = KEYS[2] +local expired_before = tonumber(ARGV[1]) +local limit = tonumber(ARGV[2]) +local result = {} +local cursor = 0 + +-- The rank index is maintained on presence changes. Remove entries whose +-- activity index has expired, and continue until the requested live entries +-- have been collected. +while #result < limit do + local candidates = redis.call('ZREVRANGE', rank_key, cursor, cursor + limit - 1) + if #candidates == 0 then + break + end + + local retained = 0 + for _, subject_id in ipairs(candidates) do + local last_seen = redis.call('ZSCORE', active_key, subject_id) + if not last_seen or tonumber(last_seen) <= expired_before then + redis.call('ZREM', rank_key, subject_id) + else + table.insert(result, subject_id) + retained = retained + 1 + if #result >= limit then + break + end + end + end + + -- Removed entries shift the sorted set left, so advance only by entries + -- retained in the current batch. + cursor = cursor + retained +end + +return result diff --git a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java index 3c823b4..62d9d9a 100644 --- a/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java +++ b/flow-client/src/test/java/com/ligg/flowclient/service/impl/PresenceServiceImplTest.java @@ -18,17 +18,17 @@ import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.data.redis.core.ValueOperations; import org.springframework.data.redis.core.ZSetOperations; +import org.springframework.data.redis.core.script.DefaultRedisScript; import java.util.List; import java.util.HashSet; import java.util.Set; -import java.util.stream.Collectors; import java.util.stream.IntStream; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.mockito.ArgumentMatchers.anyDouble; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; -import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.lenient; @@ -58,6 +58,9 @@ void setUp() { jwtTokenService, bangumiSubjectMapper, new ObjectMapper()); lenient().when(redisTemplate.opsForValue()).thenReturn(values); when(stringRedisTemplate.opsForZSet()).thenReturn(zset); + lenient().when(stringRedisTemplate.execute( + any(DefaultRedisScript.class), anyList(), any(), any())) + .thenReturn(List.of(String.valueOf(SUBJECT_ID))); } @Test @@ -80,14 +83,6 @@ void pausedPlaybackRemainsInWatchingSubjects() { subject.setId(SUBJECT_ID); subject.setName("Anime"); subject.setImages("{\"large\":\"https://example.com/cover.jpg\"}"); - when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), - anyDouble(), anyDouble(), anyLong(), anyLong())).thenAnswer(invocation -> { - double min = invocation.getArgument(1); - double max = invocation.getArgument(2); - // Spring Data takes min, max even for reverse queries. - assertEquals(179.0, max - min, "查询范围应覆盖最近 180 秒且下限小于上限"); - return Set.of(String.valueOf(SUBJECT_ID)); - }); when(bangumiSubjectMapper.selectByIds(List.of(SUBJECT_ID))).thenReturn(List.of(subject)); when(zset.rangeByScore(eq(SUBJECT_USERS), anyDouble(), anyDouble())) .thenReturn(Set.of("visitor:visitor-1")); @@ -134,9 +129,9 @@ void watchingSubjectsReturnsTopOneHundredByOnlineUsers() { return subject; }).toList(); - when(zset.reverseRangeByScore(eq(Constants.PRESENCE_WATCHING_SUBJECTS_KEY), - anyDouble(), anyDouble(), anyLong(), anyLong())) - .thenReturn(ids.stream().map(String::valueOf).collect(Collectors.toSet())); + when(stringRedisTemplate.execute( + any(DefaultRedisScript.class), anyList(), any(), any())) + .thenReturn(ids.stream().map(String::valueOf).toList()); when(bangumiSubjectMapper.selectByIds(anyList())).thenReturn(subjects); when(zset.rangeByScore(anyString(), anyDouble(), anyDouble())).thenAnswer(invocation -> { String key = invocation.getArgument(0); From a4bb4baba1ec843adadd72ab7f261ec7c03f3402 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Mon, 21 Sep 2026 12:07:40 +0800 Subject: [PATCH 15/19] chore: manage IntelliJ project settings --- .gitignore | 1 - .idea/.gitignore | 27 +++++++++++++++++ .idea/compiler.xml | 32 ++++++++++++++++++++ .idea/encodings.xml | 14 +++++++++ .idea/inspectionProfiles/Project_Default.xml | 12 ++++++++ .idea/jarRepositories.xml | 20 ++++++++++++ .idea/misc.xml | 11 +++++++ .idea/vcs.xml | 6 ++++ 8 files changed, 122 insertions(+), 1 deletion(-) create mode 100644 .idea/.gitignore create mode 100644 .idea/compiler.xml create mode 100644 .idea/encodings.xml create mode 100644 .idea/inspectionProfiles/Project_Default.xml create mode 100644 .idea/jarRepositories.xml create mode 100644 .idea/misc.xml create mode 100644 .idea/vcs.xml diff --git a/.gitignore b/.gitignore index 0781ec0..6e79c2d 100644 --- a/.gitignore +++ b/.gitignore @@ -14,7 +14,6 @@ target/ .env ### IntelliJ IDEA ### -.idea *.iws *.iml *.ipr diff --git a/.idea/.gitignore b/.idea/.gitignore new file mode 100644 index 0000000..a36db5d --- /dev/null +++ b/.idea/.gitignore @@ -0,0 +1,27 @@ +# IDEA project settings: only keep stable, project-level configuration. +/* +!/.gitignore +!/compiler.xml +!/encodings.xml +!/inspectionProfiles/ +!/inspectionProfiles/Project_Default.xml +!/jarRepositories.xml +!/misc.xml +!/vcs.xml + +# Machine-, user-, plugin- and database-specific settings. +/shelf/ +/workspace.xml +/httpRequests/ +/mavenHomeManager.xml +/dataSources/ +/dataSources.local.xml +/data_source_mapping.xml +/.cache/ +/dictionaries/ +/libraries/ +/ApifoxUploaderProjectSetting.xml +/db-forest-config.xml +/deviceManager.xml +/git_toolbox_prj.xml +/sqldialects.xml diff --git a/.idea/compiler.xml b/.idea/compiler.xml new file mode 100644 index 0000000..f5dcdd9 --- /dev/null +++ b/.idea/compiler.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/encodings.xml b/.idea/encodings.xml new file mode 100644 index 0000000..8545538 --- /dev/null +++ b/.idea/encodings.xml @@ -0,0 +1,14 @@ + + + + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/inspectionProfiles/Project_Default.xml b/.idea/inspectionProfiles/Project_Default.xml new file mode 100644 index 0000000..8d15915 --- /dev/null +++ b/.idea/inspectionProfiles/Project_Default.xml @@ -0,0 +1,12 @@ + + + + \ No newline at end of file diff --git a/.idea/jarRepositories.xml b/.idea/jarRepositories.xml new file mode 100644 index 0000000..712ab9d --- /dev/null +++ b/.idea/jarRepositories.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + \ No newline at end of file diff --git a/.idea/misc.xml b/.idea/misc.xml new file mode 100644 index 0000000..42e3abe --- /dev/null +++ b/.idea/misc.xml @@ -0,0 +1,11 @@ + + + + + + + \ No newline at end of file diff --git a/.idea/vcs.xml b/.idea/vcs.xml new file mode 100644 index 0000000..35eb1dd --- /dev/null +++ b/.idea/vcs.xml @@ -0,0 +1,6 @@ + + + + + + \ No newline at end of file From 98d8d560a0fdd9d9559d18b4ffebc0acba5d14d4 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Mon, 21 Sep 2026 12:46:04 +0800 Subject: [PATCH 16/19] perf: index sync task recovery polling --- db/anime_flow.sql | 3 ++- db/migrations/20260921_collection_sync_recovery_index.sql | 3 +++ 2 files changed, 5 insertions(+), 1 deletion(-) create mode 100644 db/migrations/20260921_collection_sync_recovery_index.sql diff --git a/db/anime_flow.sql b/db/anime_flow.sql index 6a2c3c2..c951dd4 100644 --- a/db/anime_flow.sql +++ b/db/anime_flow.sql @@ -416,7 +416,8 @@ CREATE TABLE `user_bgm_collection_sync_task` ( UNIQUE KEY `uk_collection_sync_active_user` (`active_user_id`), PRIMARY KEY (`id`), UNIQUE KEY `uk_collection_sync_request` (`user_id`,`request_id`), KEY `idx_collection_sync_active` (`user_id`,`status`,`id`), - KEY `idx_collection_sync_cleanup` (`status`,`finished_at`,`id`) + KEY `idx_collection_sync_cleanup` (`status`,`finished_at`,`id`), + KEY `idx_sync_task_recovery` (`status`,`heartbeat_at`,`created_at`,`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户 Bangumi 收藏双向同步任务'; CREATE TABLE `user_bgm_collection_sync_item` ( diff --git a/db/migrations/20260921_collection_sync_recovery_index.sql b/db/migrations/20260921_collection_sync_recovery_index.sql new file mode 100644 index 0000000..81be7c9 --- /dev/null +++ b/db/migrations/20260921_collection_sync_recovery_index.sql @@ -0,0 +1,3 @@ +-- Speed up recovery polling by filtering active statuses and ordering by recovery time. +ALTER TABLE user_bgm_collection_sync_task + ADD INDEX idx_sync_task_recovery (status, heartbeat_at, created_at, id); From bdc47ddc03bbbe9a8d8fb486338aecbc341566ec Mon Sep 17 00:00:00 2001 From: LiggMax Date: Mon, 21 Sep 2026 12:49:05 +0800 Subject: [PATCH 17/19] chore: remove obsolete database migrations --- ...60829_add_api_access_log_archive_index.sql | 3 -- .../20260919_collection_local_first.sql | 24 --------- .../20260919_collection_sync_tasks.sql | 53 ------------------- .../20260920_collection_sync_recovery.sql | 12 ----- ...0260921_collection_sync_recovery_index.sql | 3 -- ...20260921_collection_sync_scan_snapshot.sql | 3 -- 6 files changed, 98 deletions(-) delete mode 100644 db/migrations/20260829_add_api_access_log_archive_index.sql delete mode 100644 db/migrations/20260919_collection_local_first.sql delete mode 100644 db/migrations/20260919_collection_sync_tasks.sql delete mode 100644 db/migrations/20260920_collection_sync_recovery.sql delete mode 100644 db/migrations/20260921_collection_sync_recovery_index.sql delete mode 100644 db/migrations/20260921_collection_sync_scan_snapshot.sql diff --git a/db/migrations/20260829_add_api_access_log_archive_index.sql b/db/migrations/20260829_add_api_access_log_archive_index.sql deleted file mode 100644 index b1e5fa2..0000000 --- a/db/migrations/20260829_add_api_access_log_archive_index.sql +++ /dev/null @@ -1,3 +0,0 @@ --- 月度归档按 request_time、id 游标顺序分批读取。 -ALTER TABLE `api_access_log` - ADD INDEX `idx_request_time_id` (`request_time` ASC, `id` ASC); diff --git a/db/migrations/20260919_collection_local_first.sql b/db/migrations/20260919_collection_local_first.sql deleted file mode 100644 index b7f0ddf..0000000 --- a/db/migrations/20260919_collection_local_first.sql +++ /dev/null @@ -1,24 +0,0 @@ --- Phase 1. Run once, before deploying the new server/client; do not run anime_flow.sql on a live database. --- Keep legacy `private` values for audit; all new reads/writes use is_private. -ALTER TABLE user_bgm_collection - MODIFY bgm_interest_id BIGINT NULL COMMENT 'Bangumi 收藏 ID；仅本地收藏为 NULL', - MODIFY bgm_updated_at BIGINT NULL COMMENT 'Bangumi 更新时间 Unix 秒', - MODIFY sync_time DATETIME NULL DEFAULT NULL COMMENT '最后成功同步时间', - ADD local_updated_at DATETIME(3) NULL COMMENT '本地收藏内容最后修改时间（毫秒精度）', - ADD version BIGINT NOT NULL DEFAULT 0 COMMENT '本地收藏乐观锁版本，每次本地修改递增', - ADD remote_sync_status ENUM('LOCAL_ONLY','PENDING','SYNCED','AUTH_REQUIRED','CONFLICT') NOT NULL DEFAULT 'LOCAL_ONLY' COMMENT '远端同步状态', - ADD pending_payload JSON NULL COMMENT '合并后的待上传字段，不包含凭据', - ADD remote_baseline JSON NULL COMMENT '首次上传前的远端快照', - ADD sync_oauth_id BIGINT NULL COMMENT '目标绑定记录 ID，解绑后失效', - ADD bgm_account_uid BIGINT NULL COMMENT '目标 Bangumi 用户 UID，换绑后必须重新建立关联', - ADD retry_count INT NOT NULL DEFAULT 0 COMMENT '远端同步连续重试次数', - ADD next_retry_at DATETIME NULL COMMENT '下一次远端同步重试时间', - DROP INDEX uk_bgm_interest_id, - ADD INDEX idx_bgm_interest_id (bgm_interest_id), - ADD INDEX idx_user_collection_local (user_id, type, subject_type, local_updated_at DESC, id DESC), - ADD INDEX idx_collection_upload (remote_sync_status, next_retry_at); - -UPDATE user_bgm_collection -SET local_updated_at = COALESCE(sync_time, create_time, CURRENT_TIMESTAMP); - --- Historical rows remain version=0, without upload intent. A later edit/import establishes synchronization state. diff --git a/db/migrations/20260919_collection_sync_tasks.sql b/db/migrations/20260919_collection_sync_tasks.sql deleted file mode 100644 index 7014042..0000000 --- a/db/migrations/20260919_collection_sync_tasks.sql +++ /dev/null @@ -1,53 +0,0 @@ --- P2/P3: durable bidirectional sync tasks and conflict work items. -CREATE TABLE IF NOT EXISTS user_bgm_collection_sync_task ( - id BIGINT NOT NULL AUTO_INCREMENT COMMENT '同步任务主键', - user_id BIGINT NOT NULL COMMENT 'AnimeFlow 用户 ID', - subject_type TINYINT NOT NULL COMMENT '条目大类：1漫画、2动画、3音乐、4游戏、6三次元', - bgm_account_uid BIGINT NULL COMMENT '本次任务目标 Bangumi 用户 UID，换绑后不可复用', - oauth_id BIGINT NULL COMMENT '本次任务使用的 user_oauth 绑定记录 ID', - request_id VARCHAR(80) NOT NULL COMMENT '客户端幂等请求 ID', - status ENUM('QUEUED','RUNNING','WAITING_CONFLICT','PARTIAL_FAILED','SUCCESS','FAILED','CANCELLED') NOT NULL COMMENT '任务状态', - phase ENUM('SCANNING','APPLYING','RESOLVING') NOT NULL COMMENT '执行阶段', - total_count INT NOT NULL DEFAULT 0 COMMENT '本次任务去重后的条目总数', - imported_count INT NOT NULL DEFAULT 0 COMMENT '从 Bangumi 导入到本地的条目数', - uploaded_count INT NOT NULL DEFAULT 0 COMMENT '从本地上传到 Bangumi 的条目数', - unchanged_count INT NOT NULL DEFAULT 0 COMMENT '双方数据一致、无需更新的条目数', - conflict_count INT NOT NULL DEFAULT 0 COMMENT '等待用户处理的冲突数', - resolved_count INT NOT NULL DEFAULT 0 COMMENT '已完成决议并确认同步的冲突数', - failed_count INT NOT NULL DEFAULT 0 COMMENT '处理失败、可重试的条目数', - status_version BIGINT NOT NULL DEFAULT 0 COMMENT '任务状态版本，用于客户端判断状态是否变化', - error_code ENUM('SYNC_BINDING_CHANGED','SYNC_RETRY_REQUIRED') NULL COMMENT '任务级错误码', - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '任务入库时间', - started_at DATETIME(3) NULL COMMENT '任务首次开始执行时间', - finished_at DATETIME(3) NULL COMMENT '任务进入终态的时间', - heartbeat_at DATETIME(3) NULL COMMENT '后台执行器最近一次续租/心跳时间', - PRIMARY KEY (id), - UNIQUE KEY uk_collection_sync_request (user_id, request_id), - KEY idx_collection_sync_active (user_id, status, id), - KEY idx_collection_sync_cleanup (status, finished_at, id) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户 Bangumi 收藏双向同步任务'; - -CREATE TABLE IF NOT EXISTS user_bgm_collection_sync_item ( - id BIGINT NOT NULL AUTO_INCREMENT COMMENT '同步明细主键，同时作为客户端冲突 ID', - task_id BIGINT NOT NULL COMMENT '所属同步任务 ID', - user_id BIGINT NOT NULL COMMENT 'AnimeFlow 用户 ID，用于归属校验', - subject_id INT NOT NULL COMMENT 'Bangumi 条目 ID', - subject_type TINYINT NOT NULL COMMENT '条目大类，必须与任务目标类型一致', - subject_name VARCHAR(512) NULL COMMENT '冲突发生时的条目名称快照', - subject_image VARCHAR(1024) NULL COMMENT '冲突发生时的条目封面地址快照', - local_type TINYINT NULL COMMENT '冲突发生时本地收藏分类，1至5', - remote_type TINYINT NULL COMMENT '冲突发生时 Bangumi 收藏分类，1至5', - local_version BIGINT NULL COMMENT '冲突发生时本地收藏乐观锁版本', - conflict_version BIGINT NOT NULL DEFAULT 1 COMMENT '冲突快照版本，提交决议时必须匹配', - selected_type TINYINT NULL COMMENT '用户选择的最终收藏分类，决议入库前为空', - local_snapshot JSON NULL COMMENT '冲突发生时的本地收藏字段快照，不包含凭据', - remote_snapshot JSON NULL COMMENT '冲突发生时的 Bangumi 收藏字段快照，不包含凭据', - status ENUM('PLANNED','CONFLICT','RESOLUTION_PENDING','APPLY_PENDING','DONE','FAILED') NOT NULL COMMENT '明细状态', - error_code VARCHAR(64) NULL COMMENT '明细错误码，不保存 token 或完整私密内容', - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '明细创建时间', - resolved_at DATETIME(3) NULL COMMENT '远端确认并完成本地回写的时间', - PRIMARY KEY (id), - UNIQUE KEY uk_collection_sync_item (task_id, subject_id), - KEY idx_collection_sync_conflict (user_id, status, id), - CONSTRAINT fk_collection_sync_item_task FOREIGN KEY (task_id) REFERENCES user_bgm_collection_sync_task(id) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='用户 Bangumi 收藏同步明细与冲突快照'; diff --git a/db/migrations/20260920_collection_sync_recovery.sql b/db/migrations/20260920_collection_sync_recovery.sql deleted file mode 100644 index c0c4973..0000000 --- a/db/migrations/20260920_collection_sync_recovery.sql +++ /dev/null @@ -1,12 +0,0 @@ --- Run after 20260919_collection_sync_tasks.sql. No collection content is changed. --- Refuse migration if duplicate active tasks already exist; reconcile those tasks first. -ALTER TABLE user_bgm_collection_sync_task - ADD active_user_id BIGINT GENERATED ALWAYS AS - (CASE WHEN status IN ('QUEUED','RUNNING','WAITING_CONFLICT','PARTIAL_FAILED') - THEN user_id ELSE NULL END) STORED COMMENT '活动任务用户唯一约束，终态为 NULL', - ADD UNIQUE KEY uk_collection_sync_active_user (active_user_id); - -ALTER TABLE user_bgm_collection_sync_item - ADD desired_payload JSON NULL COMMENT '远端写入前持久化的目标字段，重启后校验基线再执行', - ADD operation ENUM('IMPORT','UPLOAD','UNCHANGED','RESOLVE') NULL COMMENT '完成的同步操作', - MODIFY status ENUM('PLANNED','CONFLICT','RESOLUTION_PENDING','APPLY_PENDING','DONE','FAILED') NOT NULL COMMENT '明细状态'; diff --git a/db/migrations/20260921_collection_sync_recovery_index.sql b/db/migrations/20260921_collection_sync_recovery_index.sql deleted file mode 100644 index 81be7c9..0000000 --- a/db/migrations/20260921_collection_sync_recovery_index.sql +++ /dev/null @@ -1,3 +0,0 @@ --- Speed up recovery polling by filtering active statuses and ordering by recovery time. -ALTER TABLE user_bgm_collection_sync_task - ADD INDEX idx_sync_task_recovery (status, heartbeat_at, created_at, id); diff --git a/db/migrations/20260921_collection_sync_scan_snapshot.sql b/db/migrations/20260921_collection_sync_scan_snapshot.sql deleted file mode 100644 index 36cd91e..0000000 --- a/db/migrations/20260921_collection_sync_scan_snapshot.sql +++ /dev/null @@ -1,3 +0,0 @@ --- Run after 20260920_collection_sync_recovery.sql, before deploying the snapshot reader. -ALTER TABLE user_bgm_collection_sync_item - ADD scan_snapshot JSON NULL COMMENT '收藏分页条目及interest快照，首次比较与导入复用；NULL兼容旧任务'; From 8d044c1c62a4404e1ea90d9a4f53225a235e2979 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Mon, 21 Sep 2026 13:57:56 +0800 Subject: [PATCH 18/19] feat: handle SSE connection timeout in GlobalExceptionHandler --- .../com/ligg/common/handler/GlobalExceptionHandler.java | 9 +++++++++ .../flowclient/service/CollectionSyncSseService.java | 4 +++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/common/src/main/java/com/ligg/common/handler/GlobalExceptionHandler.java b/common/src/main/java/com/ligg/common/handler/GlobalExceptionHandler.java index 6924fc1..45d2db4 100644 --- a/common/src/main/java/com/ligg/common/handler/GlobalExceptionHandler.java +++ b/common/src/main/java/com/ligg/common/handler/GlobalExceptionHandler.java @@ -28,6 +28,7 @@ import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.RestControllerAdvice; +import org.springframework.web.context.request.async.AsyncRequestTimeoutException; import org.springframework.web.method.annotation.MethodArgumentTypeMismatchException; /** @@ -224,6 +225,14 @@ public Result handleVerificationCode(VerificationCodeException e) { return Result.error(ResponseCode.PARAM_ERROR, e.getMessage()); } + /** + * SSE 连接超时或被容器关闭。 + */ + @ExceptionHandler(AsyncRequestTimeoutException.class) + public void handleAsyncRequestTimeout(AsyncRequestTimeoutException e) { + log.debug("SSE connection timed out or was closed"); + } + /** * 兜底异常处理 */ diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/CollectionSyncSseService.java b/flow-client/src/main/java/com/ligg/flowclient/service/CollectionSyncSseService.java index 68075b1..7240567 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/CollectionSyncSseService.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/CollectionSyncSseService.java @@ -43,7 +43,9 @@ public CollectionSyncSseService(CollectionSyncTaskService tasks, JwtTokenService public SseEmitter subscribe(Long userId, String accessToken) { // 定期续期也能限制消息代理或网络故障后残留连接的存活时间。 - var emitter = new SseEmitter(300_000L); + // Heartbeats keep the connection and intermediaries alive; do not let + // Spring's async request timeout close a healthy SSE stream every five minutes. + var emitter = new SseEmitter(0L); var connection = new Connection(userId, accessToken, emitter); connections.compute(userId, (id, existing) -> { Set set = existing == null ? ConcurrentHashMap.newKeySet() : existing; From 4e356f877ab9a0d1ff954d7cfbdb881569f7a5b2 Mon Sep 17 00:00:00 2001 From: LiggMax Date: Mon, 21 Sep 2026 21:24:25 +0800 Subject: [PATCH 19/19] feat: exclude current presence from subject viewers --- .../controller/PresenceController.java | 13 +-- .../flowclient/service/PresenceService.java | 2 +- .../service/PresenceSseService.java | 79 ++++++++++++++++++- .../service/impl/PresenceServiceImpl.java | 32 +++++++- 4 files changed, 112 insertions(+), 14 deletions(-) diff --git a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java index 8247e68..7d0d79c 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java +++ b/flow-client/src/main/java/com/ligg/flowclient/controller/PresenceController.java @@ -1,12 +1,9 @@ package com.ligg.flowclient.controller; import com.ligg.common.exception.AuthorizationException; -import com.ligg.common.response.Result; -import com.ligg.common.statuenum.ResponseCode; import com.ligg.flowclient.annotation.IpEndpointRateLimit; import com.ligg.flowclient.interceptor.AuthorizationInterceptor; import com.ligg.flowclient.module.dto.PresenceHeartbeatDto; -import com.ligg.flowclient.module.vo.OnlineCountVo; import com.ligg.flowclient.service.PresenceService; import com.ligg.flowclient.service.PresenceSseService; import jakarta.validation.Valid; @@ -22,6 +19,7 @@ import org.springframework.web.bind.annotation.RequestAttribute; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestHeader; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @@ -65,9 +63,12 @@ public ResponseEntity onlineCount() { return sseResponse(presenceSseService.subscribeOnlineCount()); } - @GetMapping("/subjects/{subjectId}/online-count") - public Result subjectOnlineCount(@PathVariable int subjectId) { - return Result.success(ResponseCode.SUCCESS, presenceService.subjectOnlineCount(subjectId)); + @GetMapping(value = "/subjects/{subjectId}/online-count", + produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity subjectOnlineCount( + @PathVariable int subjectId, + @RequestParam(required = false) String presenceId) { + return sseResponse(presenceSseService.subscribeSubjectOnlineCount(subjectId, presenceId)); } @GetMapping(value = "/watching-subjects", produces = MediaType.TEXT_EVENT_STREAM_VALUE) diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java index e6af7fc..513650a 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceService.java @@ -15,7 +15,7 @@ void heartbeat(String presenceId, PresenceHeartbeatDto dto, OnlineCountVo onlineCount(); - OnlineCountVo subjectOnlineCount(int subjectId); + OnlineCountVo subjectOnlineCount(int subjectId, String excludePresenceId); List watchingSubjects(); } diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java index a8b7507..9ce45be 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/PresenceSseService.java @@ -35,6 +35,11 @@ public class PresenceSseService implements MessageListener { private final PresenceService presenceService; private final Set onlineCountEmitters = ConcurrentHashMap.newKeySet(); private final Set watchingSubjectEmitters = ConcurrentHashMap.newKeySet(); + private final ConcurrentHashMap> subjectOnlineCountEmitters = + new ConcurrentHashMap<>(); + private final ConcurrentHashMap subjectOnlineCountExclusions = + new ConcurrentHashMap<>(); + private final Set subjectOnlineCountSnapshotDirty = ConcurrentHashMap.newKeySet(); private final AtomicBoolean snapshotDirty = new AtomicBoolean(); private final AtomicBoolean onlineCountSnapshotDirty = new AtomicBoolean(true); private final AtomicBoolean watchingSubjectsSnapshotDirty = new AtomicBoolean(true); @@ -82,6 +87,18 @@ public SseEmitter subscribeWatchingSubjects() { return emitter; } + public SseEmitter subscribeSubjectOnlineCount(int subjectId, String excludePresenceId) { + Set emitters = subjectOnlineCountEmitters.computeIfAbsent( + subjectId, ignored -> ConcurrentHashMap.newKeySet()); + SseEmitter emitter = register(emitters); + if (excludePresenceId != null && !excludePresenceId.isBlank()) { + subjectOnlineCountExclusions.put(emitter, excludePresenceId); + } + subjectOnlineCountSnapshotDirty.add(subjectId); + sendSubjectOnlineCount(emitter, subjectId, excludePresenceId); + return emitter; + } + @Override public void onMessage(Message message, byte[] pattern) { invalidateSnapshots(); @@ -90,7 +107,10 @@ public void onMessage(Message message, byte[] pattern) { private SseEmitter register(Set emitters) { SseEmitter emitter = new SseEmitter(CONNECTION_TIMEOUT_MS); emitters.add(emitter); - Runnable remove = () -> emitters.remove(emitter); + Runnable remove = () -> { + emitters.remove(emitter); + subjectOnlineCountExclusions.remove(emitter); + }; emitter.onCompletion(remove); emitter.onTimeout(remove); emitter.onError(error -> remove.run()); @@ -98,7 +118,9 @@ private SseEmitter register(Set emitters) { } private void markSnapshotDirty() { - if (!onlineCountEmitters.isEmpty() || !watchingSubjectEmitters.isEmpty()) { + if (!onlineCountEmitters.isEmpty() + || !watchingSubjectEmitters.isEmpty() + || !subjectOnlineCountEmitters.isEmpty()) { invalidateSnapshots(); } } @@ -113,6 +135,15 @@ private void flushDirtySnapshot() { if (!watchingSubjectEmitters.isEmpty()) { sendWatchingSubjectsSnapshot(); } + subjectOnlineCountSnapshotDirty.removeIf(subjectId -> { + Set emitters = subjectOnlineCountEmitters.get(subjectId); + if (emitters == null || emitters.isEmpty()) { + subjectOnlineCountEmitters.remove(subjectId); + return true; + } + sendSubjectOnlineCountSnapshot(subjectId, emitters); + return true; + }); } private void sendOnlineCountSnapshot() { @@ -140,6 +171,8 @@ private void sendWatchingSubjectsSnapshot() { private void sendHeartbeats() { onlineCountEmitters.forEach(emitter -> sendHeartbeat(onlineCountEmitters, emitter)); watchingSubjectEmitters.forEach(emitter -> sendHeartbeat(watchingSubjectEmitters, emitter)); + subjectOnlineCountEmitters.forEach((subjectId, emitters) -> + emitters.forEach(emitter -> sendHeartbeat(emitters, emitter))); } private void sendOnlineCount(SseEmitter emitter) { @@ -178,6 +211,32 @@ private void sendWatchingSubjects(SseEmitter emitter, List sn } } + private void sendSubjectOnlineCountSnapshot( + int subjectId, Set emitters) { + emitters.forEach(emitter -> sendSubjectOnlineCount( + emitter, + subjectId, + subjectOnlineCountExclusions.get(emitter))); + } + + private void sendSubjectOnlineCount( + SseEmitter emitter, + int subjectId, + String excludePresenceId) { + try { + OnlineCountVo snapshot = presenceService.subjectOnlineCount( + subjectId, excludePresenceId); + emitter.send(SseEmitter.event() + .name("subject-online-count") + .data(snapshot)); + } catch (Exception error) { + Set emitters = subjectOnlineCountEmitters.get(subjectId); + if (emitters != null) { + close(emitters, emitter); + } + } + } + private void sendHeartbeat(Set emitters, SseEmitter emitter) { try { emitter.send(SseEmitter.event().comment("heartbeat")); @@ -187,14 +246,24 @@ private void sendHeartbeat(Set emitters, SseEmitter emitter) { } private void close(Set emitters, SseEmitter emitter) { - emitters.remove(emitter); - emitter.completeWithError(new IllegalStateException("presence SSE closed")); + // 完成或错误回调可能会先于定时心跳发现发送失败并移除连接， + // 因此关闭操作必须具备幂等性；客户端正常断开不应被视为服务端错误。 + if (!emitters.remove(emitter)) { + return; + } + subjectOnlineCountExclusions.remove(emitter); + try { + emitter.complete(); + } catch (IllegalStateException ignored) { + // 连接可能已被其他线程并发完成。 + } } private void invalidateSnapshots() { snapshotDirty.set(true); onlineCountSnapshotDirty.set(true); watchingSubjectsSnapshotDirty.set(true); + subjectOnlineCountSnapshotDirty.addAll(subjectOnlineCountEmitters.keySet()); } private OnlineCountVo getOnlineCountSnapshot() { @@ -248,6 +317,8 @@ private static boolean isFresh(Object snapshot, long snapshotAt, void stop() { onlineCountEmitters.forEach(SseEmitter::complete); watchingSubjectEmitters.forEach(SseEmitter::complete); + subjectOnlineCountEmitters.values().forEach(emitters -> emitters.forEach(SseEmitter::complete)); + subjectOnlineCountExclusions.clear(); scheduler.shutdownNow(); } } diff --git a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java index 5aa93c5..41dbf7b 100644 --- a/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java +++ b/flow-client/src/main/java/com/ligg/flowclient/service/impl/PresenceServiceImpl.java @@ -137,7 +137,7 @@ public OnlineCountVo onlineCount() { } @Override - public OnlineCountVo subjectOnlineCount(int subjectId) { + public OnlineCountVo subjectOnlineCount(int subjectId, String excludePresenceId) { if (subjectId <= 0) { return new OnlineCountVo(0, 0, 0, 0); } @@ -152,10 +152,17 @@ public OnlineCountVo subjectOnlineCount(int subjectId) { Set users = stringRedisTemplate.opsForZSet() .rangeByScore(usersKey, expiredBefore + 1, now); + PresenceContext excludedPresence = resolveExcludedPresence( + subjectId, excludePresenceId, now); + String excludedUser = excludedPresence == null + ? null : excludedPresence.getDedupeKey(); long anonymous = 0; long loggedIn = 0; if (users != null) { for (String user : users) { + if (Objects.equals(user, excludedUser)) { + continue; + } if (user.startsWith("user:")) { loggedIn++; } else if (user.startsWith("visitor:")) { @@ -164,7 +171,26 @@ public OnlineCountVo subjectOnlineCount(int subjectId) { } } Long devices = stringRedisTemplate.opsForZSet().zCard(devicesKey); - return new OnlineCountVo(anonymous + loggedIn, devices == null ? 0 : devices, anonymous, loggedIn); + long onlineDevices = devices == null ? 0 : devices; + if (excludedPresence != null + && stringRedisTemplate.opsForZSet().score(devicesKey, excludePresenceId) != null) { + onlineDevices = Math.max(0, onlineDevices - 1); + } + return new OnlineCountVo(anonymous + loggedIn, onlineDevices, anonymous, loggedIn); + } + + private PresenceContext resolveExcludedPresence( + int subjectId, String presenceId, long now) { + if (!StringUtils.hasText(presenceId)) { + return null; + } + PresenceContext context = readPreviousContext(recordKey(presenceId)); + if (context == null + || !Objects.equals(context.getSubjectId(), subjectId) + || !isActive(context, now)) { + return null; + } + return context; } private void updateSubjectIndexes(PresenceContext previous, PresenceContext current, long now) { @@ -220,7 +246,7 @@ public List watchingSubjects() { List rankedSubjects = new java.util.ArrayList<>(candidateIds.size()); for (Integer subjectId : candidateIds) { - OnlineCountVo online = subjectOnlineCount(subjectId); + OnlineCountVo online = subjectOnlineCount(subjectId, null); if (online.getOnlineDevices() <= 0 || online.getOnlineUsers() <= 0) { stringRedisTemplate.opsForZSet().remove( Constants.PRESENCE_WATCHING_SUBJECTS_RANK_KEY,