From 025fbe73f0792c5fd4eba1ac1c247bcec9abb558 Mon Sep 17 00:00:00 2001
From: Zhou Weidong <502333560@qq.com>
Date: Wed, 28 Oct 2026 19:02:32 +0800
Subject: [PATCH] =?UTF-8?q?perf(playback):=20=E4=B8=8A=E6=8A=A5=E6=94=B9?=
=?UTF-8?q?=E7=94=A8=20Lua=20=E5=8D=95=E6=AC=A1=E5=BE=80=E8=BF=94=E5=86=99?=
=?UTF-8?q?=E5=85=A5=E7=BC=93=E5=86=B2?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
原先 HSET、EXPIRE、LPUSH、LTRIM 分四次往返，进程若死在写状态与入队之间，
该剧进度只留在 HASH 里永远不落库、12 小时后被 TTL 清掉。合并成一个脚本后
热路径只需 1 次往返，脚本顺带返回 HLEN 是否超 30 决定要不要补裁剪。
---
.../playback/service/WatchProgressBuffer.java | 37 +++++++++++++++----
1 file changed, 30 insertions(+), 7 deletions(-)
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());