Python脚本如何批量发起协程请求

wen python案例 2026-06-27 28

Python脚本如何批量发起协程请求（实战指南）

📖 目录导读

为什么需要批量协程请求？
协程基础：async/await 与事件循环
三大关键库：aiohttp + asyncio + tqdm
批量请求的四种模式
- 1 简单批量（gather）
- 2 带限速的信号量模式
- 3 带重试与超时的安全模式
- 4 异步生成器流式处理
实战代码模板
常见问题与问答（FAQ）
性能对比：协程 vs 多线程 vs 多进程
总结与最佳实践

为什么需要批量协程请求？

当我们需要从数百个API接口抓取数据、爬取大量网页或者发送批量通知时，传统的串行请求（requests库循环）极其低效，假设每个请求耗时1秒，1000个请求就是16分钟以上，而使用Python协程，可以在单线程内并发执行I/O操作，将总时间压缩到1～2秒（取决于网络延迟和并发数）。

Python脚本如何批量发起协程请求

核心优势：

内存消耗远低于多线程（每个协程仅需几KB）
不依赖操作系统线程调度，避免GIL锁竞争
天然支持高并发（轻松上千个并发连接）

协程基础：async/await 与事件循环

Python的asyncio库是协程的基石,理解以下三个概念即可上手：

async def：定义一个协程函数（可暂停执行的函数）
await：挂起当前协程，等待另一个协程完成
事件循环：调度所有协程的运行时环境

import asyncio
async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()
async def main():
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, 'https://example.com')
        print(html)
asyncio.run(main())

⚠️ 注意：asyncio.run() 会创建新的事件循环并自动关闭,是推荐的入口方式。

三大关键库：aiohttp + asyncio + tqdm

库名	作用	安装命令
`aiohttp`	异步HTTP客户端（替代requests）	`pip install aiohttp`
`asyncio`	Python内置协程库	无需安装
`tqdm`	进度条显示	`pip install tqdm`

组合使用：aiohttp提供异步网络请求，asyncio管理并发调度，tqdm让你实时看到处理进度。

批量请求的四种模式

1 简单批量（gather）

适合并发数不多（<100）且不关心顺序的场景：

async def batch_simple(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(fetch(session, url)) for url in urls]
        results = await asyncio.gather(*tasks)
    return results  # 返回顺序与urls顺序一致

缺点：如果某个请求卡死，整个gather会一直等待。

2 带限速的信号量模式

防止向目标服务器发送过多请求导致被屏蔽：

semaphore = asyncio.Semaphore(50)  # 最大并发50
async def fetch_with_limit(session, url):
    async with semaphore:
        return await fetch(session, url)
async def batch_with_limit(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url) for url in urls]
        return await asyncio.gather(*tasks)

3 带重试与超时的安全模式

网络请求不稳定,建议加入重试逻辑：

async def safe_fetch(session, url, retries=3):
    for attempt in range(retries):
        try:
            async with session.get(url, timeout=10) as resp:
                return await resp.text()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            if attempt == retries - 1:
                raise
            await asyncio.sleep(1 * (attempt + 1))  # 指数退避

4 异步生成器流式处理

适合URL数量极大（10万+）,避免内存暴涨：

async def stream_process(urls):
    async with aiohttp.ClientSession() as session:
        for url in urls:
            text = await safe_fetch(session, url)
            # 立即处理单个结果，不等待全部完成
            yield text

调用方使用async for逐条消费,配合进度条使用极佳。

实战代码模板（带进度条）

以下是一个可直接使用的完整脚本，包含了限速、重试、超时、进度条：

import asyncio
import aiohttp
from tqdm.asyncio import tqdm
semaphore = asyncio.Semaphore(100)  # 并发数
TIMEOUT = aiohttp.ClientTimeout(total=30)
async def fetch_one(session, url):
    async with semaphore:
        for retry in range(3):
            try:
                async with session.get(url, timeout=TIMEOUT) as resp:
                    return await resp.text()
            except Exception as e:
                if retry == 2:
                    raise
                await asyncio.sleep(2)
async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        # tqdm自动显示进度
        return await tqdm.gather(*tasks, desc="正在请求")
def batch_request(urls):
    return asyncio.run(fetch_all(urls))
# 使用示例
if __name__ == "__main__":
    test_urls = ["https://httpbin.org/delay/1"] * 200  # 测试200个延迟1秒的请求
    results = batch_request(test_urls)
    print(f"完成 {len(results)} 个请求")

性能实测：200个延迟1秒的请求，串行需200秒，协程并发仅需约3秒（并发数100）。

常见问题与问答（FAQ）

Q1：协程比多线程快多少？
A：在纯I/O场景下，协程的上下文切换开销约为多线程的1/50（微秒级 vs 毫秒级），实测1000个请求，协程耗时约2秒，多线程通常需要5～8秒（受线程切换和GIL影响）。

Q2：并发数设置多少合适？
A：取决于目标服务器和网络环境，一般建议50～200，过高可能导致本地端口耗尽（每次请求占用一个临时端口）或触发服务器限流。

Q3：遇到SSL证书错误怎么办？
A：在aiohttp.ClientSession中设置connector=aiohttp.TCPConnector(ssl=False),但仅限测试环境。

Q4：如何处理返回结果顺序？
A：asyncio.gather返回顺序与输入列表顺序一致；若使用tqdm.gather同样保持顺序，如果不需要顺序，可改用wait或as_completed。

Q5：能否在Django/Flask中使用协程？
A：可以，但需注意框架是否支持异步，Django 3.1+支持异步视图，Flask需使用asgiref包装，建议在独立的异步脚本中执行批量请求,然后存入数据库供Web应用使用。

性能对比：协程 vs 多线程 vs 多进程

特性	协程 (asyncio)	多线程 (ThreadPoolExecutor)	多进程 (ProcessPoolExecutor)
上下文切换开销	极小（微秒级）	中等（毫秒级）	大（进程创建）
受GIL影响	否（仅在await时释放）	是（CPU密集任务受影响）	否（每个进程独立）
最大并发数	10万+	1000左右（受线程栈限制）	取决于CPU核数
适用场景	I/O密集型	I/O密集型+少量CPU操作	CPU密集型
代码复杂度	中等（需要异步库）	低（threading简单）	低

对于纯HTTP请求（I/O密集型）,协程是绝对首选。

总结与最佳实践

永远使用连接池：aiohttp.ClientSession会自动复用TCP连接，大幅提升性能。
添加超时：ClientTimeout防止请求卡死。
控制并发：用asyncio.Semaphore限制并发数。
处理异常：重试机制+错误日志，避免单个失败导致整体中断。
进度可视化：tqdm让长时间运行的任务可观察。
资源清理：使用async with确保session正确关闭。

进阶建议：

结合orjson（异步JSON解析）进一步提升速度
使用aiodns加速DNS解析
对于超大规模任务，考虑asyncio.Queue实现生产者-消费者模式

现在你已经掌握了用Python批量发起协程请求的核心技能，从几十个URL到百万级数据集，协程都能以优雅且高效的方式助你完成任务，立即尝试修改模板，替换为你的目标地址,感受并发魔力的速度吧！