异步爬虫里 aiohttp 超时设多少不会误杀正常请求

原创

小白学大数据

发布于 2026-07-23 16:56:44

830

超时设 3 秒，跑半小时成功率从 98% 掉到 85%。目标站点 RTT 正常，代理 IP 存活，但大量请求抛 ServerTimeoutError。问题出在超时阈值与实际网络时序不匹配，把正常的处理延迟当成了网络故障。aiohttp 超时的实现机制aiohttp 的 ClientTimeout 底层依赖 asyncio 的定时器协程。每一层超时对应一个独立的 asyncio.wait_for 或 loop.call_later 调度，触发时取消对应的协程并抛出 asyncio.TimeoutError，aiohttp 再包装成 aiohttp.ServerTimeoutError。

三层超时的触发顺序和覆盖关系：

关键点：sock_read 计的是间隔不是总量。服务器花 8 秒生成页面，但生成后一次性返回，sock_read=10 不会触发。但如果服务器先返回头，然后 11 秒后才返回 body，sock_read=10 会触发。connect 超时：握手时序分析

握手时序分解：

走代理时 connect 阶段涉及多段 RTT 叠加，这是代理场景下 connect 需要调大的根本原因。sock_read 超时：事件循环与读事件

# 全局默认超时
DEFAULT_TIMEOUT = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)

# 慢接口专用超时
SLOW_API_TIMEOUT = aiohttp.ClientTimeout(total=60, connect=5, sock_read=30)

# 快接口专用超时（对响应速度有要求的场景）
FAST_API_TIMEOUT = aiohttp.ClientTimeout(total=10, connect=3, sock_read=5)


async def mixed_requests():
    connector = aiohttp.TCPConnector(limit=100, limit_per_host=10)

    async with aiohttp.ClientSession(
        timeout=DEFAULT_TIMEOUT,
        connector=connector,
    ) as session:
        # 用 session 默认超时
        async with session.get("https://api.example.com/list") as resp:
            data = await resp.json()

        # 慢接口在 request 级别覆盖
        async with session.get(
            "https://api.example.com/report",
            timeout=SLOW_API_TIMEOUT,
        ) as resp:
            report = await resp.json()

        # 快接口在 request 级别覆盖
        async with session.get(
            "https://api.example.com/health",
            timeout=FAST_API_TIMEOUT,
        ) as resp:
            status = resp.status

aiohttp 的读取流程：

sock_read 只在 read() 阻塞期间计时。数据一到，read() 返回，计时器取消，下一次 read() 重新开始计时。所以设 10 秒不影响 200ms 内完成的正常请求。容易误杀的场景：

场景	服务端行为	sock_read=3s 的结果
动态页面渲染	2-5s 无数据，然后一次性返回	误杀
数据库查询接口	3-8s 无数据，然后返回	误杀
chunked 分块传输	每 4s 发一块	误杀
大文件下载	持续有数据流	正常
静态页面	200ms 内返回	正常

total 超时与连接池的交互

# 错误：不设 total
# 如果服务端以极慢速率持续返回数据（每次间隔 < sock_read），
# 请求会长时间占用连接池中的连接，导致 limit 耗尽

timeout_bad = aiohttp.ClientTimeout(sock_read=10)

# 错误：total <= sock_read
# total 先于 sock_read 触发，sock_read 配置无意义

timeout_bad2 = aiohttp.ClientTimeout(total=10, sock_read=10)

# 正确：total 远大于正常请求耗时，但小于连接池泄漏容忍阈值

timeout_ok = aiohttp.ClientTimeout(total=30, sock_read=10)

total 超时和连接池容量直接相关。TCPConnector 的 limit 控制最大并发连接数，如果一个请求因为服务端慢速响应而长时间不释放，就会挤占其他请求的连接配额。

# 连接池配置与超时的协同
connector = aiohttp.TCPConnector(
    limit=100,              # 全局最大并发连接
    limit_per_host=10,      # 单个目标站点最大并发连接
    keepalive_timeout=30,   # 空闲连接保活时间（秒）
    enable_cleanup_closed=True,  # 关闭半开连接清理，防止 FD 泄漏
)

# 超时配置要和连接池配匹配：
# 如果 limit=100, total=60s，最坏情况下 100 个连接全被慢请求占满 60 秒
# 实际吞吐 = 100 / 60 ≈ 1.7 req/s，远低于预期
# 要么提高 limit，要么降低 total，根据目标站点响应时间算

代理链路超时分析代理引入的额外延迟分布在 connect 和 sock_read 两个阶段：

# 以亿牛云隧道代理为例
# 链路：client → 亿牛云代理服务器 → 目标站点

