From 44b7abcd0fe15c0df846d4dd44f7b77aa82e5604 Mon Sep 17 00:00:00 2001 From: Zhou Weidong <502333560@qq.com> Date: Wed, 28 Oct 2026 18:51:52 +0800 Subject: [PATCH] =?UTF-8?q?fix(playback):=20=E5=88=A0=E9=99=A4=E8=A7=82?= =?UTF-8?q?=E7=9C=8B=E5=8E=86=E5=8F=B2=E6=97=B6=E5=90=8C=E6=AD=A5=E6=B8=85?= =?UTF-8?q?=E7=90=86=E8=BF=9B=E5=BA=A6=E7=BC=93=E5=86=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 缓冲已是读接口的权威来源，只删库行会被读覆盖层读回、并被下一次落库把行复活。 缓冲补 remove/clear（HDEL/DEL），删除路径先清缓冲再删库行，读覆盖层按库行 deleted 过滤；落库时缓冲状态已失效的条目直接丢弃，不再整批退回队列。 --- .../playback/service/WatchProgressBuffer.java | 28 +++++++++++++++ .../service/WatchProgressService.java | 34 ++++++++++++++----- 2 files changed, 54 insertions(+), 8 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 3dc36c2..3cf622b 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 @@ -93,6 +93,34 @@ public class WatchProgressBuffer { } } + /** + * 删除某部剧的缓冲状态，用于用户删除单条观看记录。 + * + *

缓冲是读接口的权威来源，删库行必须同步删缓冲，否则会被读覆盖层读回； + * 同时缓冲状态消失后，dirty 队列里该条引用在落库时会被丢弃（见 + * {@code WatchProgressService#flushRefs}），不会再被写回库。 + */ + public boolean remove(InteractionIdentity identity, long dramaId) { + try { + redis.opsForHash().delete(BUF_KEY_PREFIX + identityKey(identity), Long.toString(dramaId)); + return true; + } catch (Exception e) { + log.warn("观看进度缓冲删除失败: {}", e.getMessage()); + return false; + } + } + + /** 清空某身份的全部缓冲状态，用于用户清空观看历史。 */ + public boolean clear(InteractionIdentity identity) { + try { + redis.delete(BUF_KEY_PREFIX + identityKey(identity)); + return true; + } catch (Exception e) { + log.warn("观看进度缓冲清空失败: {}", e.getMessage()); + return false; + } + } + /** 读单个剧的缓冲状态；无缓冲或解码失败返回 null。 */ public BufferedProgress get(InteractionIdentity identity, long dramaId) { return get(identityKey(identity), dramaId); diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java index 27f486c..cd4246b 100644 --- a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java +++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressService.java @@ -279,14 +279,18 @@ public class WatchProgressService { List extraDramaIds = extra.stream() .map(WatchProgressBuffer.BufferedProgress::dramaId) .toList(); - Set withRow = progressMapper.selectAnyForIdentityIn( + Map rowByDrama = progressMapper.selectAnyForIdentityIn( identity.userId(), identity.deviceId(), extraDramaIds) .stream() - .map(WatchProgress::getDramaId) - .collect(Collectors.toSet()); + .collect(Collectors.toMap(WatchProgress::getDramaId, row -> row, (a, b) -> a)); for (WatchProgressBuffer.BufferedProgress buffered : extra) { + WatchProgress existing = rowByDrama.get(buffered.dramaId()); + // 用户已删除该剧（库里是逻辑删除行）：缓冲状态不能被读覆盖层再读回来 + if (existing != null && existing.getDeleted() != null && existing.getDeleted() != 0) { + continue; + } merged.add(transientRow(buffered)); - if (!withRow.contains(buffered.dramaId())) { + if (existing == null) { total++; } } @@ -306,6 +310,12 @@ public class WatchProgressService { @Transactional(rollbackFor = Exception.class) public void remove(InteractionIdentity identity, Long dramaId) { + // 缓冲是读接口的权威来源：先删缓冲再删库行，否则删掉的记录会被读覆盖层读回、 + // 并被下一次落库把行复活。 + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer != null && dramaId != null) { + buffer.remove(identity, dramaId); + } progressMapper.delete(identityQuery(identity) .eq(WatchProgress::getDramaId, dramaId)); } @@ -317,6 +327,10 @@ public class WatchProgressService { @Transactional(rollbackFor = Exception.class) public void clear(InteractionIdentity identity) { + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer != null) { + buffer.clear(identity); + } progressMapper.delete(identityQuery(identity)); } @@ -338,8 +352,7 @@ public class WatchProgressService { return 0; } try { - flushRefs(buffer, refs); - return refs.size(); + return flushRefs(buffer, refs); } catch (Exception ex) { log.warn("观看进度缓冲合并失败，整批退回队列稍后重试: {}", ex.getMessage()); buffer.requeue(refs); @@ -347,7 +360,7 @@ public class WatchProgressService { } } - private void flushRefs(WatchProgressBuffer buffer, List refs) { + private int flushRefs(WatchProgressBuffer buffer, List refs) { // 1) pipeline 批量读缓冲状态，key -> (dramaId -> state) List identityKeys = refs.stream() .map(WatchProgressBuffer.IdentityRef::key) @@ -372,7 +385,11 @@ public class WatchProgressService { ref.userId(), ref.deviceId(), ref.dramaId())); } if (pendingRefs.isEmpty()) { - throw new IllegalStateException("缓冲状态读取为空，ref=" + refs.size()); + // 缓冲状态已消失（用户清空/删除历史，或 HASH 12h TTL 到期）。此时 ref 已无 + // 可落库的内容，必须丢弃：退回队列会在下个周期被原样弹出、再清空、再退回， + // 这批僵尸条目将永久占住 dirty 队列配额。 + log.debug("观看进度缓冲条目已无对应状态，丢弃 {} 条", refs.size()); + return 0; } // 3) 分块成对查询已有行（每块 100 组 OR 索引条件） @@ -415,6 +432,7 @@ public class WatchProgressService { } session.commit(); } + return pendingRefs.size(); } /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */