Python异步编程完全指南：async/await用法、asyncio实战、常见误区解析

异步编程是Python高级特性，能大幅提升IO密集型任务的性能，比如爬虫、API接口、文件操作等场景，比同步代码快好几倍。本文从基础概念到实战案例，带你彻底搞懂Python异步编程。

一、为什么要用异步编程

1.1 同步代码的问题

同步代码执行的时候，遇到IO操作（比如网络请求、文件读写、数据库查询）会阻塞等待，CPU闲着什么也不干，浪费时间。比如爬10个网页，每个请求等1秒，同步执行就要等10秒。

异步代码遇到IO操作的时候不会等待，会去执行其他任务，等IO完成了再回来继续处理，10个网页请求可能只需要1秒就能完成，效率提升10倍。

1.2 适用场景

适合IO密集型任务：

网络爬虫、API请求

数据库操作、文件读写

Web后端服务、WebSocket

消息队列、任务调度

不适合CPU密集型任务：

大量计算、科学计算

视频编码、图像处理

这些场景用多进程更合适

二、基础概念

2.1 核心概念

**协程（Coroutine）**：比线程更轻量级的执行单元，一个线程里可以有很多协程，切换开销很小。

**事件循环（Event Loop）**：管理协程的调度，协程遇到IO阻塞的时候，事件循环会把CPU时间分给其他协程。

**async/await**：Python3.5+引入的关键字，用来定义协程和等待异步操作。

**可等待对象（Awaitable）**：可以用await等待的对象，比如协程、Task、Future。

2.2 第一个异步程序

python

import asyncio
import time

# 定义协程函数，用async def
async def say_hello(name, delay):
    await asyncio.sleep(delay)  # 模拟IO操作，用await等待
    print(f"Hello {name}, 等待了{delay}秒")
    return f"结果：{name}"

async def main():
    start = time.time()
    
    # 创建任务，并发执行
    task1 = asyncio.create_task(say_hello("张三", 2))
    task2 = asyncio.create_task(say_hello("李四", 1))
    task3 = asyncio.create_task(say_hello("王五", 3))
    
    # 等待所有任务完成
    result1 = await task1
    result2 = await task2
    result3 = await task3
    
    print(result1, result2, result3)
    print(f"总耗时：{time.time() - start:.2f}秒")

# 运行协程
asyncio.run(main())

运行结果：

text

Hello 李四, 等待了1秒
Hello 张三, 等待了2秒
Hello 王五, 等待了3秒
结果：张三 结果：李四 结果：王五
总耗时：3.01秒

三个任务总共需要等待2+1+3=6秒，异步执行只花了3秒，因为是并发执行的。

三、asyncio核心用法

3.1 运行协程

Python3.7+推荐用`asyncio.run()`运行主协程：

python

async def main():
    await asyncio.sleep(1)
    print("完成")

asyncio.run(main())

旧版本可以用：

python

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

3.2 并发运行多个任务

方法1：用asyncio.gather()

python

async def main():
    start = time.time()
    # 并发运行多个协程
    results = await asyncio.gather(
        say_hello("张三", 2),
        say_hello("李四", 1),
        say_hello("王五", 3)
    )
    print(results)  # 按顺序返回结果
    print(f"总耗时：{time.time() - start:.2f}秒")

方法2：用asyncio.create_task()

python

async def main():
    # 创建任务，会立刻开始执行
    task1 = asyncio.create_task(say_hello("张三", 2))
    task2 = asyncio.create_task(say_hello("李四", 1))
    
    # 等待任务完成
    await task1
    await task2

方法3：按完成顺序获取结果

python

async def main():
    tasks = [
        asyncio.create_task(say_hello("张三", 2)),
        asyncio.create_task(say_hello("李四", 1)),
        asyncio.create_task(say_hello("王五", 3))
    ]
    
    # 哪个先完成就先处理哪个
    for task in asyncio.as_completed(tasks):
        result = await task
        print(f"收到结果：{result}")

