perf(playback): 上报改用 Lua 单次往返写入缓冲

原先 HSET、EXPIRE、LPUSH、LTRIM 分四次往返,进程若死在写状态与入队之间,
该剧进度只留在 HASH 里永远不落库、12 小时后被 TTL 清掉。合并成一个脚本后
热路径只需 1 次往返,脚本顺带返回 HLEN 是否超 30 决定要不要补裁剪。
This commit is contained in:
Zhou Weidong authored and trae-solo-agent committed 2026-10-28 19:02:32 +08:00
1 parent 0d161f318f
commit 025fbe73f0
1 file changed
+30 -7
@@ -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)。
*
* <p>高并发上报的落地形态:POST 只写 Redis(每次上报 1 次写),由
* <p>高并发上报的落地形态: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<Long> 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());