如何高效批量检测百万网站存活状态？Python grequests方案优化需求

针对百万级网站检测场景，现有grequests+ThreadPoolExecutor的混合并发模式存在资源利用不充分、效率偏低的问题，以下是具体优化建议：

一、重构并发模型：改用AsyncIO+AioHTTP #

放弃线程池与grequests的混合架构，直接使用asyncio+aiohttp纯异步方案。IO密集型任务中，异步协程的开销远低于线程，能支持数千级并发，彻底避免线程切换带来的阻塞问题。

二、核心请求参数调优 #

• 缩短超时时间：将请求超时从10秒降至3-5秒，快速过滤无响应站点，减少无效等待。
• 复用HTTP连接：依赖aiohttp.ClientSession的连接池功能（默认开启），避免重复建立TCP连接，大幅提升请求效率。
• 随机化请求头：维护一个常用User-Agent池，每次请求随机选取，降低被目标站点识别为爬虫的概率。

三、代理池集成与轮换机制 #

• 队列化代理管理：将可用代理存入异步队列，每次请求从队列取代理，使用后放回（或标记失效），实现自动轮换。
• 代理失败重试：当请求因代理失效/被封禁失败时，自动重试其他代理，避免任务中断。
• 定期校验代理可用性：单独启动协程定期验证代理存活状态，剔除无效代理，保证请求成功率。

四、分块与并发数优化 #

• 增大单批次并发量：根据系统资源和代理质量，将单批次并发数从100提升至500-1000（asyncio轻松支持数千级并发），减少分块处理的额外开销。
• 流式读取URL文件：不要一次性加载百万级URL到内存，改用生成器逐行读取，降低内存占用，避免OOM问题。

五、数据处理优化 #

• 预编译正则表达式：将匹配规则移至函数外部预编译，避免每次请求重复编译，节省CPU资源：

  CHECK_PATTERN = re.compile(r'(pat1|pat2)', re.IGNORECASE)
  

• 异步结果处理：用异步队列收集请求结果，单独启动协程负责写入文件/数据库，避免在请求回调中同步处理导致的阻塞。

六、优化后代码示例 #

import asyncio
import re
import random
from aiohttp import ClientSession, ClientProxyConnectionError, ClientTimeout
from asyncio import Queue

# 预编译正则匹配规则
CHECK_PATTERN = re.compile(r'(pat1|pat2)', re.IGNORECASE)

# 常用User-Agent池
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.3.1 Safari/605.1.15",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36"
]

# 代理队列（实际需从代理池加载）
proxy_queue = Queue()

async def verify_proxy(proxy):
    # 简单校验代理可用性
    try:
        async with ClientSession(timeout=ClientTimeout(total=3)) as session:
            async with session.get('http://httpbin.org/ip', proxy=proxy):
                return True
    except:
        return False

async def load_valid_proxies():
    # 加载并验证代理，存入队列
    raw_proxies = ["http://proxy1:port", "http://proxy2:port"]  # 替换为实际代理来源
    for proxy in raw_proxies:
        if await verify_proxy(proxy):
            await proxy_queue.put(proxy)

async def fetch_single_site(session, url):
    try:
        proxy = await proxy_queue.get()
        headers = {"User-Agent": random.choice(USER_AGENTS)}
        async with session.get(url, headers=headers, proxy=proxy, timeout=ClientTimeout(total=5)) as response:
            if response.status == 200:
                text = await response.text()
                match = CHECK_PATTERN.search(text)
                if match:
                    return match.group(1)
            # 代理可用，放回队列循环使用
            await proxy_queue.put(proxy)
    except (ClientProxyConnectionError, asyncio.TimeoutError):
        # 代理失效，直接丢弃
        pass
    return None

async def process_url_batch(urls):
    async with ClientSession() as session:
        tasks = [fetch_single_site(session, url.split(' ')[0].strip()) for url in urls]
        results = await asyncio.gather(*tasks)
        for site in results:
            if site:
                print(f'Site {site}')

async def main(file_path, batch_size=500):
    await load_valid_proxies()
    print('Start check!')
    start_time = asyncio.get_event_loop().time()
    
    # 流式读取文件，避免内存过载
    with open(file_path, 'r') as f:
        current_batch = []
        for line in f:
            current_batch.append(line.strip())
            if len(current_batch) >= batch_size:
                await process_url_batch(current_batch)
                current_batch = []
        # 处理剩余URL
        if current_batch:
            await process_url_batch(current_batch)
    
    print(f'Total time: {asyncio.get_event_loop().time() - start_time:.2f} seconds')

if __name__ == "__main__":
    asyncio.run(main('sites.txt'))

额外建议 #

• 监控系统资源：用psutil工具监控CPU、内存和带宽，避免并发过高导致系统过载。
• 分布式扩展：若单机器性能达到瓶颈，可将URL分片分发到多台机器并行处理，进一步压缩总耗时。

内容的提问来源于stack exchange，提问作者Anton