156 lines
4.9 KiB
Go
156 lines
4.9 KiB
Go
// 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
|
||
}
|