# connect 阶段增加：
#   client → proxy 的 TCP 握手（1 RTT）
#   proxy → target 的 TCP 握手（1 RTT，代理到目标的延迟）
#   CONNECT 隧道建立（1 RTT）
#   端到端 TLS 握手（2-3 RTT）

# sock_read 阶段增加：
#   代理服务器的转发缓冲（通常 50-200ms）
#   代理出口 IP 到目标站点的网络波动

PROXY_TIMEOUT = aiohttp.ClientTimeout(
    total=30,
    connect=8,       # 直连 5s + 代理额外 3s 余量
    sock_read=15,    # 直连 10s + 代理转发缓冲余量
)

亿牛云提供两种代理模式，超时策略不同：

import aiohttp
import asyncio
import logging
import random

logger = logging.getLogger(__name__)


class ProxyCrawler:
    """亿牛云代理爬虫：支持隧道代理和短效代理两种模式"""

    def __init__(self, mode="tunnel"):
        self.mode = mode

        if mode == "tunnel":
            # 隧道代理：固定地址，服务端自动切换出口 IP
            # 优点：不需要管理 IP 池
            # 缺点：无法控制具体出口 IP，切 IP 时请求可能中断
            self.proxy_url = "http://username:password@proxy.16yun.cn:9010"
            self.timeout = aiohttp.ClientTimeout(total=30, connect=8, sock_read=15)
        else:
            # 短效代理：拿到一批 IP，自行轮换
            # 优点：可以筛选高质量 IP
            # 缺点：需要维护 IP 池和健康检查
            self.proxy_pool = [
                "http://ip1:port", "http://ip2:port", "http://ip3:port",
            ]
            self.timeout = aiohttp.ClientTimeout(total=30, connect=8, sock_read=15)

        self.connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=10,
            keepalive_timeout=30,
            enable_cleanup_closed=True,
            # 代理场景下关闭 keepalive 可避免复用到坏 IP 的连接
            force_close=self.mode != "tunnel",
        )

    def _get_proxy(self):
        if self.mode == "tunnel":
            return self.proxy_url
        else:
            return random.choice(self.proxy_pool)

    async def fetch(self, url, session):
        """带重试的请求，超时后换代理 IP 重试"""
        max_retries = 3
        backoff = aiohttp.ClientTimeout(total=30, connect=8, sock_read=15)

        for attempt in range(max_retries):
            proxy = self._get_proxy()
            try:
                async with session.get(
                    url, proxy=proxy, timeout=self.timeout
                ) as resp:
                    if resp.status == 200:
                        return await resp.text()
                    if resp.status in (429, 502, 503):
                        # 限流或代理 IP 被封，退避后重试
                        await asyncio.sleep(1 + attempt * 2)
                        continue
                    logger.warning(f"HTTP {resp.status}: {url}")
                    return None
            except aiohttp.ServerTimeoutError:
                # 隧道代理：下次请求自动换 IP
                # 短效代理：random.choice 会选另一个 IP
                logger.warning(
                    f"timeout attempt {attempt + 1}/{max_retries} "
                    f"proxy={proxy} url={url}"
                )
                await asyncio.sleep(0.5 * (attempt + 1))
            except aiohttp.ClientProxyConnectionError:
                # 代理连接失败，立即换 IP
                logger.warning(f"proxy dead: {proxy}")
                await asyncio.sleep(0.1)
            except aiohttp.ClientError as e:
                logger.warning(f"client error: {url} {type(e).__name__}: {e}")
                await asyncio.sleep(0.5)
        return None

注意：aiohttp 的 proxy 参数只支持 HTTP 代理（通过 CONNECT 方法建立隧道）。如果代理服务提供的是 SOCKS5 或 HTTPS 代理，需要用 aiohttp-socks 包：

session 级别 vs request 级别超时

生产级完整示例：带监控和熔断

import aiohttp
 def record_success(self, host):
 self.failure_count[host] = 0
 self.open_until[host] = 0
