update

2026-09-20 00:44:47 +08:00
parent d8f667e391
commit fde57984e7
16 changed files with 3908 additions and 123 deletions
--- a/internal/fetch/fetch.go
+++ b/internal/fetch/fetch.go
@ -0,0 +1,155 @@
+// Package fetch 提供「全局并发限流 + 自动重试」的 HTTP GET，供各爬虫共用。
+//
+// 为什么需要它：
+//   - 此前 vogue / theimpression 各自实现了一份 request()，逻辑重复、错误被吞掉（失败只返回 ("", 0)）；
+//   - 更严重的是并发失控：vogue 在「品牌级」与「详情级」各开一层信号量，实际并发是 MaxCo²
+//     （如 5×5=25），与「最多 5 个并发」的本意不符。
+//
+// 本包用一个「全局信号量」统一约束进程内所有请求的并发上限，各爬虫共享同一个 Client
+// 即可保证整体并发受控，不会因嵌套而放大。
+package fetch
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"time"
+)
+
+const (
+	// DefaultTimeout 单次请求默认超时。
+	DefaultTimeout = 30 * time.Second
+	// DefaultMaxConcurrency 默认全局并发上限。
+	DefaultMaxConcurrency = 5
+	// DefaultRetries 默认重试次数（不含首次）。
+	DefaultRetries = 2
+
+	userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " +
+		"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+	acceptHeader = "text/html,application/json,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
+)
+
+// Options 构造 Client 的参数，零值字段自动回落默认。
+type Options struct {
+	MaxConcurrency int           // 全局并发上限（<=0 回落 5）
+	MinInterval    time.Duration // 两次请求之间的最小间隔，全局限速（0 = 不限速）
+	Retries        int           // 失败重试次数，不含首次（<0 回落 2；0 表示不重试）
+	Backoff        time.Duration // 重试基础退避，第 n 次等待 n*Backoff（<=0 回落 1s）
+	Timeout        time.Duration // 单次请求超时（<=0 回落 30s）
+}
+
+// Client 带并发限流与重试的 HTTP 客户端。
+type Client struct {
+	hc      *http.Client
+	sem     chan struct{} // 全局并发槽（跨所有调用方共享）
+	tick    *time.Ticker  // 非 nil 时按 MinInterval 全局限速
+	retries int
+	backoff time.Duration
+}
+
+// New 构造 Client。返回值实现了一个进程级共享的限流器，多个爬虫应复用同一个实例。
+func New(opt Options) *Client {
+	if opt.MaxConcurrency <= 0 {
+		opt.MaxConcurrency = DefaultMaxConcurrency
+	}
+	if opt.Timeout <= 0 {
+		opt.Timeout = DefaultTimeout
+	}
+	if opt.Backoff <= 0 {
+		opt.Backoff = time.Second
+	}
+	if opt.Retries < 0 {
+		opt.Retries = DefaultRetries
+	}
+	c := &Client{
+		hc:      &http.Client{Timeout: opt.Timeout},
+		sem:     make(chan struct{}, opt.MaxConcurrency),
+		retries: opt.Retries,
+		backoff: opt.Backoff,
+	}
+	if opt.MinInterval > 0 {
+		c.tick = time.NewTicker(opt.MinInterval)
+	}
+	return c
+}
+
+// Close 停止内部限速计时器（未开启限速时为空操作）。
+func (c *Client) Close() {
+	if c.tick != nil {
+		c.tick.Stop()
+	}
+}
+
+// Get 发起带限流与重试的 GET，返回响应体与最终 HTTP 状态码。
+//
+// 重试策略：网络错误 / 5xx / 429 会按「线性退避」重试（第 n 次等 n*Backoff）；
+// 其余 4xx 视为客户端错误（如 403 被反爬、404 页面不存在），不重试，直接返回。
+// 调用方拿到非 nil error 时即可判定最终失败，无需再自行区分网络错误与状态码。
+func (c *Client) Get(ctx context.Context, url string) ([]byte, int, error) {
+	// 全局限速：每次请求前消费一个 tick（ticker 通道容量为 1，不会无限堆积）。
+	if c.tick != nil {
+		select {
+		case <-c.tick.C:
+		case <-ctx.Done():
+			return nil, 0, ctx.Err()
+		}
+	}
+	// 全局并发上限：超过则在此排队，保证总量受控。
+	select {
+	case c.sem <- struct{}{}:
+		defer func() { <-c.sem }()
+	case <-ctx.Done():
+		return nil, 0, ctx.Err()
+	}
+
+	var lastErr error
+	var lastStatus int
+	for attempt := 0; attempt <= c.retries; attempt++ {
+		if attempt > 0 {
+			delay := c.backoff * time.Duration(attempt)
+			log.Printf("[fetch] 第 %d/%d 次重试 %s（等待 %s）", attempt, c.retries, url, delay)
+			select {
+			case <-time.After(delay):
+			case <-ctx.Done():
+				return nil, 0, ctx.Err()
+			}
+		}
+		body, status, err := c.do(ctx, url)
+		if err == nil && status == http.StatusOK {
+			return body, status, nil
+		}
+		lastErr, lastStatus = err, status
+		// 4xx（除 429 限流）不重试：重试也只会再次失败。
+		if status >= 400 && status < 500 && status != http.StatusTooManyRequests {
+			break
+		}
+	}
+	if lastErr == nil {
+		lastErr = fmt.Errorf("http %d", lastStatus)
+	}
+	return nil, lastStatus, lastErr
+}
+
+// do 执行单次请求。
+func (c *Client) do(ctx context.Context, url string) ([]byte, int, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, 0, err
+	}
+	req.Header.Set("User-Agent", userAgent)
+	req.Header.Set("Accept", acceptHeader)
+
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return nil, 0, err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, resp.StatusCode, err
+	}
+	return body, resp.StatusCode, nil
+}
--- a/internal/fetch/fetch_test.go
+++ b/internal/fetch/fetch_test.go
@ -0,0 +1,128 @@
+package fetch
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestGetSuccess(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("ok"))
+	}))
+	defer srv.Close()
+
+	c := New(Options{MaxConcurrency: 1, Retries: 0})
+	defer c.Close()
+
+	body, code, err := c.Get(context.Background(), srv.URL)
+	if err != nil || code != http.StatusOK || string(body) != "ok" {
+		t.Fatalf("期望 200/ok，得到 code=%d body=%q err=%v", code, body, err)
+	}
+}
+
+// TestGetRetriesOn5xx 5xx 应重试到上限后失败（首次 + 2 次重试 = 3 次请求）。
+func TestGetRetriesOn5xx(t *testing.T) {
+	var calls int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&calls, 1)
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	c := New(Options{MaxConcurrency: 1, Retries: 2, Backoff: time.Millisecond})
+	defer c.Close()
+
+	_, code, err := c.Get(context.Background(), srv.URL)
+	if err == nil {
+		t.Fatal("500 最终应返回 error")
+	}
+	if code != http.StatusInternalServerError {
+		t.Fatalf("最终状态码应为 500，得到 %d", code)
+	}
+	if got := atomic.LoadInt32(&calls); got != 3 {
+		t.Fatalf("500 应请求 3 次（首次+2 重试），实际 %d", got)
+	}
+}
+
+// TestGetNoRetryOn4xx 4xx（非 429）不应重试。
+func TestGetNoRetryOn4xx(t *testing.T) {
+	var calls int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		atomic.AddInt32(&calls, 1)
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	defer srv.Close()
+
+	c := New(Options{MaxConcurrency: 1, Retries: 3, Backoff: time.Millisecond})
+	defer c.Close()
+
+	_, code, err := c.Get(context.Background(), srv.URL)
+	if err == nil || code != http.StatusNotFound {
+		t.Fatalf("404 应报错且状态码 404，得到 code=%d err=%v", code, err)
+	}
+	if got := atomic.LoadInt32(&calls); got != 1 {
+		t.Fatalf("404 不应重试（只请求 1 次），实际 %d", got)
+	}
+}
+
+// TestGetRetriesOn429 429（限流）应重试，后续成功则返回成功。
+func TestGetRetriesOn429(t *testing.T) {
+	var calls int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if atomic.AddInt32(&calls, 1) == 1 {
+			w.WriteHeader(http.StatusTooManyRequests)
+			return
+		}
+		_, _ = w.Write([]byte("ok"))
+	}))
+	defer srv.Close()
+
+	c := New(Options{MaxConcurrency: 1, Retries: 2, Backoff: time.Millisecond})
+	defer c.Close()
+
+	body, code, err := c.Get(context.Background(), srv.URL)
+	if err != nil || code != http.StatusOK || string(body) != "ok" {
+		t.Fatalf("429 后重试应成功，得到 code=%d body=%q err=%v", code, body, err)
+	}
+}
+
+// TestConcurrencyLimit 并发上限不得被突破（修复 MaxCo² 回归的守门测试）。
+func TestConcurrencyLimit(t *testing.T) {
+	var inflight, peak int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		n := atomic.AddInt32(&inflight, 1)
+		for {
+			old := atomic.LoadInt32(&peak)
+			if n <= old || atomic.CompareAndSwapInt32(&peak, old, n) {
+				break
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+		atomic.AddInt32(&inflight, -1)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	defer srv.Close()
+
+	const limit = 2
+	c := New(Options{MaxConcurrency: limit, Retries: 0})
+	defer c.Close()
+
+	var wg sync.WaitGroup
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, _, _ = c.Get(context.Background(), srv.URL)
+		}()
+	}
+	wg.Wait()
+
+	if m := atomic.LoadInt32(&peak); m > limit {
+		t.Fatalf("并发峰值不应超过 %d，实际 %d", limit, m)
+	}
+}