From 9f76302e53f204c80d062bdfc4379680e9a9b47a Mon Sep 17 00:00:00 2001 From: jianghaifeng Date: Fri, 16 Oct 2026 12:41:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(media):=20C1=20=E8=BD=AC=E7=A0=81=20worker?= =?UTF-8?q?=EF=BC=88=E7=8B=AC=E7=AB=8B=E8=BF=9B=E7=A8=8B=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=20transcode=5Ftask=EF=BC=8C=E4=BA=A7=E5=87=BA=E5=9B=9B?= =?UTF-8?q?=E6=A1=A3=20HLS=20+=20=E5=B0=81=E9=9D=A2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：codex-研发任务清单阶段 C 第一项要求消费 transcode_task 产出多码率 HLS + 截图； 勘察发现除 worker 外还缺两件——TranscodeService.enqueue 没有任何调用方（队列永远空）、 media_asset.hls_url/poster_url 没人回写，所以一并补齐，已有能力（条件更新认领 / 3 次重试）直接复用。 - 新增 TranscodeRunner 端口 + FfmpegTranscodeRunner（每档一条 ffmpeg 出 VOD HLS + 手写 master.m3u8 + 抽封面 + ffprobe 读时长） - 新增 TranscodeWorker：claimNext → 转码 → 登记 media_rendition → 媒资 ready；失败记原因、第 3 次才落 FAILED 并把媒资置 failed - 存储统一走 MediaStorageService（新增 fileFor / objectKeyOf，含目录穿越校验），worker 不自己拼磁盘路径 - 队列隔离：app-api/admin-api 默认 worker-enabled=false，新增独立模块 fruitshorts-transcode-worker（随机端口） - MediaService 上传完成对视频素材入队并置 transcoding（图片仍直接 ready） - 实测抓到并修掉两个真问题：① scale 未加 force_divisible_by=2 时 854x480 算出奇数宽 853，libx264 拒绝转码 ② 只写 -hls_time 不强制关键帧时 8 秒片源只出 1 个分片，补 -force_key_frames 后按 6s 正确切分 - README 开发环境补 ffmpeg 9.0.2（只有 worker 需要） - 实测：mvn test 83 条全过、0 skipped（新增 worker 12 条，含真实 ffmpeg smoke）； 端到端：正常源 SUCCESS + 4 档 rendition + 媒资 ready；坏源重试 3 次后 FAILED --- README.md | 6 + .../media/config/TranscodeProperties.java | 75 ++++++ .../media/service/MediaService.java | 60 +++++ .../media/service/MediaStorageService.java | 38 +++ .../media/service/TranscodeService.java | 13 +- .../transcode/FfmpegTranscodeRunner.java | 235 ++++++++++++++++++ .../service/transcode/TranscodeRunner.java | 50 ++++ .../service/transcode/TranscodeWorker.java | 140 +++++++++++ fruitshorts-transcode-worker/pom.xml | 40 +++ .../worker/TranscodeWorkerApplication.java | 26 ++ .../src/main/resources/application.yml | 51 ++++ .../FfmpegTranscodeRunnerSmokeTest.java | 117 +++++++++ .../worker/TranscodeClaimAndStorageTest.java | 121 +++++++++ .../worker/TranscodeWorkerTest.java | 165 ++++++++++++ pom.xml | 2 + 15 files changed, 1136 insertions(+), 3 deletions(-) create mode 100644 fruitshorts-module-media/src/main/java/com/fruitshorts/media/config/TranscodeProperties.java create mode 100644 fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java create mode 100644 fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java create mode 100644 fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java create mode 100644 fruitshorts-transcode-worker/pom.xml create mode 100644 fruitshorts-transcode-worker/src/main/java/com/fruitshorts/worker/TranscodeWorkerApplication.java create mode 100644 fruitshorts-transcode-worker/src/main/resources/application.yml create mode 100644 fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java create mode 100644 fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeClaimAndStorageTest.java create mode 100644 fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java diff --git a/README.md b/README.md index 82f6801..3125d22 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,7 @@ FruitServer/ | JDK | OpenJDK 21.0.12.1 | `brew install openjdk@21`，`JAVA_HOME` 已写入 `~/.zshrc` | | Maven | 3.10.0 | `brew install maven`，命令为 `/opt/homebrew/bin/mvn` | | PostgreSQL | 18.6 | `brew install postgresql@18`，`brew services start postgresql@18` 启动 | +| ffmpeg | 9.0.2（Homebrew 当前稳定版） | `brew install ffmpeg`；**只有转码 worker 需要**，两个在线入口（8080/8081）不依赖它 | > `postgresql@18` 是 keg-only，命令行工具在 `/opt/homebrew/opt/postgresql@18/bin`， > 用之前把它加到 PATH：`export PATH="/opt/homebrew/opt/postgresql@18/bin:$PATH"`。 @@ -51,8 +52,13 @@ FruitServer/ ```sh java -version # openjdk 21.0.12.1 mvn -v # Maven 3.10.0 / Java 21.0.12.1 +ffmpeg -version # 转码 worker 用；没装也能跑在线服务，只是转码任务会一直排队 ``` +> **ffmpeg 只被 `fruitshorts-transcode-worker` 使用**：`backend-design-phase1.md` §8.5 要求转码与在线服务隔离， +> 所以 app-api / admin-api 里 `fruitshorts.media.transcode.worker-enabled` 默认 false，转码要单独起 worker 进程 +> （`java -jar fruitshorts-transcode-worker/target/fruitshorts-transcode-worker-1.0.0-SNAPSHOT.jar`）。 + 说明： - 工程编译目标为 Java 17（`maven.compiler.release=17`），在 JDK 21 上构建与运行。 diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/config/TranscodeProperties.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/config/TranscodeProperties.java new file mode 100644 index 0000000..234117f --- /dev/null +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/config/TranscodeProperties.java @@ -0,0 +1,75 @@ +// Author: fruit.com.cn + +package com.fruitshorts.media.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +import java.util.List; + +/** + * 转码 worker 配置。 + * + *

**默认关闭**（`worker-enabled=false`）：`backend-design-phase1.md` §8.5 要求转码与在线服务隔离， + * 所以 app-api / admin-api 绝不跑转码；要跑就单独起 `fruitshorts-transcode-worker` 进程 + * （那个模块的 yml 里把开关置 true）。 + * + *