3.3 超时处理

python

async def main():
    try:
        # 等待任务，超时1秒
        result = await asyncio.wait_for(say_hello("张三", 2), timeout=1)
        print(result)
    except asyncio.TimeoutError:
        print("任务超时")

3.4 任务取消

python

async def main():
    task = asyncio.create_task(say_hello("张三", 2))
    await asyncio.sleep(1)
    task.cancel()  # 取消任务
    try:
        await task
    except asyncio.CancelledError:
        print("任务被取消了")

四、常用异步库

很多常用库都有异步版本，性能更高：

4.1 异步HTTP请求：aiohttp

代替requests，异步发送HTTP请求：

bash

pip install aiohttp

python

import aiohttp

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            return await response.text(), response.status
    except Exception as e:
        return str(e), 500

async def main():
    urls = [
        "https://www.baidu.com",
        "https://www.qq.com",
        "https://www.163.com"
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for url, (content, status) in zip(urls, results):
            print(f"{url} 状态码：{status} 内容长度：{len(content)}")

asyncio.run(main())

并发请求多个网址，比requests快很多。

4.2 异步数据库：asyncpg/aiomysql

异步操作数据库，性能更高：

bash

pip install asyncpg  # PostgreSQL
pip install aiomysql  # MySQL

aiomysql示例：

python

import aiomysql

async def get_users():
    conn = await aiomysql.connect(
        host='localhost',
        port=3306,
        user='root',
        password='123456',
        db='test'
    )
    
    async with conn.cursor() as cur:
        await cur.execute("SELECT * FROM users")
        result = await cur.fetchall()
        print(result)
    
    conn.close()

asyncio.run(get_users())

4.3 异步文件操作：aiofiles

异步读写文件，避免阻塞：

bash

pip install aiofiles

python

import aiofiles

async def write_file():
    async with aiofiles.open('test.txt', 'w', encoding='utf-8') as f:
        await f.write("Hello Async\n")
        await f.write("异步写入文件")

async def read_file():
    async with aiofiles.open('test.txt', 'r', encoding='utf-8') as f:
        content = await f.read()
        print(content)

async def main():
    await write_file()
    await read_file()

五、实战案例：异步爬虫

爬取10个网页，对比同步和异步的速度：

5.1 同步版本（requests）

python

import requests
import time

def fetch_sync(url):
    response = requests.get(url)
    return len(response.text)

def main_sync():
    urls = [f"https://www.example.com" for _ in range(10)]
    start = time.time()
    for url in urls:
        length = fetch_sync(url)
        print(f"长度：{length}")
    print(f"同步总耗时：{time.time() - start:.2f}秒")

main_sync()
# 输出：同步总耗时：5.23秒

5.2 异步版本（aiohttp）

python

import aiohttp
import asyncio
import time

async def fetch_async(session, url):
    async with session.get(url) as response:
        return len(await response.text())

async def main_async():
    urls = [f"https://www.example.com" for _ in range(10)]
    start = time.time()
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_async(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        
        for length in results:
            print(f"长度：{length}")
    
    print(f"异步总耗时：{time.time() - start:.2f}秒")

asyncio.run(main_async())
# 输出：异步总耗时：0.68秒

异步比同步快了7倍多，爬的网页越多差距越大。

六、常见误区和注意事项

6.1 不要在异步代码里调用同步阻塞函数

错误示例：

python

async def bad_example():
    time.sleep(1)  # 同步阻塞，会卡住整个事件循环
    requests.get("https://www.example.com")  # 同步HTTP请求，阻塞

这些同步函数会阻塞整个事件循环，其他协程都无法执行，相当于白写异步了。

正确做法：

用对应的异步库，比如aiohttp代替requests，aiofiles代替open

必须用同步函数的话，用`asyncio.to_thread()`放到线程里执行：

python

async def good_example():
    await asyncio.to_thread(time.sleep, 1)  # 放到线程里执行，不阻塞事件循环

6.2 不要乱用async，所有函数都加async

异步不是银弹，CPU密集型任务用异步反而更慢，只有IO密集型任务才有提升。

6.3 协程不是多线程

协程是单线程执行的，同一时间只有一个协程在运行，只是切换很快，所以CPU密集型任务还是会卡住，要用多进程。

6.4 注意并发数量

不要同时开几万个协程，会占很多内存，合理控制并发数，可以用信号量限制：

python

semaphore = asyncio.Semaphore(10)  # 最多同时10个并发

async def limited_fetch(session, url):
    async with semaphore:
        return await fetch_url(session, url)

6.5 异常处理

异步任务的异常如果不捕获，会直接报错：

python

# 捕获异常
async def safe_fetch(session, url):
    try:
        return await fetch_url(session, url)
    except Exception as e:
        print(f"请求失败：{e}")
        return None

或者用`return_exceptions=True`让gather把异常当成结果返回：

python

results = await asyncio.gather(*tasks, return_exceptions=True)

七、高性能优化技巧

1. 连接池复用：aiohttp的ClientSession、数据库连接都要复用，不要每次请求都新建连接，性能提升很多。

2. 批量操作：数据库查询、网络请求尽量批量操作，减少IO次数。

3. 合理设置超时：所有异步操作都要加超时，避免某个任务卡住影响整个程序。

4. 用uvloop代替默认事件循环：uvloop是用C写的，性能比默认的asyncio事件循环快2-3倍：

bash

pip install uvloop

python

import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 然后正常运行asyncio.run()就行

Windows系统不支持uvloop，Linux和Mac可以用，性能提升明显。

八、异步框架推荐

**FastAPI**：现在最火的异步Web框架，性能高，开发快。

**Starlette**：轻量级异步Web框架，FastAPI就是基于它的。

**Sanic**：高性能异步Web框架，和Flask用法类似。

**Scrapy**：爬虫框架，内置异步支持，爬取效率很高。

**Celery 5+**：任务队列，支持异步任务执行。

异步编程能大幅提升IO密集型任务的性能，现在Python生态的异步支持也越来越完善，学会异步编程是Python后端开发的必备技能，赶紧用起来吧！

Python异步编程完全指南：async/await用法、asyncio实战、常见误区解析

Python异步编程完全指南：async/await用法、asyncio实战、常见误区解析

一、为什么要用异步编程

1.1 同步代码的问题

1.2 适用场景

二、基础概念

2.1 核心概念

2.2 第一个异步程序

三、asyncio核心用法

3.1 运行协程

3.2 并发运行多个任务

方法1：用asyncio.gather()

方法2：用asyncio.create_task()

方法3：按完成顺序获取结果

3.3 超时处理

3.4 任务取消

四、常用异步库

4.1 异步HTTP请求：aiohttp

4.2 异步数据库：asyncpg/aiomysql

4.3 异步文件操作：aiofiles

五、实战案例：异步爬虫

5.1 同步版本（requests）

5.2 异步版本（aiohttp）

六、常见误区和注意事项

6.1 不要在异步代码里调用同步阻塞函数

6.2 不要乱用async，所有函数都加async

6.3 协程不是多线程

6.4 注意并发数量

6.5 异常处理

七、高性能优化技巧

八、异步框架推荐

FastAPI入门到实战：快速搭建高性能API接口，比Flask快好几倍

Java 21新特性详解：虚拟线程、模式匹配、结构化并发，性能提升太明显了

相关文章

Python 自动化 Excel 综合实战案例

2026年Python自动化测试完全指南：8个实战框架教你从零基础到测试开发工程师

2026年Python GUI开发完全指南：8个实战项目教你快速打造桌面应用

2026年Python网络爬虫完全指南：10个实战案例教你合法抓取数据，新手也能快速上手

暂无评论