Python脚本如何批量执行异步任务：从入门到高效实战
目录导读
- 为什么需要批量异步任务？
- 核心工具：asyncio与aiohttp/aiomysql
- 三步搭建异步任务框架
- 常见陷阱与性能调优
- FAQ：新手最常问的5个问题
- 实战案例：批量抓取1000个网页
为什么需要批量异步任务？
当业务需要处理大量I/O密集型操作（如网络请求、数据库读写、文件下载）时，传统的同步方式会因等待I/O而浪费CPU资源，用requests库抓取100个网页，同步代码需要按顺序等待每个请求完成，总耗时≈单个请求耗时×100，而异步任务允许在等待某个请求时切换执行其他任务，理论上总耗时≈最慢的单个请求耗时。
适用场景：
- 爬虫批量采集数据
- 批量发送邮件/短信
- 微服务间批量API调用
- 数据库批量读写（需配合连接池）
不适用场景：
- CPU密集型计算（如图像渲染、数据分析）——此时应使用
multiprocessing。
核心工具：asyncio与aiohttp/aiomysql
1 Python原生异步库：asyncio
asyncio是Python 3.4+内置的异步I/O框架，通过async/await语法定义协程，核心概念：
- 事件循环：调度协程的核心，负责监听I/O事件并回调。
- 协程：通过
async def定义的函数，需用await挂起。 - 任务：通过
asyncio.create_task()将协程包装为可并发执行的任务。
import asyncio
async def fetch(url):
# 模拟异步请求
await asyncio.sleep(1)
return f"Fetched {url}"
async def main():
tasks = [asyncio.create_task(fetch(f"page_{i}")) for i in range(10)]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())
2 第三方异步库：aiohttp（HTTP）、aiomysql（数据库）
- aiohttp：异步HTTP客户端/服务器库，支持Session复用、超时控制。
- aiomysql：异步MySQL驱动，需配合
asyncio和连接池使用。
安装：
pip install aiohttp aiomysql
三步搭建异步任务框架
步骤1：定义异步任务函数
async def async_task(item):
async with aiohttp.ClientSession() as session:
async with session.get(f"https://api.example.com/{item}") as resp:
return await resp.json()
步骤2：批量创建任务并调度
async def main():
items = range(1000) # 假设有1000个任务
tasks = [asyncio.create_task(async_task(i)) for i in items]
results = await asyncio.gather(*tasks, return_exceptions=True)
# return_exceptions=True会捕获异常而不中断
步骤3：限流与资源控制
使用asyncio.Semaphore控制并发数，防止资源耗尽：
sem = asyncio.Semaphore(50) # 最多50个并发
async def async_task_with_limit(item):
async with sem:
return await async_task(item)
常见陷阱与性能调优
1 陷阱：阻塞任务混入
在异步协程中调用time.sleep()或同步库（如requests）会阻塞事件循环，导致所有任务等待。
解决方案： 使用asyncio.sleep()或asyncio.to_thread()包装同步代码。
2 陷阱：连接数超限
批量请求时,ClientSession默认连接数为100，超过会排队。
调优： session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=500))
3 陷阱：内存泄漏
未正确关闭Session或文件句柄会导致内存占用持续增长。
方案： 使用async with自动释放资源，或显式调用session.close()。
4 性能调优要点
- 复用Session：每个任务创建Session是低效的，应在主函数中创建全局Session。
- 控制并发数：根据目标服务器能力调整Semaphore值，通常50-200之间。
- 超时设置：为每个请求设置超时，避免死等：
session.get(url, timeout=10)
FAQ：新手最常问的5个问题
Q1：异步编程能否用多线程替代？
A：多线程适合I/O不密集或需要共享状态的场景，但Python的GIL会使多线程在CPU密集型任务中反而更慢，异步是单线程协作式调度，更轻量。
Q2：异步任务中如何打印日志？
A：使用logging模块时，注意日志写入文件是I/O操作，建议用logging.handlers.QueueHandler异步写入，或使用aiologger库。
Q3：失败任务如何重试？
A：在async_task内用循环实现重试，配合tenacity库简化：
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=5))
async def async_task_retry(item):
# ...
Q4：为何我的异步代码比同步还慢？
A：检查是否混入了同步阻塞操作，或并发数过低。asyncio适用于I/O密集型，若任务本身计算量大（如正则匹配），应考虑concurrent.futures。
Q5：异步如何与数据库交互？
A：使用aiomysql或asyncpg，避免用pymysql，对于ORM，选择sqlalchemy的异步模式（1.4+版本支持）。
实战案例：批量抓取1000个网页
目标：抓取https://httpbin.org/delay/1（模拟延迟1秒的页面）1000次，并统计总耗时。
import asyncio
import aiohttp
import time
async def fetch_page(session, url):
async with session.get(url, timeout=10) as resp:
return await resp.text()
async def main():
urls = [f"https://httpbin.org/delay/1?n={i}" for i in range(1000)]
conn = aiohttp.TCPConnector(limit=100)
sem = asyncio.Semaphore(80)
async with aiohttp.ClientSession(connector=conn) as session:
async def limited_fetch(url):
async with sem:
return await fetch_page(session, url)
tasks = [asyncio.create_task(limited_fetch(url)) for url in urls]
start = time.time()
results = await asyncio.gather(*tasks, return_exceptions=True)
end = time.time()
successes = [r for r in results if not isinstance(r, Exception)]
print(f"成功: {len(successes)}, 失败: {len(results)-len(successes)}")
print(f"总耗时: {end-start:.2f}秒")
asyncio.run(main())
执行效果：
- 若同步执行（单线程串行），1000个请求各延迟1秒，总耗时≈1000秒。
- 异步并发80个任务,总耗时约12-15秒（网络延迟+服务器处理时间）。
优化空间：
- 预解析URL,减少字符串拼接。
- 使用
asyncio.as_completed及时处理已完成的响应。
批量异步任务的核心是用协程消除I/O等待，通过asyncio+第三方异步库实现高效并发，关键在于：明确任务类型、控制并发数、避免阻塞、妥善处理异常，掌握本文的技术点后，您可以将脚本轻松改造，应对每日百万级API调用或数据库批量写入场景。
