如何编写统一控制爬虫并发

wen 实用脚本 2026-06-26 30

从原理到实战，构建高效稳定的数据采集系统

目录导读

并发控制的必要性：为什么爬虫需要统一管理并发？常见问题与风险解析
核心概念解析：同步、异步、线程池、协程与并发模型选择
统一并发控制的四大策略：限流器、信号量、队列调度、自适应控制
代码实战：Python爬虫统一控制并发完整示例（基于asyncio + aiohttp）
性能调优与陷阱规避：内存泄漏、死锁、反爬对抗
Q&A 常见问题解答：关于并发控制的10个高频疑问

并发控制的必要性

场景：你写了一个爬虫，用for循环逐个请求1000个URL，跑完需要30分钟，后来你加了ThreadPoolExecutor(10)，3分钟就跑完了，但网站返回了大量503错误——你被反爬限制了。

如何编写统一控制爬虫并发

核心问题：无节制的并发会导致三个严重后果：

IP被封：单IP瞬时请求量超过网站阈值（如每秒10次）
内存溢出：未控制的并发可能同时加载数千个响应，内存被耗尽
目标服务过载：即使是合法爬虫，过度并发也可能视为DDoS攻击

统一控制的意义：不是简单地限制并发数，而是实现动态、可配置、资源感知的调度策略，让爬虫在“最快采集”与“友好访问”之间找到平衡。

核心概念解析：并发模型对比

模型	实现方式	适用场景	并发上限	资源消耗
多线程	threading / ThreadPoolExecutor	I/O密集型（如网络请求）	50-200	中（每线程约8MB栈）
多进程	multiprocessing / Pool	CPU密集型	取决于CPU核数	高
协程	asyncio + aiohttp	高I/O异步	1000-10000	极低
混合	协程+线程	同时处理I/O和阻塞操作	灵活	中低

统一控制的核心：无论使用哪种模型，都需要一个中央调度器来管理“正在运行的任务数”，当超过阈值时，新请求进入等待队列。

统一并发控制的四大策略

1 令牌桶限流器（Rate Limiter）

维护一个桶,令牌以固定速率产生（如每秒10个），请求消耗令牌，无令牌则等待。优点：精准控制QPS；缺点：无法处理瞬时波动。

2 信号量（Semaphore）

一种计数器,控制同时运行的协程/线程数，例如asyncio.Semaphore(5)表示最多5个并发。优点：实现简单；缺点：不控制速率，只控制峰值。

3 队列式调度（Priority Queue）

将所有URL放入队列,工作进程按优先级消费。优点：支持优先级、重试、去重；缺点：需要额外实现流量控制。

4 自适应控制（Adaptive Control）

根据响应时间、错误率动态调整并发数，当错误率>10%时，自动降低并发度。优点：鲁棒性强；缺点：实现复杂。

推荐组合：令牌桶 + 信号量 + 动态退避策略。

代码实战：统一并发控制爬虫（Python）

import asyncio
import aiohttp
from aiohttp import ClientSession
from typing import List, Dict
import time
class UnifiedCrawler:
    def __init__(self, max_concurrency: int = 10, rate_limit: int = 5):
        """
        :param max_concurrency: 最大并发任务数
        :param rate_limit: 每秒允许的请求数（令牌桶速率）
        """
        self.semaphore = asyncio.Semaphore(max_concurrency)  # 信号量控制并发数
        self.rate_limit = rate_limit
        self.last_request_time = 0.0
        self.rate_lock = asyncio.Lock()  # 保证速率控制线程安全
    async def rate_limiter(self):
        """令牌桶实现：保证每秒不超过rate_limit次请求"""
        async with self.rate_lock:
            now = time.monotonic()
            min_interval = 1.0 / self.rate_limit
            elapsed = now - self.last_request_time
            if elapsed < min_interval:
                await asyncio.sleep(min_interval - elapsed)
            self.last_request_time = time.monotonic()
    async def fetch(self, session: ClientSession, url: str) -> Dict:
        """单个任务：限速 -> 获取信号量 -> 发起请求"""
        await self.rate_limiter()
        async with self.semaphore:
            try:
                async with session.get(url, timeout=10) as resp:
                    if resp.status == 200:
                        data = await resp.text()
                        return {"url": url, "data": data[:200], "status": 200}
                    else:
                        return {"url": url, "error": f"HTTP {resp.status}"}
            except Exception as e:
                return {"url": url, "error": str(e)}
    async def run(self, urls: List[str]):
        """统一调度入口"""
        connector = aiohttp.TCPConnector(limit=0)  # 连接池无限制，由信号量控制
        async with ClientSession(connector=connector) as session:
            tasks = [self.fetch(session, url) for url in urls]
            results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
# 使用示例
async def main():
    urls = [f"https://httpbin.org/delay/{i%5}" for i in range(20)]  # 模拟20个不同延迟的URL
    crawler = UnifiedCrawler(max_concurrency=5, rate_limit=3)  # 并发5，每秒3次请求
    results = await crawler.run(urls)
    for res in results:
        print(res.get("url"), res.get("status", "error"))
if __name__ == "__main__":
    asyncio.run(main())

关键设计点：

semaphore控制最大并发数（防止瞬间资源耗尽）
rate_limiter以令牌桶方式控制QPS（防止被限流）
使用asyncio.Lock()保证速率计算的原子性
所有控制逻辑集中在fetch方法，外部调用只需关注任务提交

性能调优与陷阱规避

陷阱	表现	解决方案
连接池耗尽	出现`Connection pool is full`	改用`Semaphore`替代`TCPConnector(limit)`
协程泄漏	任务完成后内存不释放	使用`asyncio.gather(return_exceptions=True)`
死锁	程序永久挂起	避免信号量嵌套；使用`timeout`参数
反爬升级	延迟稳定但爬取变慢	加入随机延迟：`await asyncio.sleep(random.uniform(0.5, 1.5))`
监控缺失	不知道当前并发状态	添加`asyncio.Queue`记录待处理任务数，周期性log

自适应增强建议：引入response_time统计，如果平均响应时间>5秒，自动将max_concurrency降低50%；如果错误率>20%，暂停队列5分钟。

Q&A 常见问题解答

Q1：并发数设置多少最合适？
A：没有绝对标准，建议从max_concurrency=5, rate_limit=3开始，观察响应时间和错误率，逐步递增，最终值取决于目标网站的反爬阈值和你的网络带宽。

Q2：asyncio.Semaphore和aiohttp.TCPConnector(limit)有什么区别？
A：TCPConnector.limit控制的是底层TCP连接数，而Semaphore控制的是业务层任务数，建议统一使用Semaphore控制，TCPConnector保持默认（无限），避免双层控制导致的死锁。

Q3：如何实现分布式并发控制？
A：使用Redis作为中央调度器：