65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import aiohttp
now = time.monotonic()
if self.open_until[host] > now:
return True
if self.open_until[host] > 0 and self.open_until[host] <= now:
# 半开状态，允许试探
self.open_until[host] = 0
self.failure_count[host] = 0
return False
def record_failure(self, host):
self.failure_count[host] += 1
if self.failure_count[host] >= 3:
self.open_until[host] = time.monotonic() + self.recovery_time
logger.warning(f"circuit breaker opened: {host}")
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),
}
待会儿见
K哥馆
mayun
文鼎_应老师
课课家运营团队
liangchsh
启程软考
