Home

主题

同步 vs 异步：为什么需要异步 ​

传统的同步编程中，一个 I/O 操作（如网络请求、文件读取）会阻塞整个线程，直到操作完成才能继续执行下一行代码：

python
# 同步代码 - 假设每个请求耗时 1 秒
import time

def fetch_user(id):
    time.sleep(1)  # 模拟网络请求
    return {"id": id, "name": f"User_{id}"}

start = time.time()
results = [fetch_user(i) for i in range(3)]  # 顺序执行，总耗时 3 秒
print(f"耗时: {time.time() - start:.2f}s")

异步编程通过事件循环和协程，在等待 I/O 时切换到其他任务，实现并发，大幅提升效率：

python
import asyncio
import time

async def fetch_user(id):
    await asyncio.sleep(1)  # 模拟异步网络请求
    return {"id": id, "name": f"User_{id}"}

async def main():
    start = time.time()
    # asyncio.gather 并发执行所有协程
    results = await asyncio.gather(*[fetch_user(i) for i in range(3)])
    print(f"耗时: {time.time() - start:.2f}s")
    return results

asyncio.run(main())  # 耗时约 1 秒，而非 3 秒

async / await 基础 ​

定义协程函数 ​

async def 定义的函数称为协程函数，调用它返回一个协程对象，必须用 await 或 asyncio.run() 执行：

python
async def greet(name):
    return f"Hello, {name}!"

# 三种调用方式的区别
result = greet("World")           # 返回协程对象，不会执行
result = asyncio.run(greet("World"))  # 完整运行并获取结果
# await 只在协程函数或事件循环内使用

await 关键字 ​

await 暂停当前协程，等待另一个协程/可等待对象完成，然后继续执行：

python
async def step1():
    print("Step 1 开始")
    await asyncio.sleep(0.5)
    print("Step 1 完成")

async def step2():
    print("Step 2 开始")
    await asyncio.sleep(0.5)
    print("Step 2 完成")

async def main():
    await step1()  # 顺序执行
    await step2()

asyncio.run(main())

并发执行：gather 和 Task ​

python
async def main():
    # gather: 并发执行多个协程
    results = await asyncio.gather(
        step1(),
        step2(),
    )
    print("全部完成")

asyncio.run(main())
# 输出: Step 1 开始 / Step 2 开始 (几乎同时)
#       Step 1 完成 / Step 2 完成 (几乎同时)

asyncio.create_task() 将协程包装为任务，立即调度执行：

python
async def main():
    # 创建任务（立即开始执行）
    task1 = asyncio.create_task(step1())
    task2 = asyncio.create_task(step2())

    # 等待所有任务完成
    await task1
    await task2

asyncio.run(main())

asyncio 常用 API ​

sleep ​

asyncio.sleep() 是异步版本的 time.sleep()，不阻塞事件循环：

python
async def demo():
    print("开始等待...")
    await asyncio.sleep(3)  # 3 秒内可以执行其他协程
    print("等待结束")

异步迭代 ​

python
async def async_range(n):
    for i in range(n):
        yield i
        await asyncio.sleep(0.1)

async def main():
    async for i in async_range(5):
        print(i)

超时控制 ​

python
import asyncio

async def slow_task():
    await asyncio.sleep(10)
    return "完成"

async def main():
    try:
        # 最多等待 2 秒
        result = await asyncio.wait_for(slow_task(), timeout=2)
    except asyncio.TimeoutError:
        print("任务超时")

asyncio.run(main())

并行等待 ​

python
async def fetch(url):
    await asyncio.sleep(1)  # 模拟网络请求
    return url

async def main():
    urls = ['a.com', 'b.com', 'c.com']

    # wait_for: 带超时的并行等待
    # as_completed: 每完成一个处理一个
    tasks = [fetch(url) for url in urls]

    # gather: 全部完成后一次性返回结果
    results = await asyncio.gather(*tasks)
    print(results)

异步上下文管理器和生成器 ​

异步上下文管理器 ​

实现异步的 __aenter__ 和 __aexit__：

python
class AsyncDB:
    async def __aenter__(self):
        print("连接数据库...")
        await asyncio.sleep(0.1)
        self.conn = "数据库连接对象"
        return self.conn

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("关闭数据库连接...")
        await asyncio.sleep(0.1)

async def main():
    async with AsyncDB() as conn:
        print(f"使用 {conn} 执行查询...")

asyncio.run(main())

使用 @contextlib.asynccontextmanager：

python
import asyncio
import contextlib

@contextlib.asynccontextmanager
async def async_transaction():
    print("开启事务")
    yield "tx"
    print("提交事务")

async def main():
    async with async_transaction() as tx:
        print(f"执行事务: {tx}")

asyncio.run(main())

异步生成器 ​

async def 中使用 yield 产生异步生成器：

python
async def async_data_stream():
    for i in range(5):
        await asyncio.sleep(0.5)
        yield i

async def main():
    async for item in async_data_stream():
        print(item)

asyncio.run(main())

实战：异步 HTTP 请求 ​

aiohttp 是 Python 中最流行的异步 HTTP 客户端库：

python
import asyncio
import aiohttp

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

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
    ]
    import time
    start = time.time()
    results = await fetch_all(urls)
    print(f"并发请求 3 个接口，总耗时: {time.time() - start:.2f}s")

asyncio.run(main())  # 约 2 秒，而非 4 秒

安装方式：pip install aiohttp

异步与同步的桥接 ​

run_in_executor - 在线程池中运行同步代码 ​

python
import asyncio
from concurrent.futures import ThreadPoolExecutor

def blocking_io():
    import time
    time.sleep(2)
    return "同步操作完成"

async def main():
    loop = asyncio.get_event_loop()
    # 在线程池中执行同步阻塞代码
    result = await loop.run_in_executor(None, blocking_io)
    print(result)

asyncio.run(main())

asyncio.to_thread - 更简洁的桥接方式（Python 3.9+） ​

python
async def main():
    result = await asyncio.to_thread(blocking_io)
    print(result)

异步编程的注意事项 ​

  1. 不要在同步代码中直接调用 async 函数：

    python
    async def async_func():
    return "result"

# 错误：返回协程对象，不会执行
result = async_func()

# 正确：使用 asyncio.run 或 await
result = asyncio.run(async_func())

  2. 区分 CPU 密集型和 I/O 密集型：异步适合 I/O 密集型（网络、文件），CPU 密集型建议用 run_in_executor 或多进程。

  3. 避免在循环中顺序 await：将独立任务收集后用 gather 并发执行。

小结 ​

  • async def 定义协程函数，await 等待协程完成
  • asyncio.run() 启动事件循环并运行协程
  • asyncio.gather() / create_task() 实现并发执行
  • aiohttp 库用于异步 HTTP 请求
  • asyncio.to_thread / run_in_executor 桥接同步代码
  • 异步适用于 I/O 密集型并发场景，不适合 CPU 密集型

[[返回 Python 首页|python/index]]

Released under the MIT License.

Copyright © 2018-present raopan 饶盼 base on VitePress 2.0.0-alpha.17