Python异步编程完全教程：asyncio/aiohttp核心用法与实战

异步编程是Python提升I/O密集型任务效率的核心技术，尤其适用于网络请求、文件读写、数据库交互等场景。

一、异步编程核心概念：同步vs异步

1. 同步编程（传统方式）

同步代码按顺序执行，一个任务未完成时，后续任务必须等待（“阻塞”）。例如：

import timedef sync_task(name, delay):print(f"任务{name}开始执行")time.sleep(delay)  # 模拟I/O阻塞（如网络请求）print(f"任务{name}执行完成")# 同步执行3个任务
start = time.time()
sync_task("A", 2)
sync_task("B", 1)
sync_task("C", 3)
print(f"总耗时：{time.time()-start:.2f}秒")

运行结果：

任务A开始执行
任务A执行完成
任务B开始执行
任务B执行完成
任务C开始执行
任务C执行完成
总耗时：6.00秒

同步执行的总耗时是所有任务耗时之和，效率极低。

2. 异步编程（非阻塞方式）

异步编程中，当某个任务遇到I/O阻塞时，程序不会等待，而是切换到其他任务执行，待阻塞任务完成后再回来继续处理。核心特点：

• 非阻塞：I/O等待期间不占用CPU；
• 单线程：异步任务在单线程内切换执行（区别于多线程）；
• 事件循环：核心调度器，负责管理异步任务的执行、切换。

二、asyncio核心基础：协程与事件循环

asyncio是Python内置的异步编程库，核心是协程（Coroutine） 和事件循环（Event Loop）。

1. 协程定义与基本语法

协程是异步任务的载体，通过async def定义，await触发阻塞并切换任务：

import asyncio
import time# 定义协程函数（必须用async def）
async def async_task(name, delay):print(f"任务{name}开始执行")await asyncio.sleep(delay)  # 异步睡眠（模拟I/O阻塞，不能用time.sleep）print(f"任务{name}执行完成")# 主协程（程序入口）
async def main():# 创建任务列表tasks = [async_task("A", 2),async_task("B", 1),async_task("C", 3)]# 并发执行所有任务await asyncio.gather(*tasks)# 启动事件循环（Python 3.7+简化写法）
start = time.time()
asyncio.run(main())
print(f"总耗时：{time.time()-start:.2f}秒")

运行结果：

任务A开始执行
任务B开始执行
任务C开始执行
任务B执行完成
任务A执行完成
任务C执行完成
总耗时：3.00秒

异步执行总耗时等于最长任务的耗时（3秒），效率提升一倍。

2. 核心语法解析

语法/函数	作用
async def 函数名()	定义协程函数，调用后返回协程对象（不会立即执行）
await 可等待对象	暂停当前协程，切换到事件循环执行其他任务，直到“可等待对象”完成后返回
asyncio.run()	创建事件循环→运行主协程→关闭事件循环（Python 3.7+）
asyncio.gather()	并发执行多个协程，等待所有协程完成后返回结果列表
asyncio.sleep()	异步睡眠（模拟I/O阻塞），必须用await调用（区别于time.sleep）

3. 任务（Task）：主动调度协程

asyncio.create_task()可将协程包装为“任务”，主动加入事件循环调度：

import asyncioasync def async_task(name, delay):print(f"任务{name}开始执行")await asyncio.sleep(delay)return f"任务{name}完成（耗时{delay}秒）"async def main():# 创建任务（立即加入事件循环）task1 = asyncio.create_task(async_task("A", 2))task2 = asyncio.create_task(async_task("B", 1))# 等待任务完成并获取返回值result1 = await task1result2 = await task2print(result1)print(result2)asyncio.run(main())

运行结果：

任务A开始执行
任务B开始执行
任务B完成（耗时1秒）
任务A完成（耗时2秒）

三、aiohttp：异步HTTP请求（实战核心）

asyncio仅处理基础异步逻辑，aiohttp是Python最常用的异步HTTP客户端/服务端库，专门解决同步requests库的性能瓶颈。

1. 环境准备

# 安装aiohttp
pip install aiohttp

2. 基础异步请求：单URL请求

import asyncio
import aiohttpasync def fetch_url(session, url):"""异步请求单个URL"""try:async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:# 获取响应状态码和内容status = response.statuscontent = await response.text()  # 文本内容（response.json()获取JSON）return {"url": url,"status": status,"content_length": len(content)}except Exception as e:return {"url": url,"error": str(e)}async def main():# 创建异步HTTP会话（复用连接，提升效率）async with aiohttp.ClientSession() as session:result = await fetch_url(session, "https://www.baidu.com")print(f"请求结果：{result}")asyncio.run(main())

运行结果：

请求结果：{'url': 'https://www.baidu.com', 'status': 200, 'content_length': 2443}

3. 批量异步请求：多URL并发

