基于aiohttp按域名限制每秒请求数的并行爬虫开发需求

嘿，这个需求我太熟悉了——做多域名并行爬虫时，既要控住单个域名的请求频率防反爬，又要把整体的并发能力拉满，确实是个挺典型的痛点。我来给你分享几个实际项目里跑通的方案，都是围绕「按域名隔离限流」这个核心来的：

既然只关心单个域名的QPS，那最直接的做法就是给每个域名单独分配限流规则，互相不干扰。这样一来，只要每个域名的请求速率不超限，全局的连接数和总QPS就能自然达到最大值（取决于你的网络带宽、硬件资源和目标域名的响应速度）。

1. 令牌桶：最可靠的单域名QPS限流实现 #

令牌桶是限流领域的经典方案，非常适合这种按主体（这里是域名）隔离的场景。每个域名对应一个令牌桶，每秒生成固定数量的令牌（等于你设定的QPS上限），每次请求必须先拿到令牌才能发送。

给你一段Python风格的伪代码参考：

from collections import defaultdict
import time
from threading import Lock

class DomainTokenBucket:
    def __init__(self, max_qps):
        self.max_qps = max_qps
        # 用defaultdict自动为新域名创建令牌桶
        self.buckets = defaultdict(lambda: {
            "available_tokens": max_qps,
            "last_refill_time": time.monotonic()  # 用单调时钟避免时间漂移
        })
        self.lock = Lock()  # 保证多线程/协程下的操作安全

    def try_acquire(self, domain):
        with self.lock:
            bucket = self.buckets[domain]
            now = time.monotonic()
            
            # 计算自上次填充以来应该补充的令牌数
            time_elapsed = now - bucket["last_refill_time"]
            tokens_to_add = time_elapsed * self.max_qps
            # 令牌数不能超过桶的上限
            bucket["available_tokens"] = min(self.max_qps, bucket["available_tokens"] + tokens_to_add)
            bucket["last_refill_time"] = now

            # 尝试获取令牌
            if bucket["available_tokens"] >= 1:
                bucket["available_tokens"] -= 1
                return True
            return False

这个实现的关键优势：

• 自动适配新域名，不用提前配置
• 用time.monotonic()避免系统时间调整导致的令牌计算错误
• 加锁保证线程/协程安全，不会出现超发请求的情况

2. 结合异步IO最大化并发能力 #

如果你的爬虫用异步框架（比如aiohttp、httpx的异步模式），可以把令牌桶和异步等待结合起来，这样单个域名的限流排队不会阻塞其他域名的请求，整体并发效率能拉到最高。

举个异步请求的示例：

import asyncio
from aiohttp import ClientSession

async def fetch_with_rate_limit(session, limiter, url):
    domain = url.split("//")[1].split("/")[0]
    # 循环等待直到拿到令牌
    while not limiter.try_acquire(domain):
        await asyncio.sleep(0.01)  # 短时间休眠，避免空转浪费资源
    
    async with session.get(url) as resp:
        return await resp.text()

async def main(target_urls):
    # 设定每个域名每秒最多5次请求
    rate_limiter = DomainTokenBucket(max_qps=5)
    # 用连接池复用TCP连接，提升效率
    connector = aiohttp.TCPConnector(limit=None)  # 全局不限制连接数，交给域名限流控
    async with ClientSession(connector=connector) as session:
        # 批量创建异步任务
        tasks = [fetch_with_rate_limit(session, rate_limiter, url) for url in target_urls]
        results = await asyncio.gather(*tasks)
        # 这里处理爬取结果
        print(f"完成{len(results)}个请求")

if __name__ == "__main__":
    asyncio.run(main(["https://domain1.com/page1", "https://domain2.com/page1", ...]))

这里特意把TCP连接池的limit设为None，就是为了让全局连接数不受限，完全靠每个域名的QPS限制来控制请求频率，最大化并发能力。

3. 实用优化点 #

• 请求队列化：如果某个域名的请求突然爆发，可以给每个域名加一个请求队列，把暂时拿不到令牌的请求存起来，按顺序处理，避免循环等待的空转开销。
• 过期桶清理：如果某个域名长时间没有请求，可以定期清理对应的令牌桶，节省内存资源。
• 动态调整QPS：如果需要应对不同域名的反爬策略，可以给DomainTokenBucket加个方法，动态修改单个域名的QPS上限，不用重启爬虫。

4. 避坑提醒 #

• 不要全局限流：千万别为了省事搞全局QPS限制，那样会直接浪费你的并发能力——比如域名A每秒只能处理5次请求，但域名B能扛100次，全局限流会把域名B的能力也限制住。
• 连接池复用很重要：复用TCP连接能大幅减少握手开销，提升整体请求速度，尤其是对同一个域名的多次请求，一定要开启客户端的连接池功能。
• 避免时钟问题：绝对不要用系统的 wall time（比如time.time()）来计算令牌填充，万一系统时间被调整（比如NTP同步），会导致令牌计算完全混乱，一定要用单调递增的时钟（比如Python的time.monotonic()）。

这种按域名隔离令牌桶的方案，既能精准控制单个域名的请求频率，又能让全局并发能力最大化，是我做爬虫项目时最常用的限流方案，亲测稳定可靠。

内容的提问来源于stack exchange，提问作者J. Taylor