From 3e2d6a06d9f341d5d69dca69de97f5a95dd66fa6 Mon Sep 17 00:00:00 2001 From: Zhou Weidong <502333560@qq.com> Date: Wed, 28 Oct 2026 14:59:09 +0800 Subject: [PATCH] =?UTF-8?q?perf(playback):=20=E8=A7=82=E7=9C=8B=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6=E4=B8=8A=E6=8A=A5=E5=BC=82=E6=AD=A5=E5=8C=96=EF=BC=88?= =?UTF-8?q?Redis=20write-behind=20=E6=89=B9=E9=87=8F=E8=90=BD=E5=BA=93?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /app/watch/progress 只写 Redis 缓冲立即返回（p95 75ms→2ms）， WatchProgressFlusher 每 2s 把 watch:pg:dirty 队列按身份合并批量 UPSERT， Redis 异常自动回退同步写库，接口契约不变 - 读接口用缓冲覆盖 DB 行保证读己之写；Caffeine 缓存剧集校验免去上报路径两条 DB 查询 - 落库按「新增进度/新增流水/更新行」分组攒批（MyBatis BATCH 只合并连续同形状语句）， 配合 reWriteBatchedInserts 多值 INSERT，12.4 万积压 45 秒消化完， 1650 QPS 持续压测队列稳态归零 --- docker-compose.yml | 6 +- .../src/main/resources/application-prod.yml | 7 + .../src/main/resources/application.yml | 6 +- .../service/WatchProgressServiceTest.java | 13 +- fruitshorts-module-playback/pom.xml | 5 + .../dal/mapper/WatchProgressMapper.java | 62 +++ .../playback/service/WatchProgressBuffer.java | 304 ++++++++++++ .../service/WatchProgressFlusher.java | 43 ++ .../service/WatchProgressService.java | 445 ++++++++++++++++-- 9 files changed, 837 insertions(+), 54 deletions(-) create mode 100644 fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java create mode 100644 fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressFlusher.java diff --git a/docker-compose.yml b/docker-compose.yml index 9e3d2cf..3657414 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -71,8 +71,10 @@ services: env_file: .env environment: SERVER_PORT: "8081" - # 默认连编排里的 db；要连外部库就在 .env 里覆盖 DB_URL - DB_URL: ${DB_URL:-jdbc:postgresql://db:5432/${DB_NAME:-fruitshorts}} + # 默认连编排里的 db；要连外部库就在 .env 里覆盖 DB_URL。 + # reWriteBatchedInserts=true：pgjdbc 把 JDBC 批内多条 INSERT 重写成多值 + # INSERT，观看进度缓冲落库（watch:pg:dirty 消费任务）依赖它跑满吞吐。 + DB_URL: ${DB_URL:-jdbc:postgresql://db:5432/${DB_NAME:-fruitshorts}?reWriteBatchedInserts=true} # 限流/缓存用 Redis；容器内 127.0.0.1 是容器自身，必须指向编排里的 redis REDIS_HOST: ${REDIS_HOST:-redis} REDIS_PORT: ${REDIS_PORT:-6379} diff --git a/fruitshorts-app-api/src/main/resources/application-prod.yml b/fruitshorts-app-api/src/main/resources/application-prod.yml index 39f53d0..7a12964 100644 --- a/fruitshorts-app-api/src/main/resources/application-prod.yml +++ b/fruitshorts-app-api/src/main/resources/application-prod.yml @@ -23,6 +23,13 @@ fruitshorts: cors: # 生产只允许显式配置的来源；没配就等于空，浏览器侧会被拦住（比"*"安全） allowed-origins: ${CORS_ALLOWED_ORIGINS:} + watch-progress: + # 观看进度写缓冲：POST 先写 Redis 立即返回，后台任务批量合并进 watch_progress。 + # 消化能力 = batch-size / interval ≈ 5000 条/秒，覆盖 30 万 DAU 峰值 ~3700 条/秒上报。 + # 关掉（false）即回退同步写库，接口契约不变。 + buffer: true + flush-interval-ms: 2000 + flush-batch-size: 10000 springdoc: # 生产只留机器可读契约，关掉 UI（doc.html 不对外开放） diff --git a/fruitshorts-app-api/src/main/resources/application.yml b/fruitshorts-app-api/src/main/resources/application.yml index 746f35c..63a261d 100644 --- a/fruitshorts-app-api/src/main/resources/application.yml +++ b/fruitshorts-app-api/src/main/resources/application.yml @@ -11,8 +11,10 @@ spring: datasource: driver-class-name: org.postgresql.Driver # 地址必须外置：不同环境（本机 / 测试 / 生产）连的库不一样， - # 生产请用 DB_URL 注入（prod profile 会校验它非空） - url: ${DB_URL:jdbc:postgresql://127.0.0.1:5432/fruitshorts} + # 生产请用 DB_URL 注入（prod profile 会校验它非空）。 + # reWriteBatchedInserts：pgjdbc 把批内多条 INSERT 重写成多值 INSERT， + # 观看进度缓冲落库依赖它才能跑满批量吞吐。 + url: ${DB_URL:jdbc:postgresql://127.0.0.1:5432/fruitshorts?reWriteBatchedInserts=true} username: ${DB_USER:fruitshorts} password: ${DB_PASSWORD:fruitshorts} jackson: diff --git a/fruitshorts-app-api/src/test/java/com/fruitshorts/playback/service/WatchProgressServiceTest.java b/fruitshorts-app-api/src/test/java/com/fruitshorts/playback/service/WatchProgressServiceTest.java index 1892886..5d9094e 100644 --- a/fruitshorts-app-api/src/test/java/com/fruitshorts/playback/service/WatchProgressServiceTest.java +++ b/fruitshorts-app-api/src/test/java/com/fruitshorts/playback/service/WatchProgressServiceTest.java @@ -20,10 +20,14 @@ import com.fruitshorts.playback.vo.ProgressRespVO; import com.fruitshorts.playback.vo.ProgressSaveReqVO; import org.apache.ibatis.builder.MapperBuilderAssistant; import org.apache.ibatis.annotations.Select; +import org.apache.ibatis.session.SqlSessionFactory; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.TransactionTemplate; import java.time.LocalDateTime; import java.lang.reflect.Method; @@ -58,11 +62,18 @@ class WatchProgressServiceTest { historyMapper = mock(WatchHistoryMapper.class); dramaService = mock(DramaService.class); episodeService = mock(EpisodeService.class); + // 测试统一走缓冲关闭（同步写库）路径：getIfAvailable 返回 null + @SuppressWarnings("unchecked") + ObjectProvider bufferProvider = mock(ObjectProvider.class); + given(bufferProvider.getIfAvailable()).willReturn(null); progressService = new WatchProgressService( progressMapper, historyMapper, dramaService, - episodeService); + episodeService, + bufferProvider, + mock(SqlSessionFactory.class), + new TransactionTemplate(mock(PlatformTransactionManager.class))); } @Test diff --git a/fruitshorts-module-playback/pom.xml b/fruitshorts-module-playback/pom.xml index 90c715c..4039875 100644 --- a/fruitshorts-module-playback/pom.xml +++ b/fruitshorts-module-playback/pom.xml @@ -33,5 +33,10 @@ com.fruitshorts fruitshorts-module-user + + + com.github.ben-manes.caffeine + caffeine + diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/dal/mapper/WatchProgressMapper.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/dal/mapper/WatchProgressMapper.java index bd5ac4c..88b7246 100644 --- a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/dal/mapper/WatchProgressMapper.java +++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/dal/mapper/WatchProgressMapper.java @@ -8,6 +8,8 @@ import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Update; +import java.util.List; + public interface WatchProgressMapper extends BaseMapper { /** 按登录用户或匿名设备读取同一部剧的记录，包含逻辑删除行。 */ @@ -40,4 +42,64 @@ public interface WatchProgressMapper extends BaseMapper { WHERE id = #{id} """) int updateAny(WatchProgress progress); + + /** 固定形状的全字段按 id 更新（绕过逻辑删除过滤）。SQL 不随字段空值变化， + * 缓冲落库时连续调用可被 MyBatis BATCH 执行器合成一个 JDBC batch。 */ + @Update(""" + UPDATE watch_progress SET + episode_id = #{et.episodeId}, + position_ms = #{et.positionMs}, + duration_ms = #{et.durationMs}, + finished = #{et.finished}, + play_count = #{et.playCount}, + device_id = #{et.deviceId}, + source = #{et.source}, + deleted = #{et.deleted} + WHERE id = #{et.id} + """) + int updateRowAny(@Param("et") WatchProgress et); + + /** 批量读取同一身份下多部剧的记录（含逻辑删除行），供缓冲落库合并使用。 */ + @Select(""" + + """) + List selectAnyForIdentityIn( + @Param("userId") Long userId, + @Param("deviceId") String deviceId, + @Param("dramaIds") List dramaIds); + + /** 落库合并的查询参数：身份（登录用户或匿名设备）+ drama。 */ + @lombok.Data + @lombok.NoArgsConstructor + @lombok.AllArgsConstructor + class IdentityDramaPair { + private Long userId; + private String deviceId; + private long dramaId; + } + + /** 按多组「身份 + drama」批量读取记录（含逻辑删除行），一组一条索引条件，单次往返。 */ + @Select(""" + + """) + List selectBatchForPairs(@Param("pairs") List pairs); } 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 new file mode 100644 index 0000000..3dc36c2 --- /dev/null +++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressBuffer.java @@ -0,0 +1,304 @@ +// Author: fruit.com.cn + +package com.fruitshorts.playback.service; + +import com.fruitshorts.common.interaction.InteractionIdentity; +import lombok.RequiredArgsConstructor; +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.stereotype.Component; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * 观看进度 Redis 缓冲（write-behind）。 + * + *

