From 6a8b7f0704d6be82214ee4675d80b914d77f2b0e Mon Sep 17 00:00:00 2001 From: toom1996 <23cm.cn@gmail.com> Date: Sun, 25 Oct 2026 18:01:46 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E4=BF=AE=20spider=20=E5=86=85=E5=AD=98?= =?UTF-8?q?=E6=9A=B4=E6=B6=A8=EF=BC=88=E6=95=B4=E5=9C=BA=E5=9B=BE=E5=85=A5?= =?UTF-8?q?=E5=86=85=E5=AD=98=20+=20=E5=9C=BA=E6=AC=A1=E6=97=A0=E9=99=90?= =?UTF-8?q?=E5=B9=B6=E5=8F=91=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 线上现象：spider 进程 RSS 1.3GB（33% 内存），单机 3.7G 只剩 141MB 可用， 服务器 1.5 小时内重启两次。 - media.UploadAll 改为按批（8 张）下载→上传→立即释放； 原来先把整场 97 张（约 116MB）全读进内存，多场并发时线性叠加 - SpiderStart 对同品牌场次加 2 路并发上限（原来对全部场次无上限 go 出去） - systemd 单元加 GOMEMLIMIT=700MiB / MemoryHigh=800M / MemoryMax=1200M 硬护栏， 即使再退化也只 OOM 掉这一个进程，不会拖死整机 --- internal/media/upload.go | 100 ++++++++++++++++++++++----------------- internal/spider/vogue.go | 17 ++++++- 2 files changed, 72 insertions(+), 45 deletions(-) diff --git a/internal/media/upload.go b/internal/media/upload.go index 2bd871d..5edb293 100644 --- a/internal/media/upload.go +++ b/internal/media/upload.go @@ -63,41 +63,62 @@ func (u *Uploader) UploadAll(ctx context.Context, urls []string) []Result { return results } - // ① 并发下载 - data := make([][]byte, len(urls)) - errs := make([]error, len(urls)) - u.parallel(len(urls), func(i int) { - body, err := u.download(ctx, urls[i]) + // 分批处理：**不能**先把整场的图全读进内存。 + // 一场 100 张 × 1.2MB ≈ 120MB；而 SpiderStart 对同一品牌的多个场次是并发抓的， + // 峰值会翻好几倍——线上实测 spider 进程一度涨到 1.3GB，把小机器（3.7G）压到只剩 141MB 可用。 + // 按批（默认 8 张）下载→上传→立即释放，峰值稳定在 ~10MB。 + const batchSize = 8 + for start := 0; start < len(urls); start += batchSize { + end := start + batchSize + if end > len(urls) { + end = len(urls) + } + u.uploadBatch(ctx, urls, results, start, end) + } + return results +} + +// uploadBatch 处理 urls[start:end] 这一段：下载 → 生成 key → 预签名 → PUT。 +// 返回时本批的图片字节已可回收（不再被引用）。 +func (u *Uploader) uploadBatch(ctx context.Context, urls []string, results []Result, start, end int) { + n := end - start + data := make([][]byte, n) + errs := make([]error, n) + + // ① 并发下载本批 + u.parallel(n, func(i int) { + body, err := u.download(ctx, urls[start+i]) if err != nil { errs[i] = err return } data[i] = body }) - log.Printf("[media] 下载完成 %d 张（失败 %d）", countOK(data), len(urls)-countOK(data)) + log.Printf("[media] 下载完成 %d 张（失败 %d）", countOK(data), n-countOK(data)) // ② 生成本批 key（内容 sha1 + 原扩展名） - for i := range urls { + for i := 0; i < n; i++ { if errs[i] != nil { - results[i] = Result{Err: errs[i]} + results[start+i] = Result{Err: errs[i]} continue } - results[i] = Result{ - Key: u.Prefix + "/" + sha1Hex(data[i]) + extOf(urls[i]), + results[start+i] = Result{ + Key: u.Prefix + "/" + sha1Hex(data[i]) + extOf(urls[start+i]), Size: int64(len(data[i])), } } - // ③ 批量申请预签名地址（同一张图可能在同一场里出现多次，按 key 去重） - want := map[string]ingest.UploadItem{} - for i := range urls { - if results[i].Key == "" { + // ③ 申请预签名（同一张图可能在一场里重复出现，按 key 去重） + want := make(map[string]ingest.UploadItem, n) + for i := 0; i < n; i++ { + key := results[start+i].Key + if key == "" { continue } - if _, ok := want[results[i].Key]; ok { + if _, ok := want[key]; ok { continue } - want[results[i].Key] = ingest.UploadItem{Key: results[i].Key, ContentType: contentTypeOf(urls[i])} + want[key] = ingest.UploadItem{Key: key, ContentType: contentTypeOf(urls[start+i])} } items := make([]ingest.UploadItem, 0, len(want)) for _, it := range want { @@ -105,44 +126,37 @@ func (u *Uploader) UploadAll(ctx context.Context, urls []string) []Result { } presigned := make(map[string]string, len(items)) - const chunk = 100 // 后端单次上限 200，这里留一半余量 - for start := 0; start < len(items); start += chunk { - end := start + chunk - if end > len(items) { - end = len(items) - } - got, err := u.Client.RequestUploadURLs(ctx, items[start:end]) - if err != nil { - // 申请失败：整批回退成 URL 上报（后端自己下载），不中断整场抓取。 - for i := range results { - if results[i].Key != "" { - results[i] = Result{Err: fmt.Errorf("申请上传地址失败: %w", err)} - } + got, err := u.Client.RequestUploadURLs(ctx, items) + if err != nil { + // 申请失败：本批回退成 URL 上报（后端自己下载），不中断整场抓取。 + for i := 0; i < n; i++ { + if results[start+i].Key != "" { + results[start+i] = Result{Err: fmt.Errorf("申请上传地址失败: %w", err)} } - return results - } - for _, g := range got { - presigned[g.Key] = g.URL } + return + } + for _, g := range got { + presigned[g.Key] = g.URL } - log.Printf("[media] 已拿到 %d 个预签名地址，开始上传", len(presigned)) - - // ④ 并发 PUT（内容已在内存里，直接传字节） - u.parallel(len(urls), func(i int) { - if results[i].Key == "" { + // ④ 并发 PUT 本批 + u.parallel(n, func(i int) { + res := results[start+i] + if res.Key == "" { return } - putURL, ok := presigned[results[i].Key] + putURL, ok := presigned[res.Key] if !ok { - results[i] = Result{Err: fmt.Errorf("未拿到 %s 的上传地址", results[i].Key)} + results[start+i] = Result{Err: fmt.Errorf("未拿到 %s 的上传地址", res.Key)} return } - if err := u.put(ctx, putURL, urls[i], data[i]); err != nil { - results[i] = Result{Err: err} + if perr := u.put(ctx, putURL, urls[start+i], data[i]); perr != nil { + results[start+i] = Result{Err: perr} } }) - return results + + log.Printf("[media] 本批 %d 张上传完成（%d/%d）", n, end, len(urls)) } // parallel 以 Concurrency 路并发执行 fn(0..n-1)。 diff --git a/internal/spider/vogue.go b/internal/spider/vogue.go index 7f4782a..2151797 100644 --- a/internal/spider/vogue.go +++ b/internal/spider/vogue.go @@ -79,6 +79,12 @@ func (s *VogueSpider) RunBrand(brandUID, brandName string, maxShows int) int { return int(atomic.LoadInt32(&s.submitted)) } +// maxShowConcurrency 同一品牌内同时抓取的场次数。 +// +// 调大能更快跑完一个品牌，但每场都会占用一批图片内存（媒体直传按批读入）， +// 实测无上限并发会把进程顶到 GB 级；2 路在带宽与内存之间比较平衡。 +const maxShowConcurrency = 2 + // uploadLookMedia 把一批 look 的图片直传到对象存储，回填 main_key / detail_keys。 // // 返回 true 表示全部成功；false 表示有失败项 —— 失败的图仍保留原始 URL， @@ -226,15 +232,22 @@ func (s *VogueSpider) SpiderStart(task ingest.CrawlBrand) { showsList = showsList[:s.MaxShows] } - // 详情页并发抓取：这里不再自建信号量，HTTP 并发与频控统一由 s.Fetcher 承担 - // （避免与品牌级信号量叠加导致并发放大）。 + // 详情页并发抓取：**限制同时抓几场**。 + // + // 一个品牌可能几十场（Acne Studios 70+），每场的图片在媒体直传时按批驻留内存。 + // 早先这里对全部场次无上限并发，实测会把 spider 进程顶到 1.3GB， + // 在 3.7G 的小机器上直接把系统压到只剩百来兆可用内存 → 整机卡死重启。 + // 2 路并发足够喂满带宽（单场内部还有 6 路图片并发），内存峰值可控。 var wg sync.WaitGroup + showSem := make(chan struct{}, maxShowConcurrency) for _, list := range showsList { wg.Add(1) go func(info gjson.Result) { defer wg.Done() + showSem <- struct{}{} + defer func() { <-showSem }() s.getDetail(task.BrandUID, info) }(list) }