// 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 }