diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java index 3cf622b..ad60849 100644 --- a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java +++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java @@ -8,6 +8,7 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.data.redis.core.SessionCallback; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; import org.springframework.stereotype.Component; import java.time.Duration; @@ -19,7 +20,7 @@ import java.util.Map; /** * 观看进度 Redis 缓冲(write-behind)。 * - *

高并发上报的落地形态:POST 只写 Redis(每次上报 1 次写),由 + *

高并发上报的落地形态:POST 只写 Redis(Lua 脚本一次往返完成写状态与入队),由 * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后批量 UPSERT * 进权威表 watch_progress。Redis 不可用时 put 返回 false,调用方回退同步写库。 * @@ -45,6 +46,22 @@ public class WatchProgressBuffer { private static final int MAX_DRAMAS_PER_IDENTITY = 30; private static final Duration TTL = Duration.ofHours(12); + /** + * 上报热路径的单次往返脚本:写状态、续 TTL、入待落库队列、截断队列一次做完。 + * 拆成多条命令时进程若死在「写状态」与「入队」之间,该剧进度只留在 HASH 里、 + * 永远不会落库,12 小时后被 TTL 清掉。返回 1 表示该身份的剧数超限,需要补充裁剪。 + */ + private static final DefaultRedisScript PUT_SCRIPT = new DefaultRedisScript<>(""" + redis.call('HSET', KEYS[1], ARGV[1], ARGV[2]) + redis.call('EXPIRE', KEYS[1], ARGV[3]) + redis.call('LPUSH', KEYS[2], ARGV[4]) + redis.call('LTRIM', KEYS[2], 0, ARGV[5]) + if redis.call('HLEN', KEYS[1]) > tonumber(ARGV[6]) then + return 1 + end + return 0 + """, Long.class); + private final StringRedisTemplate redis; /** @@ -80,12 +97,18 @@ public class WatchProgressBuffer { String key = identityKey(identity); String hashKey = BUF_KEY_PREFIX + key; String field = Long.toString(dramaId); - redis.opsForHash().put(hashKey, field, encode( - episodeId, positionMs, durationMs, finished, source, deviceId)); - redis.expire(hashKey, TTL); - trimIdentity(hashKey); - redis.opsForList().leftPush(DIRTY_KEY, key + "|" + field); - redis.opsForList().trim(DIRTY_KEY, 0, MAX_DIRTY - 1); + Long overflow = redis.execute(PUT_SCRIPT, + List.of(hashKey, DIRTY_KEY), + field, + encode(episodeId, positionMs, durationMs, finished, source, deviceId), + Long.toString(TTL.toSeconds()), + key + "|" + field, + Long.toString(MAX_DIRTY - 1), + Integer.toString(MAX_DRAMAS_PER_IDENTITY)); + if (overflow != null && overflow > 0) { + // 超出 30 部才做一次 HGETALL + HDEL 裁剪,热路径保持单次往返 + trimIdentity(hashKey); + } return true; } catch (Exception e) { log.warn("观看进度写入 Redis 缓冲失败,本次回退同步写库: {}", e.getMessage());