class ProductionCrawler:
 """生产级异步爬虫：代理 + 超时 + 重试 + 监控 + 熔断"""
 PROXY_URL = "http://username:password@proxy.16yun.cn:9010"
 def __init__(self, concurrency=50):
 self.timeout = aiohttp.ClientTimeout(total=30, connect=8, sock_read=15)
 self.connector = aiohttp.TCPConnector(
 limit=concurrency,
 limit_per_host=10,
 keepalive_timeout=30,
 enable_cleanup_closed=True,
 )
 self.semaphore = asyncio.Semaphore(concurrency)
 self.metrics = MetricsCollector()
 self.breaker = CircuitBreaker(threshold=0.3, recovery_time=60)
 async def fetch(self, url, session):
 host = url.split("/")[2]
 # 熔断检查
 if self.breaker.is_open(host):
 logger.info(f"skipped (circuit open): {host}")
 return None
 async with self.semaphore:
 start = time.monotonic()
 for attempt in range(3):
 try:
 async with session.get(
 url, proxy=self.PROXY_URL, timeout=self.timeout
 ) as resp:
 latency = time.monotonic() - start
 if resp.status == 200:
 text = await resp.text()
 self.metrics.record(host, latency)
 self.breaker.record_success(host)
 return text
 if resp.status in (429, 502, 503):
 await asyncio.sleep(1 + attempt * 2)
 continue
 self.metrics.record(host, latency, error=True)
 return None
 except aiohttp.ServerTimeoutError:
 logger.warning(f"timeout attempt {attempt + 1}: {url}")
 await asyncio.sleep(0.5 * (attempt + 1))
 except aiohttp.ClientError as e:
 logger.warning(f"error attempt {attempt + 1}: {url} {e}")
 await asyncio.sleep(0.5 * (attempt + 1))
 # 所有重试用完
 self.metrics.record(host, 0, timed_out=True)
 self.breaker.record_failure(host)
 return None
 async def crawl(self, urls):
 async with aiohttp.ClientSession(
 timeout=self.timeout,
 connector=self.connector,
 ) as session:
 tasks = [self.fetch(url, session) for url in urls]
 results = await asyncio.gather(*tasks, return_exceptions=True)
 # 输出指标
 for host, stats in self.metrics.summary().items():
 logger.info(f"metrics {host}: {stats}")
 return results
if __name__ == "__main__":
 urls = [f"https://example.com/page/{i}" for i in range(500)]
 crawler = ProductionCrawler(concurrency=50)
 asyncio.run(crawler.crawl(urls))

超时与重试的量化分析超时设太短再靠重试补救，本质上是把串行的等待时间拉长了：

# 场景：目标站点正常响应需要 4 秒
# 错误做法：total=2s + retry=3
# 每次 2s 超时 → 重试 → 2s 超时 → 重试 → 2s 超时
# 总耗时 6s，结果还是失败，白白浪费 3 次请求和 6 秒
BAD = aiohttp.ClientTimeout(total=2)
# 正确做法：total=30s + retry=2（仅对网络错误重试）
# 第一次请求 4s 成功，不需要重试
GOOD = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
重试应该留给确定性失败（连接断开、代理不可用），而不是用重试来掩盖超时配置不当。推荐配置矩阵
import aiohttp
# ┌─────────────────────┬────────┬─────────┬───────────┬───────────┐
# │ 场景                 │ total  │ connect │ sock_read │ 适用说明   │
# ├─────────────────────┼────────┼─────────┼───────────┼───────────┤
# │ 直连普通网站          │ 20     │ 5       │ 10        │ 响应 <2s   │
# │ 直连慢速网站          │ 60     │ 5       │ 30        │ 政府/查询   │
# │ 走代理普通网站        │ 30     │ 8       │ 15        │ 亿牛云隧道  │
# │ 走代理慢速网站        │ 90     │ 10      │ 40        │ 代理+慢站  │
# │ 走 SOCKS5 代理       │ 30     │ 10      │ 15        │ 额外握手   │
# └─────────────────────┴────────┴─────────┴───────────┴───────────┘
CONFIGS = {
 "direct_normal": aiohttp.ClientTimeout(total=20, connect=5, sock_read=10),
 "direct_slow": aiohttp.ClientTimeout(total=60, connect=5, sock_read=30),
 "proxy_normal": aiohttp.ClientTimeout(total=30, connect=8, sock_read=15),
 "proxy_slow": aiohttp.ClientTimeout(total=90, connect=10, sock_read=40),
 "proxy_socks5": aiohttp.ClientTimeout(total=30, connect=10, sock_read=15),
}

配置选型逻辑：先确认链路类型（直连 / HTTP 代理 / SOCKS5 代理），决定 connect 基准值。评估目标站点响应特征（静态 / 动态 / 慢查询），决定 sock_read 基准值。total 设为正常请求耗时的 3 到 5 倍，兜底防泄漏。上线后看 P99 延迟和超时率，超时率 >5% 就调大 sock_read，连接池频繁耗尽就调大 limit 或调小 total。这些值是起点。每个目标站点的网络特征不同，最终参数应该基于实际监控数据（P50/P99 延迟、超时率、错误率）持续调整。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

python