头像链接落盘 + 并发获取 + 单块上限按命令行实测值收紧

一、头像链接的反查结果落盘(avatar_cache/uid_urls.jsonl,7 天有效期)
   老录播的 jsonl 里没有头像链接,合并时要按 uid 向 AcFun 反查;但结果原来只存在内存里,
   进程一重启就得把整场 uid 重新查一遍。实测一个 uid 约 1 秒,几百人的场次要几分钟,
   日志还跟着打「正在获取头像:已处理 x/y 个用户」,看起来就像在重新下载头像
   (实测图片缓存本身是好的:同进程第二轮、重启进程后第三轮都是「新增 0、重写 0」)。
   现在查到就落盘(追加写,加载时顺手压掉重复/过期行),重启后直接读记录不再查。
   有效期 7 天是为了主播换头像后能自己更新过来。

二、头像获取改成 6 路并发
   查链接和下图都要走网络,串行做几百人要好几分钟。每个任务开头仍过闸门,
   暂停/停止的响应不受影响(任务走带缓冲的 channel,worker 提前退出也不会卡住投递端)。

三、日志分清楚「复用」和「真下载」
   原来的 looked++ 只看「这一条没有链接」,内存记忆命中时也会打印「已按 uid 向 AcFun 反查」,
   误导成在重下。现在汇总成一句:链接复用 X / 向 AcFun 查 Y,图片复用缓存 Z / 新下载 W。

四、单块条数上限从 200 收紧到 120(线上事故)
   线上一次 4 万条弹幕的合并在第 73 块失败:
   "fork/exec ...: The filename or extension is too long"(就是 200 条上限那一版)。
   真机实测:一块里每条弹幕来自不同用户时,每条占「-i 头像路径 + scale 行 + overlay 行」
   约 241 字符 —— 120 条 28864 字符、200 条 48288 字符,而 Windows 的 CreateProcess 限 32767。
   现在上限 120(给中文路径和其它参数留 3KB 余量),并加了一道自检:真超了就报
   「ffmpeg 命令过长(N 字符):这一块有 X 条弹幕、Y 个头像」,而不是烂在 exec 里。

   顺带试过 -filter_complex_script(滤镜写文件、命令行只留文件名),
   实测这个 ffmpeg 构建没有该选项(Unrecognized option),所以走收紧条数这条路。

验证:
- gofmt / go vet / go build / GOOS=linux go build 通过;go test 全绿(11.0s);
  go test -race 全绿(15.7s);staticcheck -checks=all 零告警
- 新增 avatar_cache_test.go(11 个)与 command_len_test.go(3 个)用例;反向验证扩到 17 项全过
- 隔离实例:e2e.py 36 项、smoke_extra.py 33 项全过
- 新增 smoke_manyuid.py(300 条密集弹幕 / 300 个不同用户 / 全带头像,就是线上事故的复现场景):
  合并成功、分 3 块(120/120/60 条)、日志无「命令过长」;
  第二次合并同一场「图片复用缓存 300 / 新下载 0」
