feat(media): C1 转码 worker(独立进程消费 transcode_task,产出四档 HLS + 封面)

为什么: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
This commit is contained in:
jianghaifeng committed 2026-10-16 12:41:48 +08:00
1 parent 1dc97b4bf6
commit 9f76302e53
15 files changed
+1136 -3

No files matched your search

@@ -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 独立进程。
*
* <p>为什么单独一个进程:`backend-design-phase1.md` §8.5 要求"转码队列与在线服务隔离,避免抢占资源"。
* app-api / admin-api 里的 `fruitshorts.media.transcode.worker-enabled` 保持 false,
* 只有本进程把它打开,所以**跑不跑转码不影响 /app/** 的联调与播放**。
*
* <p>启动:`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);
}
}
@@ -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
@@ -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 + 封面确实产出。
*
* <p>本机没有 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<Path> files = Files.list(dir)) {
return (int) files.filter(path -> path.getFileName().toString().endsWith(".ts")).count();
}
}
private List<String> readLines(Path path) throws IOException {
return Files.readAllLines(path);
}
private void deleteRecursively(Path path) {
try (Stream<Path> files = Files.walk(path)) {
files.sorted(Comparator.reverseOrder()).forEach(item -> {
try {
Files.deleteIfExists(item);
} catch (IOException ignored) {
// 临时目录清理失败不影响结论
}
});
} catch (IOException ignored) {
// 同上
}
}
}
@@ -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;
}
}
@@ -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 的编排逻辑:取任务 → 转码 → 登记产物 → 推状态;失败按次数重试;抢不到任务就什么都不做。
*
* <p>用替身 `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<List<MediaRendition>> 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);
}
}
}