码率阶梯按 `backend-design-phase1.md` §3.1：360p/480p/720p/1080p。 + */ +@Data +@Component +@ConfigurationProperties(prefix = "fruitshorts.media.transcode") +public class TranscodeProperties { + + /** 是否在本进程跑转码（默认 false，只有独立 worker 进程开）。 */ + private boolean workerEnabled = false; + + /** 队列空时的轮询间隔（毫秒）。 */ + private long pollDelayMs = 5000L; + + /** ffmpeg 可执行文件（生产建议绝对路径或容器内固定路径）。 */ + private String ffmpegPath = "ffmpeg"; + + /** ffprobe 可执行文件，用于读时长。 */ + private String ffprobePath = "ffprobe"; + + /** HLS 分片时长（秒）。 */ + private int segmentSeconds = 6; + + /** 封面截图取视频第几秒。 */ + private int posterSecond = 1; + + /** 单条 ffmpeg 的超时（秒）：防止坏片源把 worker 卡死。 */ + private int timeoutSeconds = 1800; + + /** 码率阶梯（默认四档，改这里不用改代码）。 */ + private List ladder = List.of( + new Rung("360p", 640, 360, 400, 64), + new Rung("480p", 854, 480, 800, 64), + new Rung("720p", 1280, 720, 1600, 96), + new Rung("1080p", 1920, 1080, 3200, 128)); + + /** 一档码率。 */ + @Data + public static class Rung { + + /** 码率标识，写进 media_rendition.quality，也是产物目录名。 */ + private String quality; + private int width; + private int height; + private int videoBitrateKbps; + private int audioBitrateKbps; + + public Rung() { + } + + public Rung(String quality, int width, int height, int videoBitrateKbps, int audioBitrateKbps) { + this.quality = quality; + this.width = width; + this.height = height; + this.videoBitrateKbps = videoBitrateKbps; + this.audioBitrateKbps = audioBitrateKbps; + } + } +} diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java index 39758bf..376c56d 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaService.java @@ -29,6 +29,8 @@ public class MediaService { private static final String STATUS_READY = "ready"; private static final String STATUS_UPLOADED = "uploaded"; + private static final String STATUS_TRANSCODING = "transcoding"; + private static final String STATUS_FAILED = "failed"; private static final String TASK_UPLOADING = "uploading"; private static final String TASK_COMPLETED = "completed"; private static final Set IMAGE_EXTENSIONS = @@ -38,6 +40,7 @@ public class MediaService { private final MediaStorageService storageService; private final MediaAssetMapper mediaAssetMapper; private final MediaUploadTaskMapper mediaUploadTaskMapper; + private final TranscodeService transcodeService; public UploadInitRespVO init(UploadInitReqVO req, Long userId) { if (req.getFileSize() > properties.getMaxFileSize()) { @@ -121,6 +124,14 @@ public class MediaService { asset.setCreateBy(task.getCreateBy()); mediaAssetMapper.insert(asset); + // 视频入转码队列（§7.2 流水线：合并校验 → 写 transcode_task）；图片没有转码概念，直接 ready + if (!isImage(task.getFileName())) { + asset.setStatus(STATUS_TRANSCODING); + asset.setTranscodeStatus(TranscodeService.STATUS_QUEUED); + mediaAssetMapper.updateById(asset); + transcodeService.enqueue(asset.getId(), 0L, 0L, publicUrl); + } + task.setStatus(TASK_COMPLETED); task.setAssetId(asset.getId()); task.setReceivedChunks(chunks); @@ -131,6 +142,55 @@ public class MediaService { return toResp(mediaAssetMapper.selectById(asset.getId())); } + /** + * 转码开始：把媒资标成 transcoding。 + * + *

由转码 worker 调用（worker 与上传在不同进程），所以这里是独立的小事务方法。 + */ + @Transactional(rollbackFor = Exception.class) + public void markTranscoding(Long assetId) { + updateAssetStatus(assetId, STATUS_TRANSCODING, TranscodeService.STATUS_RUNNING); + } + + /** 转码成功：写 hls_url / poster_url 并把状态推到 ready。 */ + @Transactional(rollbackFor = Exception.class) + public void markReady(Long assetId, String hlsUrl, String posterUrl, Integer durationSec) { + MediaAsset asset = assetId == null ? null : mediaAssetMapper.selectById(assetId); + if (asset == null) { + return; + } + asset.setStatus(STATUS_READY); + asset.setTranscodeStatus(TranscodeService.STATUS_SUCCESS); + asset.setHlsUrl(hlsUrl == null ? "" : hlsUrl); + asset.setPosterUrl(posterUrl == null ? "" : posterUrl); + if (durationSec != null && durationSec > 0) { + asset.setDurationSec(durationSec); + } + mediaAssetMapper.updateById(asset); + } + + /** 转码最终失败（重试耗尽）：状态置 failed，原因留给运营在后台排查。 */ + @Transactional(rollbackFor = Exception.class) + public void markFailed(Long assetId, String reason) { + MediaAsset asset = assetId == null ? null : mediaAssetMapper.selectById(assetId); + if (asset == null) { + return; + } + asset.setStatus(STATUS_FAILED); + asset.setTranscodeStatus(TranscodeService.STATUS_FAILED); + mediaAssetMapper.updateById(asset); + } + + private void updateAssetStatus(Long assetId, String status, String transcodeStatus) { + MediaAsset asset = assetId == null ? null : mediaAssetMapper.selectById(assetId); + if (asset == null) { + return; + } + asset.setStatus(status); + asset.setTranscodeStatus(transcodeStatus); + mediaAssetMapper.updateById(asset); + } + public MediaAssetRespVO get(Long id) { MediaAsset asset = id == null ? null : mediaAssetMapper.selectById(id); if (asset == null) { diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaStorageService.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaStorageService.java index 4c57671..55ccb94 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaStorageService.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/MediaStorageService.java @@ -53,6 +53,44 @@ public class MediaStorageService { return root; } + /** + * 把 objectKey 解析成**本地可写路径**：转码产物落地用它，避免各处在外面自己拼 `./data/uploads`。 + * + *

切 OSS 时这里是唯一的替换点（改成"下载到临时目录 / 上传回对象存储"的适配）。 + * 同时做一次越权校验：objectKey 不允许跑出根目录。 + */ + public Path fileFor(String objectKey) { + if (objectKey == null || objectKey.isBlank()) { + throw new ServiceException(ErrorCode.BAD_REQUEST, "objectKey 不能为空"); + } + Path root = rootDir().toAbsolutePath().normalize(); + Path target = root.resolve(objectKey.replaceAll("^/+", "")).normalize(); + if (!target.startsWith(root)) { + throw new ServiceException(ErrorCode.BAD_REQUEST, "非法的 objectKey：" + objectKey); + } + return target; + } + + /** + * 把对外 URL 反解成 objectKey（`/files/xxx/yyy.mp4` → `xxx/yyy.mp4`，顺带去掉签名查询串）。 + * worker 拿到的 `transcode_task.source_url` 就是这种对外 URL。 + */ + public String objectKeyOf(String publicUrl) { + if (publicUrl == null || publicUrl.isBlank()) { + return ""; + } + String clean = publicUrl; + int query = clean.indexOf('?'); + if (query >= 0) { + clean = clean.substring(0, query); + } + String prefix = properties.getPublicPrefix(); + if (clean.startsWith(prefix)) { + clean = clean.substring(prefix.length()); + } + return clean.replaceAll("^/+", ""); + } + public Path chunkDir(String uploadId) { return rootDir().resolve("tmp").resolve(uploadId); } diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java index d81aabd..b2282dd 100644 --- a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/TranscodeService.java @@ -33,7 +33,8 @@ public class TranscodeService { public static final String STATUS_RUNNING = "RUNNING"; public static final String STATUS_SUCCESS = "SUCCESS"; public static final String STATUS_FAILED = "FAILED"; - private static final int MAX_RETRY = 3; + /** 失败重试上限（`backend-design-phase1.md` §7.2：失败重试 3 次并记录原因）。 */ + public static final int MAX_RETRY = 3; private final TranscodeTaskMapper transcodeTaskMapper; private final MediaRenditionMapper mediaRenditionMapper; @@ -95,8 +96,13 @@ public class TranscodeService { transcodeTaskMapper.updateById(task); } - /** 失败重试最多 3 次，超过则停在 FAILED，由运营在后台看到并处理。 */ - public void fail(Long taskId, String errorMessage) { + /** + * 失败重试最多 3 次，超过则停在 FAILED，由运营在后台看到并处理。 + * + * @return 这次失败之后任务的新状态（QUEUED = 还会重试；FAILED = 已放弃）， + * 调用方据此决定要不要把媒资也置为 failed + */ + public String fail(Long taskId, String errorMessage) { TranscodeTask task = getExisting(taskId); int retry = task.getRetryCount() == null ? 0 : task.getRetryCount(); if (retry + 1 >= MAX_RETRY) { @@ -108,6 +114,7 @@ public class TranscodeService { task.setErrorMsg(errorMessage == null ? "" : errorMessage); task.setFinishedAt(LocalDateTime.now()); transcodeTaskMapper.updateById(task); + return task.getStatus(); } public List listRenditions(Long episodeId) { diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java new file mode 100644 index 0000000..8f45d89 --- /dev/null +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/FfmpegTranscodeRunner.java @@ -0,0 +1,235 @@ +// Author: fruit.com.cn + +package com.fruitshorts.media.service.transcode; + +import com.fruitshorts.common.exception.ErrorCode; +import com.fruitshorts.common.exception.ServiceException; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +/** + * 用本机 ffmpeg 产出多码率 HLS + 封面。 + * + *

每档码率一条 ffmpeg 命令出 VOD 型 HLS 到 `<输出目录>//`， + * 再由本类写一个 `master.m3u8`（手写比 `var_stream_map` 可控，也不用依赖 ffmpeg 的 muxer 细节）。 + * 封面单独抽一帧 jpg。时长用 ffprobe 读，读不到就记 0（不因为探测失败就把整单判失败）。 + * + *

**注意**：本实现需要"本地可写路径"，也就是本地磁盘存储形态； + * 将来切 OSS 时，这一层要改成"下载到临时目录 → 转码 → 上传"，替换点就是 {@code MediaStorageService} 与这里。 + */ +@Slf4j +@Component +public class FfmpegTranscodeRunner implements TranscodeRunner { + + private static final String MASTER_PLAYLIST = "master.m3u8"; + private static final String VARIANT_PLAYLIST = "index.m3u8"; + private static final String POSTER_FILE = "poster.jpg"; + private static final int STDERR_TAIL = 800; + + @Override + public TranscodeResult run(TranscodeRequest request) { + if (!Files.isReadable(request.sourceFile())) { + throw new ServiceException(ErrorCode.DATA_NOT_FOUND, + "源文件不存在或不可读：" + request.sourceFile()); + } + createDirectories(request.outputDir()); + int durationSec = probeDuration(request); + + List renditions = new ArrayList<>(request.ladder().size()); + for (LadderRung rung : request.ladder()) { + renditions.add(transcodeRung(request, rung)); + } + writeMasterPlaylist(request, renditions); + ProducedRendition poster = extractPoster(request); + log.info("转码完成 task={}：{} 档 HLS + 封面，时长 {}s", + request.taskId(), renditions.size(), durationSec); + return new TranscodeResult(joinKey(request.objectKeyPrefix(), MASTER_PLAYLIST), renditions, + poster.objectKey(), durationSec); + } + + private ProducedRendition transcodeRung(TranscodeRequest request, LadderRung rung) { + String relativeDir = rung.quality() + "/"; + Path rungDir = request.outputDir().resolve(rung.quality()); + createDirectories(rungDir); + List command = List.of( + request.ffmpegPath(), "-y", + "-i", request.sourceFile().toString(), + // 等比缩放到目标档位（不拉伸）。force_divisible_by=2 不能省： + // 640x360 的源在 854x480 档会算出 853x480 这种奇数宽度，libx264 直接拒绝（width not divisible by 2） + "-vf", "scale=" + rung.width() + ":" + rung.height() + + ":force_original_aspect_ratio=decrease:force_divisible_by=2", + "-c:v", "libx264", "-preset", "veryfast", + "-b:v", rung.videoBitrateKbps() + "k", + // 强制在每个分片边界插关键帧：只写 -hls_time 的话，片源关键帧间隔大于分片时长时 + // 分片会被"并"成一个超长分片（实测 8 秒片源只出了 1 个分片），起播与拖动都会变差 + "-force_key_frames", "expr:gte(t,n_forced*" + request.segmentSeconds() + ")", + "-c:a", "aac", "-b:a", rung.audioBitrateKbps() + "k", + "-hls_time", String.valueOf(request.segmentSeconds()), + "-hls_playlist_type", "vod", + "-hls_segment_filename", rungDir.resolve("seg_%03d.ts").toString(), + rungDir.resolve(VARIANT_PLAYLIST).toString()); + runProcess(command, request.timeoutSeconds(), "转码 " + rung.quality() + " 失败"); + + Path playlist = rungDir.resolve(VARIANT_PLAYLIST); + if (!Files.exists(playlist)) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + "ffmpeg 未产出播放列表：" + playlist); + } + return new ProducedRendition(rung.quality(), rung.width(), rung.height(), + rung.videoBitrateKbps(), rung.audioBitrateKbps(), + joinKey(request.objectKeyPrefix(), relativeDir + VARIANT_PLAYLIST), + sizeOf(rungDir), countSegments(rungDir)); + } + + /** 手写 master.m3u8：按带宽升序，把每档的 index.m3u8 串起来。 */ + private void writeMasterPlaylist(TranscodeRequest request, List renditions) { + StringBuilder body = new StringBuilder("#EXTM3U\n#EXT-X-VERSION:3\n"); + List sorted = renditions.stream() + .sorted(Comparator.comparingInt(ProducedRendition::videoBitrateKbps)) + .toList(); + for (ProducedRendition rendition : sorted) { + int bandwidth = (rendition.videoBitrateKbps() + rendition.audioBitrateKbps()) * 1000; + body.append("#EXT-X-STREAM-INF:BANDWIDTH=").append(bandwidth) + .append(",RESOLUTION=").append(rendition.width()).append('x') + .append(rendition.height()).append('\n') + .append(rendition.quality()).append('/').append(VARIANT_PLAYLIST).append('\n'); + } + Path master = request.outputDir().resolve(MASTER_PLAYLIST); + try { + Files.writeString(master, body.toString(), StandardCharsets.UTF_8); + } catch (IOException ex) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, "写 master.m3u8 失败：" + ex.getMessage()); + } + } + + private ProducedRendition extractPoster(TranscodeRequest request) { + Path poster = request.outputDir().resolve(POSTER_FILE); + List command = List.of( + request.ffmpegPath(), "-y", + "-ss", String.valueOf(request.posterSecond()), + "-i", request.sourceFile().toString(), + "-frames:v", "1", "-q:v", "2", + poster.toString()); + runProcess(command, request.timeoutSeconds(), "抽取封面失败"); + long size = 0L; + try { + if (Files.exists(poster)) { + size = Files.size(poster); + } + } catch (IOException ignored) { + // 封面大小只用于登记，读不到不影响转码结果 + } + return new ProducedRendition("poster", 0, 0, 0, 0, + joinKey(request.objectKeyPrefix(), POSTER_FILE), size, 1); + } + + private int probeDuration(TranscodeRequest request) { + try { + String output = runProcess(List.of( + request.ffprobePath(), "-v", "error", + "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", + request.sourceFile().toString()), 60, "探测时长失败"); + return (int) Math.round(Double.parseDouble(output.trim())); + } catch (Exception ex) { + // 探测失败不阻断：时长只是登记信息，转码本身可能仍然成功 + log.warn("读取时长失败（task={}）：{}", request.taskId(), ex.getMessage()); + return 0; + } + } + + /** + * 起进程并等结果；非 0 退出把输出尾部带进异常，方便定位是片源问题还是参数问题。 + * + *

输出**重定向到临时文件**而不是直接读管道：直接 `readAllBytes()` 会阻塞到进程结束， + * 那样 `waitFor(timeout)` 的超时就形同虚设，坏片源能把 worker 永久卡住。 + */ + private String runProcess(List command, int timeoutSeconds, String errorPrefix) { + Path logFile = null; + try { + logFile = Files.createTempFile("fruitshorts-ffmpeg-", ".log"); + ProcessBuilder builder = new ProcessBuilder(command) + .redirectErrorStream(true) + .redirectOutput(logFile.toFile()); + Process process = builder.start(); + boolean finished = process.waitFor(timeoutSeconds, TimeUnit.SECONDS); + if (!finished) { + process.destroyForcibly(); + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + errorPrefix + "：超时（" + timeoutSeconds + "s）"); + } + String output = Files.readString(logFile, StandardCharsets.UTF_8); + if (process.exitValue() != 0) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + errorPrefix + "：" + tail(output)); + } + return output; + } catch (ServiceException ex) { + throw ex; + } catch (IOException ex) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, + errorPrefix + "：无法执行 ffmpeg/ffprobe（" + ex.getMessage() + "）"); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new ServiceException(ErrorCode.INTERNAL_ERROR, errorPrefix + "：被中断"); + } finally { + if (logFile != null) { + try { + Files.deleteIfExists(logFile); + } catch (IOException ignored) { + // 临时日志删不掉不是业务问题 + } + } + } + } + + private void createDirectories(Path path) { + try { + Files.createDirectories(path); + } catch (IOException ex) { + throw new ServiceException(ErrorCode.INTERNAL_ERROR, "创建产物目录失败：" + path); + } + } + + private long sizeOf(Path dir) { + try (Stream files = Files.walk(dir)) { + return files.filter(Files::isRegularFile).mapToLong(path -> { + try { + return Files.size(path); + } catch (IOException ex) { + return 0L; + } + }).sum(); + } catch (IOException ex) { + return 0L; + } + } + + private int countSegments(Path dir) { + try (Stream files = Files.list(dir)) { + return (int) files.filter(path -> path.getFileName().toString().endsWith(".ts")).count(); + } catch (IOException ex) { + return 0; + } + } + + private String tail(String output) { + String trimmed = output == null ? "" : output.trim(); + return trimmed.length() <= STDERR_TAIL ? trimmed : trimmed.substring(trimmed.length() - STDERR_TAIL); + } + + private String joinKey(String prefix, String relative) { + String cleanPrefix = prefix == null ? "" : prefix.replaceAll("/+$", ""); + return cleanPrefix.isEmpty() ? relative : cleanPrefix + "/" + relative; + } +} diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java new file mode 100644 index 0000000..cd9fdf2 --- /dev/null +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeRunner.java @@ -0,0 +1,50 @@ +// Author: fruit.com.cn + +package com.fruitshorts.media.service.transcode; + +import java.nio.file.Path; +import java.util.List; + +/** + * 转码执行端口：把"调 ffmpeg 出多码率 HLS + 封面"封在一层后面。 + * + *

这样做的好处是 worker 的逻辑（取任务 / 登记产物 / 推状态 / 重试）可以**不依赖 ffmpeg** 做单测， + * 生产实现见 {@link FfmpegTranscodeRunner}。 + */ +public interface TranscodeRunner { + + /** 执行一次转码。失败时抛异常（异常消息会被写进 `transcode_task.error_msg`）。 */ + TranscodeResult run(TranscodeRequest request); + + /** 一档码率。 */ + record LadderRung(String quality, int width, int height, int videoBitrateKbps, int audioBitrateKbps) { + } + + /** + * 一次转码请求。 + * + * @param sourceFile 源视频的**本地路径**（由 MediaStorageService 解析，worker 不自己拼路径） + * @param outputDir 产物落地的本地目录 + * @param objectKeyPrefix 产物 objectKey 前缀，返回值里的 objectKey 基于它拼 + */ + record TranscodeRequest(Long taskId, Path sourceFile, Path outputDir, String objectKeyPrefix, + List ladder, String ffmpegPath, String ffprobePath, + int segmentSeconds, int posterSecond, int timeoutSeconds) { + } + + /** 一档产物。 */ + record ProducedRendition(String quality, int width, int height, int videoBitrateKbps, + int audioBitrateKbps, String objectKey, long sizeBytes, int segmentCount) { + } + + /** + * 转码产物。 + * + * @param masterObjectKey HLS 主播放列表的 objectKey（客户端可播入口） + * @param posterObjectKey 封面截图 objectKey + * @param durationSec 视频时长（ffprobe 读出，0 表示读不到） + */ + record TranscodeResult(String masterObjectKey, List renditions, + String posterObjectKey, int durationSec) { + } +} diff --git a/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java new file mode 100644 index 0000000..b2772d8 --- /dev/null +++ b/fruitshorts-module-media/src/main/java/com/fruitshorts/media/service/transcode/TranscodeWorker.java @@ -0,0 +1,140 @@ +// Author: fruit.com.cn + +package com.fruitshorts.media.service.transcode; + +import com.fruitshorts.media.config.TranscodeProperties; +import com.fruitshorts.media.dal.entity.TranscodeTask; +import com.fruitshorts.media.service.MediaService; +import com.fruitshorts.media.service.MediaStorageService; +import com.fruitshorts.media.service.TranscodeService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/** + * 转码 worker：消费 `transcode_task`，产出多码率 HLS + 封面。 + * + *

跑法：**默认不在 app-api / admin-api 里跑**（`fruitshorts.media.transcode.worker-enabled=false`）， + * 要转码就单独起 `fruitshorts-transcode-worker`（`backend-design-phase1.md` §8.5 要求队列与在线服务隔离）。 + * + *

并发安全靠 `TranscodeService.claimNext()` 的条件更新（`status=QUEUED` 才会被改成 `RUNNING`）， + * 多进程同时抢也只有一个能拿到；拿不到就下一轮再来。 + * + *

失败重试 3 次（`TranscodeService.fail`），每次都把原因写进 `error_msg`； + * 到第三次才落到 `FAILED`，同时把 `media_asset.status` 置 `failed`。 + */ +@Slf4j +@Component +@RequiredArgsConstructor +@ConditionalOnProperty(name = "fruitshorts.media.transcode.worker-enabled", havingValue = "true") +public class TranscodeWorker { + + private final TranscodeService transcodeService; + private final MediaService mediaService; + private final MediaStorageService storageService; + private final TranscodeRunner transcodeRunner; + private final TranscodeProperties properties; + + /** 定时轮询：队列空就按 poll-delay 再来一轮；上一轮没跑完不会被下一轮叠加（fixedDelay）。 */ + @Scheduled(fixedDelayString = "${fruitshorts.media.transcode.poll-delay-ms:5000}") + public void poll() { + try { + processOne(); + } catch (Exception ex) { + // 轮询里不能让异常逃出去，否则调度线程会被取消，整个 worker 静默停摆 + log.error("转码轮询异常：{}", ex.getMessage(), ex); + } + } + + /** + * 处理一条任务。 + * + * @return true 表示确实领到并处理了一条；false 表示队列空（或没抢到） + */ + public boolean processOne() { + TranscodeTask task = transcodeService.claimNext(); + if (task == null) { + return false; + } + process(task); + return true; + } + + /** 处理指定任务：转码成功 → 登记产物 + 资产就绪；失败 → 记原因并决定是否重试。 */ + public void process(TranscodeTask task) { + long startedAt = System.currentTimeMillis(); + String objectKeyPrefix = outputPrefix(task); + try { + Path sourceFile = storageService.fileFor(storageService.objectKeyOf(task.getSourceUrl())); + Path outputDir = storageService.fileFor(objectKeyPrefix); + TranscodeRunner.TranscodeResult result = transcodeRunner.run(new TranscodeRunner.TranscodeRequest( + task.getId(), sourceFile, outputDir, objectKeyPrefix, ladder(), properties.getFfmpegPath(), + properties.getFfprobePath(), properties.getSegmentSeconds(), properties.getPosterSecond(), + properties.getTimeoutSeconds())); + + transcodeService.succeed(task.getId(), toRenditions(result)); + if (task.getMediaAssetId() != null && task.getMediaAssetId() > 0) { + mediaService.markReady(task.getMediaAssetId(), + storageService.toPublicUrl(result.masterObjectKey()), + storageService.toPublicUrl(result.posterObjectKey()), + result.durationSec()); + } + log.info("转码成功 task={} asset={} 档数={} 耗时={}ms 产物前缀={}", + task.getId(), task.getMediaAssetId(), result.renditions().size(), + System.currentTimeMillis() - startedAt, objectKeyPrefix); + } catch (Exception ex) { + String reason = ex.getMessage() == null ? ex.getClass().getSimpleName() : ex.getMessage(); + int attempt = (task.getRetryCount() == null ? 0 : task.getRetryCount()) + 1; + String status = transcodeService.fail(task.getId(), reason); + if (TranscodeService.STATUS_FAILED.equals(status) + && task.getMediaAssetId() != null && task.getMediaAssetId() > 0) { + mediaService.markFailed(task.getMediaAssetId(), reason); + } + log.error("转码失败 task={} 第 {}/{} 次尝试 耗时={}ms 结果={} 原因={}", + task.getId(), attempt, TranscodeService.MAX_RETRY, System.currentTimeMillis() - startedAt, + status, reason); + } + } + + /** 产物 objectKey 前缀：`episode/transcoded/{taskId}`，与上传目录分开放，便于按前缀清理。 */ + private String outputPrefix(TranscodeTask task) { + return "transcoded/" + task.getId(); + } + + private List ladder() { + List ladder = new ArrayList<>(); + for (TranscodeProperties.Rung rung : properties.getLadder()) { + ladder.add(new TranscodeRunner.LadderRung(rung.getQuality(), rung.getWidth(), rung.getHeight(), + rung.getVideoBitrateKbps(), rung.getAudioBitrateKbps())); + } + return ladder; + } + + private List toRenditions( + TranscodeRunner.TranscodeResult result) { + List renditions = new ArrayList<>(); + for (TranscodeRunner.ProducedRendition produced : result.renditions()) { + com.fruitshorts.media.dal.entity.MediaRendition rendition = + new com.fruitshorts.media.dal.entity.MediaRendition(); + rendition.setQuality(produced.quality()); + rendition.setFormat("HLS"); + rendition.setObjectKey(produced.objectKey()); + rendition.setPlayUrl(storageService.toPublicUrl(produced.objectKey())); + rendition.setWidth(produced.width()); + rendition.setHeight(produced.height()); + rendition.setVideoBitrate(produced.videoBitrateKbps()); + rendition.setAudioBitrate(produced.audioBitrateKbps()); + rendition.setSizeBytes(produced.sizeBytes()); + rendition.setSegmentCount(produced.segmentCount()); + rendition.setDurationSec(result.durationSec()); + renditions.add(rendition); + } + return renditions; + } +} diff --git a/fruitshorts-transcode-worker/pom.xml b/fruitshorts-transcode-worker/pom.xml new file mode 100644 index 0000000..9526e2a --- /dev/null +++ b/fruitshorts-transcode-worker/pom.xml @@ -0,0 +1,40 @@ + + + + + 4.0.0 + + + com.fruitshorts + fruitshorts + 1.0.0-SNAPSHOT + + + fruitshorts-transcode-worker + fruitshorts-transcode-worker + 转码 worker：独立进程消费 transcode_task（与在线服务隔离） + + + + com.fruitshorts + fruitshorts-module-media + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/fruitshorts-transcode-worker/src/main/java/com/fruitshorts/worker/TranscodeWorkerApplication.java b/fruitshorts-transcode-worker/src/main/java/com/fruitshorts/worker/TranscodeWorkerApplication.java new file mode 100644 index 0000000..f608af1 --- /dev/null +++ b/fruitshorts-transcode-worker/src/main/java/com/fruitshorts/worker/TranscodeWorkerApplication.java @@ -0,0 +1,26 @@ +// Author: fruit.com.cn + +package com.fruitshorts.worker; + +import org.mybatis.spring.annotation.MapperScan; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +/** + * 转码 worker 独立进程。 + * + *

为什么单独一个进程：`backend-design-phase1.md` §8.5 要求"转码队列与在线服务隔离，避免抢占资源"。 + * app-api / admin-api 里的 `fruitshorts.media.transcode.worker-enabled` 保持 false， + * 只有本进程把它打开，所以**跑不跑转码不影响 /app/** 的联调与播放**。 + * + *

启动：`java -jar fruitshorts-transcode-worker/target/fruitshorts-transcode-worker-1.0.0-SNAPSHOT.jar` + * （需要 ffmpeg / ffprobe 在 PATH 里，或用 `fruitshorts.media.transcode.ffmpeg-path` 指定绝对路径）。 + */ +@SpringBootApplication(scanBasePackages = "com.fruitshorts") +@MapperScan("com.fruitshorts.**.dal.mapper") +public class TranscodeWorkerApplication { + + public static void main(String[] args) { + SpringApplication.run(TranscodeWorkerApplication.class, args); + } +} diff --git a/fruitshorts-transcode-worker/src/main/resources/application.yml b/fruitshorts-transcode-worker/src/main/resources/application.yml new file mode 100644 index 0000000..d66b8dd --- /dev/null +++ b/fruitshorts-transcode-worker/src/main/resources/application.yml @@ -0,0 +1,51 @@ +# Author: fruit.com.cn +# +# 转码 worker 独立进程：只做队列消费，不对外提供接口。 +# 端口用 0（随机）——本进程不需要被访问，也不会跟 8080/8081 抢端口。 +server: + port: 0 + +spring: + application: + name: fruitshorts-transcode-worker + autoconfigure: + exclude: org.springframework.boot.autoconfigure.security.servlet.UserDetailsServiceAutoConfiguration + datasource: + driver-class-name: org.postgresql.Driver + url: jdbc:postgresql://127.0.0.1:5432/fruitshorts + username: ${DB_USER:fruitshorts} + password: ${DB_PASSWORD:fruitshorts} + +mybatis-plus: + configuration: + map-underscore-to-camel-case: true + global-config: + banner: false + db-config: + # 与两个入口保持一致：主键由应用侧雪花生成（S0），否则 worker 插入 media_rendition 会没有 id + id-type: assign_id + logic-delete-field: deleted + logic-delete-value: 1 + logic-not-delete-value: 0 + sequence: + # 单机固定值；多实例部署前改成按实例分配 + worker-id: 2 + datacenter-id: 1 + +fruitshorts: + media: + # 必须与 app-api/admin-api 指向同一个存储根，worker 才找得到上传后的源文件 + root-dir: ./data/uploads + public-prefix: /files + transcode: + # 只有本进程打开转码（§8.5 隔离要求） + worker-enabled: true + poll-delay-ms: 3000 + ffmpeg-path: ${FFMPEG_PATH:ffmpeg} + ffprobe-path: ${FFPROBE_PATH:ffprobe} + segment-seconds: 6 + poster-second: 1 + +logging: + level: + com.fruitshorts: info diff --git a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java new file mode 100644 index 0000000..1534f38 --- /dev/null +++ b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/FfmpegTranscodeRunnerSmokeTest.java @@ -0,0 +1,117 @@ +// Author: fruit.com.cn + +package com.fruitshorts.worker; + +import com.fruitshorts.media.service.transcode.FfmpegTranscodeRunner; +import com.fruitshorts.media.service.transcode.TranscodeRunner; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +/** + * 真实 ffmpeg smoke：**自己生成 2 秒测试片**（不依赖外部素材），跑一遍真转码，验证 HLS + 封面确实产出。 + * + *

本机没有 ffmpeg 时用 {@code assumeTrue} 跳过 —— 在 surefire 里会显示成 **Skipped**（不是通过）， + * 回执里按实际情况说明，不把"跳过"当"验过"。 + */ +class FfmpegTranscodeRunnerSmokeTest { + + @Test + void should_produce_hls_and_poster_when_a_tiny_sample_is_transcoded() throws Exception { + String ffmpeg = binaryPath("ffmpeg"); + String ffprobe = binaryPath("ffprobe"); + assumeTrue(ffmpeg != null && ffprobe != null, + "本机没有 ffmpeg/ffprobe，跳过真实转码 smoke"); + + Path work = Files.createTempDirectory("fruitshorts-transcode-smoke"); + try { + Path sample = work.resolve("sample.mp4"); + generateSample(ffmpeg, sample); + assertThat(Files.size(sample)).isGreaterThan(0L); + + Path output = work.resolve("out"); + TranscodeRunner runner = new FfmpegTranscodeRunner(); + // 只跑两档（360p/480p）保持秒级；四档阶梯的逻辑在 TranscodeWorkerTest 里断言 + TranscodeRunner.TranscodeResult result = runner.run(new TranscodeRunner.TranscodeRequest( + 1L, sample, output, "transcoded/smoke", + List.of(new TranscodeRunner.LadderRung("360p", 640, 360, 400, 64), + new TranscodeRunner.LadderRung("480p", 854, 480, 800, 64)), + ffmpeg, ffprobe, 2, 1, 120)); + + assertThat(Files.exists(output.resolve("master.m3u8"))).isTrue(); + assertThat(Files.exists(output.resolve("poster.jpg"))).isTrue(); + assertThat(Files.size(output.resolve("poster.jpg"))).isGreaterThan(0L); + assertThat(result.renditions()).hasSize(2); + for (TranscodeRunner.ProducedRendition rendition : result.renditions()) { + Path variant = output.resolve(rendition.quality()).resolve("index.m3u8"); + assertThat(Files.exists(variant)).as("变体播放列表 %s", rendition.quality()).isTrue(); + assertThat(countSegments(output.resolve(rendition.quality()))) + .as("%s 分片数", rendition.quality()).isGreaterThan(0); + assertThat(rendition.sizeBytes()).isGreaterThan(0L); + } + assertThat(result.durationSec()).isBetween(1, 5); + assertThat(readLines(output.resolve("master.m3u8"))) + .anyMatch(line -> line.contains("360p/index.m3u8")) + .anyMatch(line -> line.contains("480p/index.m3u8")); + } finally { + deleteRecursively(work); + } + } + + private void generateSample(String ffmpeg, Path sample) throws Exception { + Process process = new ProcessBuilder(ffmpeg, "-y", + "-f", "lavfi", "-i", "testsrc=size=640x360:rate=15", + "-f", "lavfi", "-i", "sine=frequency=440:duration=2", + "-shortest", "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", + sample.toString()) + .redirectErrorStream(true) + .start(); + assertThat(process.waitFor(120, TimeUnit.SECONDS)).as("生成测试片超时").isTrue(); + assertThat(process.exitValue()).as("生成测试片失败").isZero(); + } + + /** 在 PATH 里找可执行文件；找不到返回 null（由调用方 assume）。 */ + private String binaryPath(String name) { + try { + Process process = new ProcessBuilder("which", name).redirectErrorStream(true).start(); + String path = new String(process.getInputStream().readAllBytes()).trim(); + process.waitFor(10, TimeUnit.SECONDS); + return process.exitValue() == 0 && !path.isEmpty() ? path : null; + } catch (Exception ex) { + return null; + } + } + + private int countSegments(Path dir) throws IOException { + try (Stream files = Files.list(dir)) { + return (int) files.filter(path -> path.getFileName().toString().endsWith(".ts")).count(); + } + } + + private List readLines(Path path) throws IOException { + return Files.readAllLines(path); + } + + private void deleteRecursively(Path path) { + try (Stream files = Files.walk(path)) { + files.sorted(Comparator.reverseOrder()).forEach(item -> { + try { + Files.deleteIfExists(item); + } catch (IOException ignored) { + // 临时目录清理失败不影响结论 + } + }); + } catch (IOException ignored) { + // 同上 + } + } +} diff --git a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeClaimAndStorageTest.java b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeClaimAndStorageTest.java new file mode 100644 index 0000000..90145d6 --- /dev/null +++ b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeClaimAndStorageTest.java @@ -0,0 +1,121 @@ +// Author: fruit.com.cn + +package com.fruitshorts.worker; + +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fruitshorts.common.exception.ServiceException; +import com.fruitshorts.media.config.MediaProperties; +import com.fruitshorts.media.dal.entity.TranscodeTask; +import com.fruitshorts.media.dal.mapper.MediaRenditionMapper; +import com.fruitshorts.media.dal.mapper.TranscodeTaskMapper; +import com.fruitshorts.media.service.MediaStorageService; +import com.fruitshorts.media.service.TranscodeService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.nio.file.Path; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * 两个"防重复 / 防越权"的底层保障： + * ① 认领任务靠状态条件更新，抢不到就是 null（多进程同时跑也只有一个能转）； + * ② objectKey → 本地路径要挡住目录穿越。 + */ +class TranscodeClaimAndStorageTest { + + /** + * MyBatis-Plus 的 lambda 条件构造器需要实体的 TableInfo 缓存， + * 纯单测没有启动 MP 上下文，所以手动初始化一次（否则报 "can not find lambda cache"）。 + */ + @BeforeAll + static void initTableInfoCache() { + TableInfoHelper.initTableInfo( + new MapperBuilderAssistant(new MybatisConfiguration(), ""), TranscodeTask.class); + } + + private TranscodeTaskMapper transcodeTaskMapper; + private MediaRenditionMapper mediaRenditionMapper; + private TranscodeService transcodeService; + private MediaProperties mediaProperties; + private MediaStorageService storageService; + + @BeforeEach + void setUp() { + transcodeTaskMapper = mock(TranscodeTaskMapper.class); + mediaRenditionMapper = mock(MediaRenditionMapper.class); + transcodeService = new TranscodeService(transcodeTaskMapper, mediaRenditionMapper); + mediaProperties = new MediaProperties(); + mediaProperties.setRootDir("/tmp/fruitshorts-storage-test"); + mediaProperties.setPublicPrefix("/files"); + storageService = new MediaStorageService(mediaProperties); + } + + @Test + void should_return_null_when_another_worker_already_claimed_the_task() { + given(transcodeTaskMapper.selectOne(any())).willReturn(task(1L)); + // 条件更新返回 0：说明状态已经被别的 worker 改成 RUNNING 了 + given(transcodeTaskMapper.update(any(), any())).willReturn(0); + + assertThat(transcodeService.claimNext()).isNull(); + } + + @Test + void should_return_task_when_the_conditional_claim_succeeds() { + given(transcodeTaskMapper.selectOne(any())).willReturn(task(2L)); + given(transcodeTaskMapper.update(any(), any())).willReturn(1); + + TranscodeTask claimed = transcodeService.claimNext(); + + assertThat(claimed).isNotNull(); + assertThat(claimed.getId()).isEqualTo(2L); + } + + @Test + void should_return_null_when_queue_is_empty() { + given(transcodeTaskMapper.selectOne(any())).willReturn(null); + + assertThat(transcodeService.claimNext()).isNull(); + verify(transcodeTaskMapper, never()).update(any(), any()); + } + + @Test + void should_reject_object_key_that_escapes_the_storage_root() { + assertThatThrownBy(() -> storageService.fileFor("../../etc/passwd")) + .isInstanceOf(ServiceException.class) + .hasMessageContaining("非法的 objectKey"); + } + + @Test + void should_resolve_file_inside_root_when_object_key_is_normal() { + Path path = storageService.fileFor("episode/202610/1.mp4"); + + assertThat(path.startsWith(storageService.rootDir())).isTrue(); + assertThat(path.toString()).endsWith("episode/202610/1.mp4"); + } + + @Test + void should_strip_prefix_and_query_when_converting_public_url_to_object_key() { + assertThat(storageService.objectKeyOf("/files/episode/202610/1.mp4?e=1&s=abc")) + .isEqualTo("episode/202610/1.mp4"); + assertThat(storageService.objectKeyOf("/files/transcoded/88/master.m3u8")) + .isEqualTo("transcoded/88/master.m3u8"); + } + + private TranscodeTask task(Long id) { + TranscodeTask task = new TranscodeTask(); + task.setId(id); + task.setStatus(TranscodeService.STATUS_QUEUED); + task.setRetryCount(0); + return task; + } +} diff --git a/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java new file mode 100644 index 0000000..89eb541 --- /dev/null +++ b/fruitshorts-transcode-worker/src/test/java/com/fruitshorts/worker/TranscodeWorkerTest.java @@ -0,0 +1,165 @@ +// Author: fruit.com.cn + +package com.fruitshorts.worker; + +import com.fruitshorts.media.config.TranscodeProperties; +import com.fruitshorts.media.dal.entity.MediaRendition; +import com.fruitshorts.media.dal.entity.TranscodeTask; +import com.fruitshorts.media.service.MediaService; +import com.fruitshorts.media.service.MediaStorageService; +import com.fruitshorts.media.service.TranscodeService; +import com.fruitshorts.media.service.transcode.TranscodeRunner; +import com.fruitshorts.media.service.transcode.TranscodeWorker; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.nio.file.Paths; +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.contains; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * 转码 worker 的编排逻辑：取任务 → 转码 → 登记产物 → 推状态；失败按次数重试；抢不到任务就什么都不做。 + * + *

用替身 `TranscodeRunner`，所以**不依赖 ffmpeg**；真实 ffmpeg 的验证在 ffmpeg smoke 测试里。 + */ +class TranscodeWorkerTest { + + private static final Long TASK_ID = 88L; + private static final Long ASSET_ID = 99L; + + private TranscodeService transcodeService; + private MediaService mediaService; + private MediaStorageService storageService; + private StubRunner runner; + private TranscodeWorker worker; + + @BeforeEach + void setUp() { + transcodeService = mock(TranscodeService.class); + mediaService = mock(MediaService.class); + storageService = mock(MediaStorageService.class); + runner = new StubRunner(); + worker = new TranscodeWorker(transcodeService, mediaService, storageService, runner, + new TranscodeProperties()); + given(storageService.objectKeyOf(anyString())).willReturn("episode/202610/1.mp4"); + given(storageService.fileFor(anyString())).willReturn(Paths.get("/tmp/fruitshorts-test/1.mp4")); + given(storageService.toPublicUrl(anyString())) + .willAnswer(invocation -> "/files/" + invocation.getArgument(0)); + } + + @Test + void should_register_renditions_and_mark_asset_ready_when_transcode_succeeds() { + given(transcodeService.claimNext()).willReturn(queuedTask()); + + assertThat(worker.processOne()).isTrue(); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(List.class); + verify(transcodeService).succeed(eq(TASK_ID), captor.capture()); + assertThat(captor.getValue()).hasSize(2); + // 对外地址一律经 MediaStorageService，worker 不自己拼磁盘路径 + assertThat(captor.getValue().get(0).getPlayUrl()).startsWith("/files/transcoded/88/"); + assertThat(captor.getValue().get(0).getFormat()).isEqualTo("HLS"); + verify(mediaService).markReady(eq(ASSET_ID), eq("/files/transcoded/88/master.m3u8"), + eq("/files/transcoded/88/poster.jpg"), eq(120)); + verify(transcodeService, never()).fail(any(), anyString()); + } + + @Test + void should_cover_all_ladder_rungs_when_requesting_transcode() { + given(transcodeService.claimNext()).willReturn(queuedTask()); + + worker.processOne(); + + // 默认阶梯就是 backend-design-phase1.md §3.1 的四档 + assertThat(runner.request.ladder()).extracting(TranscodeRunner.LadderRung::quality) + .containsExactly("360p", "480p", "720p", "1080p"); + assertThat(runner.request.objectKeyPrefix()).isEqualTo("transcoded/88"); + } + + @Test + void should_keep_task_queued_and_record_reason_when_transcode_fails_before_max_retry() { + TranscodeTask task = queuedTask(); + task.setRetryCount(1); + given(transcodeService.claimNext()).willReturn(task); + given(transcodeService.fail(eq(TASK_ID), anyString())).willReturn(TranscodeService.STATUS_QUEUED); + runner.failure = new IllegalStateException("ffmpeg 退出码 1：Invalid data found"); + + worker.processOne(); + + verify(transcodeService).fail(eq(TASK_ID), contains("Invalid data found")); + // 还会重试，所以媒资不能置 failed（仍期待下一次成功） + verify(mediaService, never()).markFailed(any(), anyString()); + verify(mediaService, never()).markReady(any(), anyString(), anyString(), any()); + } + + @Test + void should_mark_asset_failed_when_retries_are_exhausted() { + TranscodeTask task = queuedTask(); + task.setRetryCount(2); + given(transcodeService.claimNext()).willReturn(task); + given(transcodeService.fail(eq(TASK_ID), anyString())).willReturn(TranscodeService.STATUS_FAILED); + runner.failure = new IllegalStateException("ffmpeg 未产出播放列表"); + + worker.processOne(); + + verify(transcodeService).fail(eq(TASK_ID), contains("未产出播放列表")); + verify(mediaService).markFailed(eq(ASSET_ID), contains("未产出播放列表")); + } + + @Test + void should_do_nothing_when_no_task_can_be_claimed() { + // 另一个 worker 已经抢走（或队列为空）：claimNext 返回 null + given(transcodeService.claimNext()).willReturn(null); + + assertThat(worker.processOne()).isFalse(); + assertThat(runner.called).isFalse(); + verify(transcodeService, never()).succeed(any(), any()); + verify(transcodeService, never()).fail(any(), anyString()); + } + + private TranscodeTask queuedTask() { + TranscodeTask task = new TranscodeTask(); + task.setId(TASK_ID); + task.setMediaAssetId(ASSET_ID); + task.setDramaId(5L); + task.setEpisodeId(12L); + task.setSourceUrl("/files/episode/202610/1.mp4"); + task.setStatus(TranscodeService.STATUS_RUNNING); + task.setRetryCount(0); + return task; + } + + /** 替身转码器：记录请求，按需抛异常。 */ + private static final class StubRunner implements TranscodeRunner { + private TranscodeRequest request; + private RuntimeException failure; + private boolean called; + + @Override + public TranscodeResult run(TranscodeRequest request) { + this.request = request; + this.called = true; + if (failure != null) { + throw failure; + } + return new TranscodeResult( + request.objectKeyPrefix() + "/master.m3u8", + List.of(new ProducedRendition("360p", 640, 360, 400, 64, + request.objectKeyPrefix() + "/360p/index.m3u8", 1024L, 4), + new ProducedRendition("480p", 854, 480, 800, 64, + request.objectKeyPrefix() + "/480p/index.m3u8", 2048L, 4)), + request.objectKeyPrefix() + "/poster.jpg", 120); + } + } +} diff --git a/pom.xml b/pom.xml index 5139a1b..192eefb 100644 --- a/pom.xml +++ b/pom.xml @@ -33,6 +33,8 @@ fruitshorts-module-creator fruitshorts-admin-api fruitshorts-app-api + + fruitshorts-transcode-worker