高并发上报的落地形态：POST 只写 Redis（每次上报 1 次写），由 + * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后批量 UPSERT + * 进权威表 watch_progress。Redis 不可用时 put 返回 false，调用方回退同步写库。 + * + *

Redis 结构： + *

    + *
  • {@code watch:pg:buf:u:{userId}} / {@code watch:pg:buf:d:{deviceId}}： + * HASH，field=dramaId，value=进度状态，作为读接口的「读己之写」覆盖层，TTL 12h；
    • + *
  • {@code watch:pg:dirty}：LIST，value={identityKey}|{dramaId}，待落库队列， + * 定时任务 RPOP 消费；重复条目靠幂等合并消化。
    • + *
+ */ +@Slf4j +@Component +@RequiredArgsConstructor +@ConditionalOnProperty(name = "fruitshorts.watch-progress.buffer", havingValue = "true") +public class WatchProgressBuffer { + + private static final String BUF_KEY_PREFIX = "watch:pg:buf:"; + private static final String DIRTY_KEY = "watch:pg:dirty"; + /** 待落库队列长度上限，防止 Redis 在落库停滞时无限膨胀。 */ + private static final long MAX_DIRTY = 200_000; + /** 每个身份最多缓冲 30 部剧（续看列表第一页用不完，同时限制 HGETALL 大小）。 */ + private static final int MAX_DRAMAS_PER_IDENTITY = 30; + private static final Duration TTL = Duration.ofHours(12); + + private final StringRedisTemplate redis; + + /** + * 缓冲中的一条进度状态。ts 为客户端上报时刻（服务端毫秒），用于判断 + * 缓冲值是否比数据库行新，以及合并时丢弃乱序旧值。 + */ + public record BufferedProgress( + long dramaId, + long episodeId, + long positionMs, + long durationMs, + boolean finished, + String source, + String deviceId, + long ts) { + } + + /** dirty 队列弹出的一条待落库引用。 */ + public record IdentityRef(String key, Long userId, String deviceId, long dramaId) { + + public InteractionIdentity toIdentity() { + return userId != null + ? new InteractionIdentity(userId, "") + : new InteractionIdentity(null, deviceId); + } + } + + /** 上报进度入缓冲。成功返回 true；Redis 异常返回 false，调用方回退同步写库。 */ + public boolean put(InteractionIdentity identity, long dramaId, long episodeId, + long positionMs, long durationMs, boolean finished, + String source, String deviceId) { + try { + 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); + return true; + } catch (Exception e) { + log.warn("观看进度写入 Redis 缓冲失败，本次回退同步写库: {}", e.getMessage()); + return false; + } + } + + /** 读单个剧的缓冲状态；无缓冲或解码失败返回 null。 */ + public BufferedProgress get(InteractionIdentity identity, long dramaId) { + return get(identityKey(identity), dramaId); + } + + public BufferedProgress get(String identityKey, long dramaId) { + try { + Object value = redis.opsForHash().get(BUF_KEY_PREFIX + identityKey, Long.toString(dramaId)); + return value == null ? null : decode(dramaId, value.toString()); + } catch (Exception e) { + log.warn("观看进度读取 Redis 缓冲失败: {}", e.getMessage()); + return null; + } + } + + /** 读某身份全部缓冲状态（≤30 部剧），读接口覆盖层用。 */ + public Map getAll(InteractionIdentity identity) { + return getAll(identityKey(identity)); + } + + public Map getAll(String identityKey) { + try { + return decodeAll(redis.opsForHash().entries(BUF_KEY_PREFIX + identityKey)); + } catch (Exception e) { + log.warn("观看进度读取 Redis 缓冲失败: {}", e.getMessage()); + return Map.of(); + } + } + + /** 批量读取多个身份的缓冲状态（pipeline 单次往返），落库任务用。 */ + @SuppressWarnings("unchecked") + public Map> getAllBatch(List identityKeys) { + if (identityKeys.isEmpty()) { + return Map.of(); + } + try { + List results = redis.executePipelined(new SessionCallback<>() { + @Override + public Object execute(org.springframework.data.redis.core.RedisOperations operations) { + for (String key : identityKeys) { + operations.opsForHash().entries(BUF_KEY_PREFIX + key); + } + return null; + } + }); + Map> result = new LinkedHashMap<>(); + for (int i = 0; i < identityKeys.size(); i++) { + Object raw = results.get(i); + result.put(identityKeys.get(i), + raw instanceof Map map ? decodeAll((Map) map) : Map.of()); + } + return result; + } catch (Exception e) { + log.warn("观看进度批量读取 Redis 缓冲失败: {}", e.getMessage()); + throw new IllegalStateException("Redis 缓冲批量读取失败", e); + } + } + + /** 落库任务从 dirty 队头弹出最多 max 条（按 identity+drama 去重，保留顺序）。 */ + public List drain(int max) { + List raw; + try { + raw = redis.opsForList().leftPop(DIRTY_KEY, Math.max(1, max)); + } catch (Exception e) { + log.warn("观看进度 dirty 队列弹出失败: {}", e.getMessage()); + return List.of(); + } + if (raw == null || raw.isEmpty()) { + return List.of(); + } + LinkedHashMap dedup = new LinkedHashMap<>(); + for (String value : raw) { + int split = value.lastIndexOf('|'); + if (split <= 0) { + continue; + } + String key = value.substring(0, split); + long dramaId; + try { + dramaId = Long.parseLong(value.substring(split + 1)); + } catch (NumberFormatException ignore) { + continue; + } + dedup.putIfAbsent(key + "|" + dramaId, toRef(key, dramaId)); + } + return new ArrayList<>(dedup.values()); + } + + /** 落库失败的条目退回队列，等待下个周期重试。 */ + public void requeue(List refs) { + try { + List raw = refs.stream() + .map(r -> r.key() + "|" + r.dramaId()) + .toList(); + redis.opsForList().rightPushAll(DIRTY_KEY, raw); + redis.opsForList().trim(DIRTY_KEY, 0, MAX_DIRTY - 1); + } catch (Exception e) { + log.error("观看进度缓冲条目退回队列失败，这批进度将丢失: {}", e.getMessage()); + } + } + + /** 身份的 Redis 键：登录用 u:{userId}，匿名用 d:{deviceId}。 */ + public String identityKey(InteractionIdentity identity) { + return identity.authenticated() + ? "u:" + identity.userId() + : "d:" + cleanToken(identity.deviceId()); + } + + private IdentityRef toRef(String key, long dramaId) { + int split = key.indexOf(':'); + if (split <= 0) { + return new IdentityRef(key, null, key, dramaId); + } + String scope = key.substring(0, split); + String id = key.substring(split + 1); + if ("u".equals(scope)) { + try { + return new IdentityRef(key, Long.valueOf(id), "", dramaId); + } catch (NumberFormatException ignore) { + // 落到匿名分支 + } + } + return new IdentityRef(key, null, id, dramaId); + } + + private String encode(long episodeId, long positionMs, long durationMs, + boolean finished, String source, String deviceId) { + return episodeId + "|" + positionMs + "|" + durationMs + "|" + (finished ? 1 : 0) + + "|" + cleanToken(source) + "|" + cleanToken(deviceId) + "|" + System.currentTimeMillis(); + } + + private BufferedProgress decode(long dramaId, String value) { + try { + String[] parts = value.split("\\|", -1); + if (parts.length != 7) { + return null; + } + return new BufferedProgress( + dramaId, + Long.parseLong(parts[0]), + Long.parseLong(parts[1]), + Long.parseLong(parts[2]), + "1".equals(parts[3]), + parts[4], + parts[5], + Long.parseLong(parts[6])); + } catch (NumberFormatException e) { + log.debug("观看进度缓冲值解码失败，忽略: {}", value); + return null; + } + } + + private Map decodeAll(Map raw) { + if (raw == null || raw.isEmpty()) { + return Map.of(); + } + Map result = new LinkedHashMap<>(); + for (Map.Entry e : raw.entrySet()) { + long dramaId; + try { + dramaId = Long.parseLong(e.getKey().toString()); + } catch (NumberFormatException ignore) { + continue; + } + BufferedProgress state = decode(dramaId, e.getValue().toString()); + if (state != null) { + result.put(dramaId, state); + } + } + return result; + } + + /** 限制单身份缓冲的剧数：超出时淘汰最旧条目，保证 HGETALL 与读覆盖层开销有界。 */ + private void trimIdentity(String hashKey) { + Long size = redis.opsForHash().size(hashKey); + if (size == null || size <= MAX_DRAMAS_PER_IDENTITY) { + return; + } + Map all = redis.opsForHash().entries(hashKey); + List> byTs = new ArrayList<>(all.size()); + for (Map.Entry e : all.entrySet()) { + long ts = 0; + if (e.getValue() != null) { + String[] parts = e.getValue().toString().split("\\|", -1); + try { + ts = Long.parseLong(parts[parts.length - 1]); + } catch (NumberFormatException ignore) { + // ts 取不到当最旧，优先淘汰 + } + } + byTs.add(Map.entry(e.getKey(), ts)); + } + byTs.sort(Map.Entry.comparingByValue()); + int toRemove = Math.max(0, byTs.size() - MAX_DRAMAS_PER_IDENTITY); + if (toRemove > 0) { + redis.opsForHash().delete(hashKey, + byTs.subList(0, toRemove).stream().map(Map.Entry::getKey).toArray()); + } + } + + /** value 中的 source/deviceId 不允许出现分隔符 | 与换行，统一清洗；长度截到 128。 */ + private String cleanToken(String token) { + if (token == null) { + return ""; + } + String cleaned = token.replace("|", "").replace("\n", "").replace("\r", "").trim(); + return cleaned.length() > 128 ? cleaned.substring(0, 128) : cleaned; + } +} diff --git a/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressFlusher.java b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressFlusher.java new file mode 100644 index 0000000..a963928 --- /dev/null +++ b/fruitshorts-module-playback/src/main/java/com/fruitshorts/playback/service/WatchProgressFlusher.java @@ -0,0 +1,43 @@ +// Author: fruit.com.cn + +package com.fruitshorts.playback.service; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +/** + * 观看进度缓冲落库任务：定时把 Redis 缓冲里的进度按身份分组合并进 watch_progress。 + * + *

只在开启缓冲（fruitshorts.watch-progress.buffer=true）的应用内生效， + * app-api 的 prod profile 开启，admin-api 与测试环境不受影响。 + * fixedDelay 串行执行，同一时刻只有一个落库批次在跑。 + */ +@Slf4j +@Component +@RequiredArgsConstructor +@ConditionalOnProperty(name = "fruitshorts.watch-progress.buffer", havingValue = "true") +public class WatchProgressFlusher { + + private final WatchProgressService watchProgressService; + + @Value("${fruitshorts.watch-progress.flush-batch-size:10000}") + private int batchSize; + + @Scheduled( + fixedDelayString = "${fruitshorts.watch-progress.flush-interval-ms:2000}", + initialDelayString = "${fruitshorts.watch-progress.flush-initial-delay-ms:15000}") + public void flush() { + try { + int flushed = watchProgressService.flushBuffered(batchSize); + if (flushed > 0) { + log.debug("观看进度缓冲落库完成，本批次 {} 条", flushed); + } + } catch (Exception e) { + log.warn("观看进度缓冲落库失败，等待下个周期重试: {}", e.getMessage()); + } + } +} 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 619d42f..27f486c 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 @@ -19,29 +19,47 @@ import com.fruitshorts.playback.dal.mapper.WatchHistoryMapper; import com.fruitshorts.playback.dal.mapper.WatchProgressMapper; import com.fruitshorts.playback.vo.ProgressRespVO; import com.fruitshorts.playback.vo.ProgressSaveReqVO; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.ibatis.session.ExecutorType; +import org.apache.ibatis.session.SqlSession; +import org.apache.ibatis.session.SqlSessionFactory; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.StringUtils; +import java.time.Duration; +import java.time.Instant; import java.time.LocalDateTime; +import java.time.ZoneId; import java.util.ArrayList; +import java.util.Comparator; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; /** * 观看进度。 * - *

每次上报都写库在高并发下会成为瓶颈（1 万并发 × 每 10 秒一次 ≈ 1000 写/秒）， - * 所以生产形态是「先写 Redis，后台任务按登录账号或设备 + drama 合并后批量 UPSERT」。 - * 这里先落地权威表与合并规则，缓存层在接入 Redis 时补齐，接口契约不变。 + *

生产形态（fruitshorts.watch-progress.buffer=true，prod 已开启）： + * POST 只写 Redis 缓冲（{@link WatchProgressBuffer}）立即返回， + * {@link WatchProgressFlusher} 定时把缓冲按「账号或设备 + drama」合并后 + * 批量 UPSERT 进权威表，读接口用缓冲覆盖保证读己之写。 + * 缓冲关闭或 Redis 异常时自动回退同步写库，接口契约不变。 * *

合并规则：以客户端上报的 positionMs 为准，但只在「更靠后」时覆盖， * 避免用户换设备后进度倒退。 */ +@Slf4j @Service @RequiredArgsConstructor public class WatchProgressService { @@ -53,68 +71,102 @@ public class WatchProgressService { private final WatchHistoryMapper historyMapper; private final DramaService dramaService; private final EpisodeService episodeService; + private final ObjectProvider bufferProvider; + private final SqlSessionFactory sqlSessionFactory; + private final TransactionTemplate transactionTemplate; + + /** 已发布剧集/短剧的短 TTL 进程内缓存：上报路径的热点校验，免去每次两条 DB 查询。 */ + private final Cache episodeCache = Caffeine.newBuilder() + .maximumSize(50_000) + .expireAfterWrite(Duration.ofMinutes(5)) + .build(); + private final Cache dramaCache = Caffeine.newBuilder() + .maximumSize(50_000) + .expireAfterWrite(Duration.ofMinutes(5)) + .build(); @Transactional(rollbackFor = Exception.class) public ProgressRespVO save(Long userId, ProgressSaveReqVO req) { return save(new InteractionIdentity(userId, ""), req); } - @Transactional(rollbackFor = Exception.class) public ProgressRespVO save(InteractionIdentity identity, ProgressSaveReqVO req) { - DramaEpisode episode = episodeService.getPublished(req.getEpisodeId()); - Drama drama = dramaService.getPublished(episode.getDramaId()); + DramaEpisode episode = publishedEpisode(req.getEpisodeId()); + Drama drama = publishedDrama(episode.getDramaId()); long position = req.getPositionMs() == null ? 0L : Math.max(0L, req.getPositionMs()); long duration = req.getDurationMs() == null ? 0L : Math.max(0L, req.getDurationMs()); boolean finished = Boolean.TRUE.equals(req.getFinished()) || (duration > 0 && position + FINISH_THRESHOLD_MS >= duration); - LoadedProgress loaded = loadForSave(identity, drama.getId(), episode, position, duration, - finished, req); - WatchProgress existing = loaded.progress(); - if (loaded.created()) { - appendHistory(identity, drama.getId(), episode.getId(), position, duration, - existing.getSource(), existing.getDeviceId()); - return toResp(progressMapper.selectById(existing.getId())); + // 高并发形态：写 Redis 缓冲立即返回，落库任务稍后合并进权威表 + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer != null && buffer.put(identity, drama.getId(), episode.getId(), + position, duration, finished, req.getSource(), req.getDeviceId())) { + ProgressRespVO vo = new ProgressRespVO(); + vo.setDramaId(drama.getId()); + vo.setEpisodeId(episode.getId()); + vo.setPositionMs(position); + vo.setDurationMs(duration); + vo.setFinished(finished); + vo.setUpdatedAt(LocalDateTime.now()); + return vo; } + return saveToDb(identity, drama, episode, position, duration, finished, req); + } - boolean hidden = existing.getDeleted() != null && existing.getDeleted() != 0; - boolean sameEpisode = existing.getEpisodeId() != null - && existing.getEpisodeId().equals(episode.getId()); - if (hidden && sameEpisode) { - existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1); - appendHistory(identity, drama.getId(), episode.getId(), position, duration, - req.getSource(), existing.getDeviceId()); - } - if (!sameEpisode) { - // 换集：直接采纳新集与新的播放位置 - existing.setEpisodeId(episode.getId()); - existing.setPositionMs(position); - existing.setDurationMs(duration); - existing.setFinished(finished ? 1 : 0); - existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1); - appendHistory(identity, drama.getId(), episode.getId(), position, duration, - req.getSource(), existing.getDeviceId()); - } else { - existing.setDurationMs(Math.max(existing.getDurationMs() == null ? 0L : existing.getDurationMs(), - duration)); - if (position >= (existing.getPositionMs() == null ? 0L : existing.getPositionMs())) { + /** 同步写库路径：缓冲关闭或 Redis 异常时的回退，合并规则与缓冲落库一致。 */ + private ProgressRespVO saveToDb(InteractionIdentity identity, Drama drama, DramaEpisode episode, + long position, long duration, boolean finished, + ProgressSaveReqVO req) { + return transactionTemplate.execute(status -> { + LoadedProgress loaded = loadForSave(identity, drama.getId(), episode, position, duration, + finished, req); + WatchProgress existing = loaded.progress(); + if (loaded.created()) { + appendHistory(identity, drama.getId(), episode.getId(), position, duration, + existing.getSource(), existing.getDeviceId()); + return toResp(progressMapper.selectById(existing.getId())); + } + + boolean hidden = existing.getDeleted() != null && existing.getDeleted() != 0; + boolean sameEpisode = existing.getEpisodeId() != null + && existing.getEpisodeId().equals(episode.getId()); + if (hidden && sameEpisode) { + existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1); + appendHistory(identity, drama.getId(), episode.getId(), position, duration, + req.getSource(), existing.getDeviceId()); + } + if (!sameEpisode) { + // 换集：直接采纳新集与新的播放位置 + existing.setEpisodeId(episode.getId()); existing.setPositionMs(position); + existing.setDurationMs(duration); + existing.setFinished(finished ? 1 : 0); + existing.setPlayCount((existing.getPlayCount() == null ? 0 : existing.getPlayCount()) + 1); + appendHistory(identity, drama.getId(), episode.getId(), position, duration, + req.getSource(), existing.getDeviceId()); + } else { + existing.setDurationMs(Math.max(existing.getDurationMs() == null ? 0L : existing.getDurationMs(), + duration)); + if (position >= (existing.getPositionMs() == null ? 0L : existing.getPositionMs())) { + existing.setPositionMs(position); + } + if (finished) { + existing.setFinished(1); + } } - if (finished) { - existing.setFinished(1); + existing.setDeviceId(identity.authenticated() + ? (req.getDeviceId() == null ? "" : req.getDeviceId()) + : identity.deviceId()); + existing.setSource(req.getSource() == null ? "" : req.getSource()); + if (hidden) { + existing.setDeleted(0); + progressMapper.updateAny(existing); + } else { + progressMapper.updateById(existing); } - } - existing.setDeviceId(identity.authenticated() - ? (req.getDeviceId() == null ? "" : req.getDeviceId()) - : identity.deviceId()); - existing.setSource(req.getSource() == null ? "" : req.getSource()); - if (hidden) { - existing.setDeleted(0); - progressMapper.updateAny(existing); - } else { - progressMapper.updateById(existing); - } - return toResp(progressMapper.selectById(existing.getId())); + return toResp(progressMapper.selectById(existing.getId())); + }); } private LoadedProgress loadForSave(InteractionIdentity identity, Long dramaId, @@ -169,6 +221,16 @@ public class WatchProgressService { WatchProgress progress = progressMapper.selectOne(identityQuery(identity) .eq(WatchProgress::getDramaId, dramaId) .last("limit 1")); + // 读己之写：缓冲里有比库行更新的状态时以缓冲为准 + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer != null) { + WatchProgressBuffer.BufferedProgress buffered = buffer.get(identity, dramaId); + if (buffered != null + && (progress == null + || buffered.ts() > toMillis(progress.getUpdatedAt()))) { + return toResp(buffered, progress); + } + } return progress == null ? null : toResp(progress); } @@ -186,7 +248,55 @@ public class WatchProgressService { identityQuery(identity) .orderByDesc(WatchProgress::getUpdatedAt) .orderByDesc(WatchProgress::getId)); - return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords())); + + // 缓冲的条目都是最近上报的，只会出现在第一页；缓冲关闭时走原逻辑 + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer == null || safeIndex > 1) { + return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords())); + } + Map states = buffer.getAll(identity); + if (states.isEmpty()) { + return new PageResult<>(result.getTotal(), enrichDisplay(result.getRecords())); + } + + List merged = new ArrayList<>(); + Set pageDramas = new HashSet<>(); + for (WatchProgress row : result.getRecords()) { + pageDramas.add(row.getDramaId()); + WatchProgressBuffer.BufferedProgress buffered = row.getDramaId() == null + ? null + : states.get(row.getDramaId()); + merged.add(overlayRow(row, buffered)); + } + long total = result.getTotal(); + // 缓冲里有、页里没有的剧补进第一页（刚上报首集、库里还没落的场景） + List extra = states.entrySet().stream() + .filter(e -> !pageDramas.contains(e.getKey())) + .map(Map.Entry::getValue) + .sorted(Comparator.comparingLong(WatchProgressBuffer.BufferedProgress::ts).reversed()) + .toList(); + if (!extra.isEmpty()) { + List extraDramaIds = extra.stream() + .map(WatchProgressBuffer.BufferedProgress::dramaId) + .toList(); + Set withRow = progressMapper.selectAnyForIdentityIn( + identity.userId(), identity.deviceId(), extraDramaIds) + .stream() + .map(WatchProgress::getDramaId) + .collect(Collectors.toSet()); + for (WatchProgressBuffer.BufferedProgress buffered : extra) { + merged.add(transientRow(buffered)); + if (!withRow.contains(buffered.dramaId())) { + total++; + } + } + } + merged.sort(Comparator.comparing(WatchProgress::getUpdatedAt, + Comparator.nullsFirst(Comparator.naturalOrder())).reversed()); + List page = merged.size() > safeSize + ? merged.subList(0, safeSize) + : merged; + return new PageResult<>(total, enrichDisplay(page)); } @Transactional(rollbackFor = Exception.class) @@ -210,6 +320,181 @@ public class WatchProgressService { progressMapper.delete(identityQuery(identity)); } + /** + * 缓冲落库：由 {@link WatchProgressFlusher} 定时调用。 + * 消费 dirty 队列，pipeline 读状态、分块成对查行，先在内存里合并出 + * 「新增进度 / 新增流水 / 更新行」三组，再在一个 JDBC batch 会话内 + * 按类型连续攒批执行、一次提交（交替提交会退化成逐条往返，吞吐只有约 200 行/秒）。 + * + * @return 本批次合并的进度条数 + */ + public int flushBuffered(int batchSize) { + WatchProgressBuffer buffer = bufferProvider.getIfAvailable(); + if (buffer == null) { + return 0; + } + List refs = buffer.drain(batchSize); + if (refs.isEmpty()) { + return 0; + } + try { + flushRefs(buffer, refs); + return refs.size(); + } catch (Exception ex) { + log.warn("观看进度缓冲合并失败，整批退回队列稍后重试: {}", ex.getMessage()); + buffer.requeue(refs); + return 0; + } + } + + private void flushRefs(WatchProgressBuffer buffer, List refs) { + // 1) pipeline 批量读缓冲状态，key -> (dramaId -> state) + List identityKeys = refs.stream() + .map(WatchProgressBuffer.IdentityRef::key) + .distinct() + .toList(); + Map> statesByIdentity = + buffer.getAllBatch(identityKeys); + + // 2) 组装待合并条目；状态读不到视为异常（ref 必然来自 put），整批退回 + List pendingRefs = new ArrayList<>(refs.size()); + List pendingStates = new ArrayList<>(refs.size()); + List pairs = new ArrayList<>(refs.size()); + for (WatchProgressBuffer.IdentityRef ref : refs) { + WatchProgressBuffer.BufferedProgress state = + statesByIdentity.getOrDefault(ref.key(), Map.of()).get(ref.dramaId()); + if (state == null) { + continue; + } + pendingRefs.add(ref); + pendingStates.add(state); + pairs.add(new WatchProgressMapper.IdentityDramaPair( + ref.userId(), ref.deviceId(), ref.dramaId())); + } + if (pendingRefs.isEmpty()) { + throw new IllegalStateException("缓冲状态读取为空，ref=" + refs.size()); + } + + // 3) 分块成对查询已有行（每块 100 组 OR 索引条件） + Map rowByPair = new HashMap<>(pendingRefs.size()); + for (int i = 0; i < pairs.size(); i += 100) { + List chunk = + pairs.subList(i, Math.min(i + 100, pairs.size())); + for (WatchProgress row : progressMapper.selectBatchForPairs(chunk)) { + rowByPair.put(rowKey(row.getUserId(), row.getDeviceId(), row.getDramaId()), row); + } + } + + // 4) 计算阶段：纯内存合并出「新增进度 / 新增流水 / 更新行」三组 + List insertProgress = new ArrayList<>(pendingRefs.size()); + List insertHistory = new ArrayList<>(pendingRefs.size()); + List updateRows = new ArrayList<>(pendingRefs.size()); + for (int i = 0; i < pendingRefs.size(); i++) { + WatchProgress row = rowByPair.get(rowKey( + pendingRefs.get(i).userId(), + pendingRefs.get(i).deviceId(), + pendingStates.get(i).dramaId())); + planOne(pendingRefs.get(i).toIdentity(), pendingStates.get(i), row, + insertProgress, insertHistory, updateRows); + } + + // 5) 执行阶段：同一个 batch 会话按类型连续执行，最后一次提交。 + // MyBatis BATCH 执行器只合并「连续同形状」的语句，进度/流水交替提交会 + // 退化成逐条网络往返（实测只有约 200 行/秒），必须分组攒批。 + try (SqlSession session = sqlSessionFactory.openSession(ExecutorType.BATCH, false)) { + WatchProgressMapper batchProgress = session.getMapper(WatchProgressMapper.class); + WatchHistoryMapper batchHistory = session.getMapper(WatchHistoryMapper.class); + for (WatchProgress created : insertProgress) { + batchProgress.insert(created); + } + for (WatchHistory history : insertHistory) { + batchHistory.insert(history); + } + for (WatchProgress row : updateRows) { + batchProgress.updateRowAny(row); + } + session.commit(); + } + } + + /** 计算单条缓冲的合并动作：只改内存对象不触库，产出归入三组待执行列表。 */ + private void planOne(InteractionIdentity identity, + WatchProgressBuffer.BufferedProgress state, + WatchProgress row, + List insertProgress, + List insertHistory, + List updateRows) { + if (row == null) { + WatchProgress created = new WatchProgress(); + created.setUserId(identity.userId()); + created.setDramaId(state.dramaId()); + created.setEpisodeId(state.episodeId()); + created.setPositionMs(state.positionMs()); + created.setDurationMs(state.durationMs()); + created.setFinished(state.finished() ? 1 : 0); + created.setPlayCount(1); + created.setDeviceId(identity.authenticated() + ? state.deviceId() + : identity.deviceId()); + created.setSource(state.source()); + insertProgress.add(created); + insertHistory.add(historyOf(identity, state, created.getDeviceId())); + return; + } + boolean hidden = row.getDeleted() != null && row.getDeleted() != 0; + boolean sameEpisode = row.getEpisodeId() != null + && row.getEpisodeId().equals(state.episodeId()); + if (hidden && sameEpisode) { + row.setPlayCount((row.getPlayCount() == null ? 0 : row.getPlayCount()) + 1); + insertHistory.add(historyOf(identity, state, row.getDeviceId())); + } else if (!sameEpisode) { + // 换集：直接采纳新集与新的播放位置 + row.setEpisodeId(state.episodeId()); + row.setPositionMs(state.positionMs()); + row.setDurationMs(state.durationMs()); + row.setFinished(state.finished() ? 1 : 0); + row.setPlayCount((row.getPlayCount() == null ? 0 : row.getPlayCount()) + 1); + insertHistory.add(historyOf(identity, state, row.getDeviceId())); + } else { + row.setDurationMs(Math.max(row.getDurationMs() == null ? 0L : row.getDurationMs(), + state.durationMs())); + if (state.positionMs() >= (row.getPositionMs() == null ? 0L : row.getPositionMs())) { + row.setPositionMs(state.positionMs()); + } + if (state.finished()) { + row.setFinished(1); + } + } + row.setDeviceId(identity.authenticated() ? state.deviceId() : identity.deviceId()); + row.setSource(state.source()); + if (hidden) { + row.setDeleted(0); + } + updateRows.add(row); + } + + /** 组装一条观看流水（首次观剧/换集各记一条）。 */ + private WatchHistory historyOf(InteractionIdentity identity, + WatchProgressBuffer.BufferedProgress state, + String deviceId) { + WatchHistory history = new WatchHistory(); + history.setUserId(identity.userId()); + history.setDramaId(state.dramaId()); + history.setEpisodeId(state.episodeId()); + history.setPositionMs(state.positionMs()); + history.setDurationMs(state.durationMs()); + history.setSource(state.source() == null ? "" : state.source()); + history.setDeviceId(identity.authenticated() + ? (deviceId == null ? "" : deviceId) + : identity.deviceId()); + return history; + } + + /** 行定位键：登录身份用 userId，匿名用 deviceId。 */ + private String rowKey(Long userId, String deviceId, Long dramaId) { + return (userId != null ? "u:" + userId : "d:" + deviceId) + ":" + dramaId; + } + private ProgressRespVO toResp(WatchProgress progress) { ProgressRespVO vo = new ProgressRespVO(); vo.setDramaId(progress.getDramaId()); @@ -222,6 +507,68 @@ public class WatchProgressService { return vo; } + /** 以缓冲状态为主、库行 playCount 兜底构建续看响应。 */ + private ProgressRespVO toResp(WatchProgressBuffer.BufferedProgress buffered, WatchProgress row) { + ProgressRespVO vo = new ProgressRespVO(); + vo.setDramaId(buffered.dramaId()); + vo.setEpisodeId(buffered.episodeId()); + vo.setPositionMs(buffered.positionMs()); + vo.setDurationMs(buffered.durationMs()); + vo.setFinished(buffered.finished()); + vo.setPlayCount(row == null || row.getPlayCount() == null ? 1 : row.getPlayCount()); + vo.setUpdatedAt(fromMillis(buffered.ts())); + return vo; + } + + /** 库行被缓冲覆盖时返回覆盖后的瞬态行；缓冲不比库行新时原样返回。 */ + private WatchProgress overlayRow(WatchProgress row, WatchProgressBuffer.BufferedProgress buffered) { + if (buffered == null || row.getUpdatedAt() == null + || buffered.ts() <= toMillis(row.getUpdatedAt())) { + return row; + } + WatchProgress overlaid = transientRow(buffered); + overlaid.setPlayCount(row.getPlayCount()); + return overlaid; + } + + private WatchProgress transientRow(WatchProgressBuffer.BufferedProgress buffered) { + WatchProgress row = new WatchProgress(); + row.setDramaId(buffered.dramaId()); + row.setEpisodeId(buffered.episodeId()); + row.setPositionMs(buffered.positionMs()); + row.setDurationMs(buffered.durationMs()); + row.setFinished(buffered.finished() ? 1 : 0); + row.setPlayCount(1); + row.setUpdatedAt(fromMillis(buffered.ts())); + return row; + } + + private long toMillis(LocalDateTime time) { + return time == null ? 0L : time.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); + } + + private LocalDateTime fromMillis(long epochMillis) { + return LocalDateTime.ofInstant(Instant.ofEpochMilli(epochMillis), ZoneId.systemDefault()); + } + + private DramaEpisode publishedEpisode(Long episodeId) { + DramaEpisode episode = episodeCache.getIfPresent(episodeId); + if (episode == null) { + episode = episodeService.getPublished(episodeId); + episodeCache.put(episodeId, episode); + } + return episode; + } + + private Drama publishedDrama(Long dramaId) { + Drama drama = dramaCache.getIfPresent(dramaId); + if (drama == null) { + drama = dramaService.getPublished(dramaId); + dramaCache.put(dramaId, drama); + } + return drama; + } + private List enrichDisplay(List records) { if (records.isEmpty()) { return List.of();