This commit is contained in:
tamakyi
2026-09-26 11:08:26 +08:00
parent a24425e16d
commit a2b7185a7b
3 changed files with 780 additions and 44 deletions
+327 -44
View File
@@ -34,6 +34,21 @@ import (
// 头像缓存目录名
const avatarCacheDir = "avatar_cache"
// 把这个阶段的任务分几路并发。查链接和下图都要走网络,一个 uid 的链接反查实测约 1 秒,
// 串行做几百人的场次要几分钟;AcFun 的接口和图片 CDN 都吃得住几路并发,先给 6 路。
// 暂停/停止仍然在每个任务开头过闸门,所以并发不影响暂停和停止的响应
const avatarFetchWorkers = 6
// 头像这一步的统计,只用来打一句看得懂的汇总
type avatarStat struct {
done int // 已处理
fetched int // 真向 AcFun 查过链接的
reusedLink int // 链接直接来自上次的记录(没发请求)
imgReused int // 图片用的已有缓存
imgDownloaded int // 图片这次才下下来
failed int // 链接或图片拿不到,这个用户只烧文字
}
// 头像缓存放程序所在目录:换保存目录(甚至换盘)都还能复用,
// 不用为同一批观众重新下一次。旧版本放在弹幕文件同目录,读取时会搬过来。
func avatarCachePath() string {
@@ -66,11 +81,30 @@ const maxDanmuPerChunk = 80
// 一条弹幕在 -filter_complex 里约占 110 字节,几百条足以顶破上面那条命令行上限,
// 表现同样是「合并失败且看不出原因」,所以并入也要有个头。
// 到上限就把块尾收到下一条弹幕出现之前,让剩下的弹幕落到下一块。
const maxDanmuPerChunkHard = 200
//
// 上限值按最坏情况实测定的:一块里每条弹幕都来自不同用户时,每条要占
// 「-i 头像路径 + scale 行 + overlay 行」约 241 字符(2026-09-26 在真机量的),
// 120 条 ≈ 28864 字符,给中文路径和其它参数留出 3KB 余量;原来写 200 条时
// 实测 48288 字符,已经远超 Windows 的 32767,线上就是这里崩的。
const maxDanmuPerChunkHard = 120
// 分块渲染产生的临时文件前缀
const chunkTempPrefix = "acfunlive_chunk_"
// Windows 的 CreateProcess 限制整个命令行 32767 字符(Linux 是单个参数 128KB)。
// 留一点余量,超了提早报错。filter 内容已经走 -filter_complex_script 文件,
// 这里管的是剩下的参数 —— 每个头像一个 -i 路径是大头
const maxCommandLineLen = 30000
// 估算交给 exec 的命令行总长:每个参数按「一个空格 + 内容」算
func commandLength(exe string, args []string) int {
n := len(exe)
for _, a := range args {
n += len(a) + 1
}
return n
}
// 预估成片体积时给音频留的码率(Kbps):分块时音频固定重编码成 192k,
// 单块时是直接拷流,这里统一按 192 估,反正渲染过 5% 后会被实测外推值顶掉
const estimateAudioKbps = 192
@@ -293,8 +327,6 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error {
// 弹道按最终字号统一重排,避免还原弹幕或改字号后同弹道追尾
relayoutLanes(records, meta)
// 下载并圆形裁切头像
avatarPaths := make(map[int64]string)
// 缓存放在程序所在目录,换保存目录(甚至换盘)都不用重新下载;
// 旧版本放在弹幕文件同目录,作为回退来源读一次并搬过去
cacheDir := avatarCachePath()
@@ -314,34 +346,9 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error {
urlByUID[r.UID] = r.Avatar
}
}
looked := 0
jobSetPhase(jobPhaseAvatar)
jobAvatars(0, len(uids))
for i, uid := range uids {
// 这一阶段既要下网络又要解图片,允许在这里暂停 / 停止
if err := jobGate(); err != nil {
return err
}
url := urlByUID[uid]
if url == "" {
// 从 .ass 还原出来的弹幕没有头像链接,按 uid 反查一次
url = avatarURLByUID(uid)
looked++
// 反查比较慢,定期报一下进度,避免看起来像卡住了
if looked%50 == 0 || i+1 == len(uids) {
lPrintf("正在获取头像:已处理 %d/%d 个用户", i+1, len(uids))
}
}
jobAvatars(i+1, len(uids))
if url == "" {
continue
}
if path := ensureAvatar(uid, url, cacheDir, legacyCache); path != "" {
avatarPaths[uid] = path
}
}
if looked > 0 {
lPrintf("有 %d 个用户没有自带头像链接(多半是从 .ass 还原的弹幕),已按 uid 向 AcFun 反查", looked)
avatarPaths, _, err := fetchAvatars(uids, urlByUID, cacheDir, legacyCache)
if err != nil {
return err
}
if len(avatarPaths) == 0 {
lPrintWarn("没有任何头像可用,弹幕将只烧录文字,请检查网络后重试合并")
@@ -395,6 +402,111 @@ func burnDanmu(jsonlFile, videoFile, outFile string, offset float64) error {
return ctx.burnChunked(chunks)
}
// 获取所有用户的头像:没有自带链接的先按 uid 查一次,再决定用已有缓存还是下载。
// 之所以并发:查链接实测约 1 秒/个,几百人的场次串行做要好几分钟。
// 返回 uid → 缓存文件路径、本阶段统计;暂停时在闸门处等待,停止时返回 errJobStopped。
func fetchAvatars(uids []int64, urlByUID map[int64]string, cacheDir, legacyCache string) (map[int64]string, avatarStat, error) {
type avatarTask struct {
uid int64
url string
}
tasks := make([]avatarTask, 0, len(uids))
for _, uid := range uids {
tasks = append(tasks, avatarTask{uid: uid, url: urlByUID[uid]})
}
paths := make(map[int64]string, len(tasks))
var stat avatarStat
jobSetPhase(jobPhaseAvatar)
jobAvatars(0, len(tasks))
if len(tasks) == 0 {
return paths, stat, nil
}
lPrintf("开始获取 %d 个用户的头像(%d 路并发)", len(tasks), avatarFetchWorkers)
workers := avatarFetchWorkers
if workers > len(tasks) {
workers = len(tasks)
}
// 缓冲足够大:worker 因停止提前退出时,投递端也不会卡住
jobs := make(chan avatarTask, len(tasks))
for _, t := range tasks {
jobs <- t
}
close(jobs)
var (
mu sync.Mutex
stopErr error
)
var wg sync.WaitGroup
for w := 0; w < workers; w++ {
wg.Add(1)
go func() {
defer wg.Done()
for t := range jobs {
// 每个任务开头过闸门:暂停时在这里等,停止时直接收工
if err := jobGate(); err != nil {
mu.Lock()
if stopErr == nil {
stopErr = err
}
mu.Unlock()
return
}
url, fetched := t.url, false
if url == "" {
// 从 .ass 还原出来的弹幕没有头像链接,按 uid 查一次(结果会落盘)
url, fetched = avatarURLByUID(t.uid, cacheDir)
}
// 下载前先看这个头像在不在,用来区分「复用缓存」和「真的下载了」
existed := url != "" && fileExists(filepath.Join(cacheDir, avatarCacheName(t.uid, url)))
path := ""
if url != "" {
path = ensureAvatar(t.uid, url, cacheDir, legacyCache)
}
mu.Lock()
stat.done++
if fetched {
stat.fetched++
} else if t.url == "" {
stat.reusedLink++
}
switch {
case existed:
stat.imgReused++
case path != "":
stat.imgDownloaded++
case url != "":
stat.failed++
}
if path != "" {
paths[t.uid] = path
}
done := stat.done
mu.Unlock()
jobAvatars(done, len(tasks))
}
}()
}
wg.Wait()
if stopErr != nil {
return paths, stat, stopErr
}
summary := fmt.Sprintf("头像处理完成:%d 个用户,链接复用 %d / 向 AcFun 查 %d,图片复用缓存 %d / 新下载 %d",
len(tasks), stat.reusedLink, stat.fetched, stat.imgReused, stat.imgDownloaded)
if stat.failed > 0 {
summary += fmt.Sprintf(",%d 个不可用(这些用户只烧文字)", stat.failed)
}
lPrintf("%s", summary)
return paths, stat, nil
}
// 渲染前按最终字号重排弹道:jsonl 里的 lane/width 是录制或 .ass 还原时算的,
// 从原格式 .ass 还原的弹幕没把昵称和头像算进宽度(占位提前放行了同弹道的下一条,
// 画面上就追尾叠字),改过字号后更是全对不上。这里用和录制时相同的碰撞公式、
@@ -814,6 +926,15 @@ func (c *burnContext) renderChunk(chunk danmuChunk, assBase, out string, cut boo
args = append(args, "-t", formatSeconds(chunk.end-chunk.start))
}
// 自检:每条弹幕都要生成一段 overlay,条数一多整个命令行就会顶破 Windows 的
// 32767 字符上限,而 exec 只会回一句 "The filename or extension is too long",
// 完全看不出跟弹幕数量有关(这个构建的 ffmpeg 也没有 -filter_complex_script 可以绕)。
// 这里按实际参数估长,超了就给一句能定位的错,而不是让它烂在 exec 里
if n := commandLength(c.ffmpegFile, args); n > maxCommandLineLen {
return fmt.Errorf("ffmpeg 命令过长(%d 字符,Windows 上限 %d 左右):这一块有 %d 条弹幕、%d 个头像,请调小单块弹幕数",
n, maxCommandLineLen, len(chunk.records), len(uidOrder))
}
if !burnMode {
// 透明头像层模式固定用 ProRes 4444,不能再追加编码器覆盖
full := append(append([]string{}, args...), out)
@@ -1054,17 +1175,28 @@ func fileSizeOrZero(path string) int64 {
return 0
}
// 文件存不存在(只关心存在性,不关心是不是目录)
func fileExists(path string) bool {
_, err := os.Stat(path)
return err == nil
}
// 把秒格式化成 ffmpeg 参数,保留毫秒
func formatSeconds(sec float64) string {
return strconv.FormatFloat(sec, 'f', 3, 64)
}
// 头像缓存的文件名:uid + 头像链接的摘要。链接变了文件名跟着变,
// 主播换头像后不会一直用着旧缓存
func avatarCacheName(uid int64, url string) string {
sum := sha1.Sum([]byte(url))
return fmt.Sprintf("%d_%x.png", uid, sum[:6])
}
// 下载头像并圆形裁切,失败自动重试,返回缓存的 PNG 路径,失败返回空字符串。
// legacyDirs 是旧版本放缓存的位置,命中就搬过来用,省得换保存目录后重新下一次。
func ensureAvatar(uid int64, url, cacheDir string, legacyDirs ...string) string {
// 文件名里带上头像链接的摘要,主播换头像后不会一直用着旧缓存
sum := sha1.Sum([]byte(url))
name := fmt.Sprintf("%d_%x.png", uid, sum[:6])
name := avatarCacheName(uid, url)
path := filepath.Join(cacheDir, name)
if _, err := os.Stat(path); err == nil {
return path
@@ -1189,36 +1321,187 @@ func decodeAvatar(data []byte, contentType string) (image.Image, error) {
return src, nil
}
// uid → 头像链接的记忆,包含查不到的结果,避免同一个 uid 反复去问
// uid → 头像链接的记录,包含查不到的结果,避免同一个 uid 反复去问。
// 查到的结果还会落盘(见 avatarURLCacheFile),否则进程一重启就要把整场 uid 重查一遍。
var avatarURLMemo struct {
sync.Mutex
m map[int64]string
m map[int64]string
at map[int64]int64 // 记录写入时刻(Unix 秒),0 表示本进程刚查到的
dir string // 落盘文件所在目录,换目录会重新加载
loaded bool
}
// 反查结果的落盘文件(放在头像缓存目录里)与有效期。
// 有效期是为了让主播换头像后能自己更新过来:过期就重新查一次,查到了用新的
const (
avatarURLCacheFile = "uid_urls.jsonl"
avatarURLCacheTTL = 7 * 24 * time.Hour
)
type avatarURLCacheEntry struct {
UID int64 `json:"uid"`
URL string `json:"url"`
At int64 `json:"at"` // 写入时刻(Unix 秒)
}
// 实际去查头像的动作,单独拎出来方便测试替换掉
var lookupAvatarURL = fetchAvatarURL
// 按 uid 反查头像链接。原版 .ass 里没有头像链接,只能这样补回来
func avatarURLByUID(uid int64) string {
// 读取落盘的反查记录。返回整理后的记录,以及文件里原有的行数(行数多于记录数
// 说明有重复或过期的行,调用方会顺手把文件压一遍)
func loadAvatarURLCache(dir string) (map[int64]avatarURLCacheEntry, int, error) {
path := filepath.Join(dir, avatarURLCacheFile)
f, err := os.Open(path)
if err != nil {
return nil, 0, err
}
defer f.Close()
entries := make(map[int64]avatarURLCacheEntry, 256)
total := 0
now := time.Now().Unix()
ttl := int64(avatarURLCacheTTL / time.Second)
scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 64*1024), 64*1024)
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
}
var e avatarURLCacheEntry
if json.Unmarshal([]byte(line), &e) != nil || e.UID <= 0 || e.URL == "" {
continue
}
total++
if now-e.At > ttl {
continue // 过期了:主播可能换过头像,重新查
}
entries[e.UID] = e // 后写的覆盖先写的
}
if err := scanner.Err(); err != nil {
return entries, total, err
}
return entries, total, nil
}
// 重写整个落盘文件(先写临时文件再改名,避免写一半断电留个坏文件)
func saveAvatarURLCache(dir string, entries map[int64]avatarURLCacheEntry) error {
if err := os.MkdirAll(dir, 0755); err != nil {
return err
}
path := filepath.Join(dir, avatarURLCacheFile)
tmp := path + ".tmp"
f, err := os.Create(tmp)
if err != nil {
return err
}
w := bufio.NewWriter(f)
for _, e := range entries {
b, err := json.Marshal(e)
if err != nil {
continue
}
w.Write(b)
w.WriteByte('\n')
}
if err := w.Flush(); err != nil {
f.Close()
return err
}
if err := f.Close(); err != nil {
return err
}
return os.Rename(tmp, path)
}
// 追加一条反查结果(调用方持锁)。写不进去最多下次多查一遍,不影响合并
func appendAvatarURLCache(dir string, uid int64, url string) {
if url == "" {
return
}
if err := os.MkdirAll(dir, 0755); err != nil {
return
}
f, err := os.OpenFile(filepath.Join(dir, avatarURLCacheFile),
os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
if err != nil {
return
}
defer f.Close()
b, err := json.Marshal(avatarURLCacheEntry{UID: uid, URL: url, At: time.Now().Unix()})
if err != nil {
return
}
b = append(b, '\n')
_, _ = f.Write(b)
}
// 把落盘记录读进内存(调用方持锁)。本进程刚查到的结果优先,不被文件里的旧值覆盖
func loadAvatarURLCacheInto(dir string) {
if avatarURLMemo.loaded && avatarURLMemo.dir == dir {
return
}
entries, total, err := loadAvatarURLCache(dir)
if err != nil && !os.IsNotExist(err) {
lPrintWarnf("读取头像链接记录失败,本次仍会按 uid 反查:%v", err)
}
if avatarURLMemo.m == nil {
avatarURLMemo.m = make(map[int64]string, len(entries))
}
if avatarURLMemo.at == nil {
avatarURLMemo.at = make(map[int64]int64, len(entries))
}
added := 0
for uid, e := range entries {
if _, ok := avatarURLMemo.m[uid]; ok {
continue
}
avatarURLMemo.m[uid] = e.URL
avatarURLMemo.at[uid] = e.At
added++
}
avatarURLMemo.dir = dir
avatarURLMemo.loaded = true
if total > len(entries) {
// 有重复或过期的行,顺手压一遍,免得文件一直长
_ = saveAvatarURLCache(dir, entries)
}
if added > 0 {
lPrintf("头像链接记录:从 %s 读到 %d 个用户的链接,这些不用再查", avatarURLCacheFile, added)
}
}
// 按 uid 反查头像链接。原版 .ass 里没有头像链接,只能这样补回来。
// 查到的结果会落盘,重启进程后不用把整场 uid 重新查一遍。
// 第二个返回值表示这次是不是真的发了请求(false = 用了内存或落盘里的记录)
func avatarURLByUID(uid int64, cacheDir string) (string, bool) {
if uid <= 0 {
return ""
return "", false
}
avatarURLMemo.Lock()
if avatarURLMemo.m == nil {
avatarURLMemo.m = make(map[int64]string)
}
loadAvatarURLCacheInto(cacheDir)
if url, ok := avatarURLMemo.m[uid]; ok {
avatarURLMemo.Unlock()
return url
return url, false
}
avatarURLMemo.Unlock()
url := lookupAvatarURL(uid)
avatarURLMemo.Lock()
if avatarURLMemo.m == nil {
avatarURLMemo.m = make(map[int64]string)
}
if avatarURLMemo.at == nil {
avatarURLMemo.at = make(map[int64]int64)
}
avatarURLMemo.m[uid] = url
avatarURLMemo.at[uid] = time.Now().Unix()
if url != "" {
appendAvatarURLCache(cacheDir, uid, url)
}
avatarURLMemo.Unlock()
return url
return url, true
}
// 查头像链接用的 client,同样复用连接