import asyncio
import aiohttp
import timeasync def fetch_url(session, url):"""异步请求单个URL"""try:async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response:await response.text()return f"{url} → 状态码：{response.status}"except Exception as e:return f"{url} → 错误：{str(e)}"async def main():# 待请求的URL列表urls = ["https://www.baidu.com","https://www.taobao.com","https://www.jd.com","https://www.163.com","https://www.github.com"]# 创建会话并批量请求async with aiohttp.ClientSession() as session:# 创建任务列表tasks = [fetch_url(session, url) for url in urls]# 并发执行所有任务results = await asyncio.gather(*tasks)# 打印结果for res in results:print(res)# 计时对比同步请求
start = time.time()
asyncio.run(main())
print(f"总耗时：{time.time()-start:.2f}秒")

运行结果：

https://www.baidu.com → 状态码：200
https://www.taobao.com → 状态码：200
https://www.jd.com → 状态码：200
https://www.163.com → 状态码：200
https://www.github.com → 状态码：200
总耗时：0.85秒

（同步requests请求相同URL约需3-5秒，异步效率提升4-6倍）

4. 进阶：限制并发数（避免被封IP）

批量请求时直接并发所有任务可能触发目标服务器反爬，需限制并发数：

import asyncio
import aiohttp
import time# 限制最大并发数为3
MAX_CONCURRENT = 3async def fetch_url(session, url, semaphore):"""带并发限制的异步请求"""# 信号量控制并发数async with semaphore:try:async with session.get(url, timeout=10) as response:await response.text()return f"{url} → 成功"except Exception as e:return f"{url} → 失败：{str(e)}"async def main():urls = [f"https://httpbin.org/get?num={i}" for i in range(10)]  # 10个测试URLsemaphore = asyncio.Semaphore(MAX_CONCURRENT)  # 信号量async with aiohttp.ClientSession() as session:tasks = [fetch_url(session, url, semaphore) for url in urls]results = await asyncio.gather(*tasks)for res in results:print(res)start = time.time()
asyncio.run(main())
print(f"总耗时：{time.time()-start:.2f}秒")

核心原理：asyncio.Semaphore(3)限制同时执行的任务数为3，避免一次性发起大量请求。

四、异步编程常见问题与避坑指南

1. 避免在异步代码中使用同步阻塞函数

time.sleep()、requests.get()、pymysql.connect()等同步阻塞函数会卡住整个事件循环，必须替换为异步版本：

同步操作	异步替代方案
time.sleep(delay)	await asyncio.sleep(delay)
requests.get(url)	aiohttp.ClientSession().get(url)
pymysql（MySQL）	aiomysql
redis（Redis）	aioredis

2. 协程函数调用必须加await

直接调用协程函数不会执行，仅返回协程对象：

async def test():print("协程执行")# 错误：仅返回协程对象，不执行
test()
# 正确：await触发执行
await test()

3. 异常处理：捕获异步任务的错误

import asyncioasync def error_task():raise ValueError("异步任务出错了")async def main():try:await error_task()except ValueError as e:print(f"捕获到错误：{e}")asyncio.run(main())

4. 异步代码无法在同步函数中直接运行

如需在同步函数中调用异步代码，需通过asyncio.run()或事件循环：

import asyncioasync def async_func():return "异步函数返回值"# 同步函数中调用异步代码
def sync_func():result = asyncio.run(async_func())print(result)sync_func()

五、实战案例：异步爬虫（爬取网页标题）

以下案例实现“批量爬取URL标题 → 保存到CSV文件”的完整异步流程：

import asyncio
import aiohttp
import csv
from bs4 import BeautifulSoup  # 需安装：pip install beautifulsoup4# 限制并发数
MAX_CONCURRENT = 5async def fetch_title(session, url, semaphore):"""异步爬取网页标题"""async with semaphore:try:async with session.get(url, timeout=10) as response:if response.status != 200:return {"url": url, "title": "请求失败", "status": response.status}html = await response.text()soup = BeautifulSoup(html, "html.parser")title = soup.title.string.strip() if soup.title else "无标题"return {"url": url, "title": title, "status": response.status}except Exception as e:return {"url": url, "title": f"错误：{str(e)}", "status": "异常"}async def main():# 待爬取的URL列表urls = ["xxxxx",]# 初始化信号量和会话semaphore = asyncio.Semaphore(MAX_CONCURRENT)async with aiohttp.ClientSession() as session:# 创建任务并执行tasks = [fetch_title(session, url, semaphore) for url in urls]results = await asyncio.gather(*tasks)# 保存到CSV文件with open("url_titles.csv", "w", newline="", encoding="utf-8") as f:writer = csv.DictWriter(f, fieldnames=["url", "title", "status"])writer.writeheader()writer.writerows(results)print("爬取完成！结果已保存到url_titles.csv")# 启动程序
if __name__ == "__main__":asyncio.run(main())