Python 异步编程进阶：asyncio + aiohttp + aiomysql 完整实战

小飞兽 Python 10 次阅读 2026-07-24

Introduction

asyncio 是 Python 3.5+ 的异步编程标准库。本文讲解 async/await 语法、事件循环、并发任务（gather/semaphore）、异步上下文管理器、以及 aiohttp/aiomysql 在异步 Web 开发中的用法。

---

async/await 基础

import asyncio

async def fetch_data(n):
    await asyncio.sleep(1)  # 模拟 I/O
    return f"数据 {n}"

async def main():
    # 顺序执行（慢）
    r1 = await fetch_data(1)
    r2 = await fetch_data(2)
    print(r1, r2)

    # 并发执行（快）
    results = await asyncio.gather(fetch_data(1), fetch_data(2), fetch_data(3))
    print(results)  # ['数据 1', '数据 2', '数据 3']

asyncio.run(main())

并发限制（信号量）

import asyncio, time

async def task(n):
    async with semaphore:
        print(f"任务 {n} 开始")
        await asyncio.sleep(1)
        return n

async def main():
    global semaphore
    semaphore = asyncio.Semaphore(3)  # 最多3个并发
    start = time.time()
    tasks = [task(i) for i in range(10)]
    results = await asyncio.gather(*tasks)
    print(f"总耗时: {time.time()-start:.1f}秒")  # ~4秒（10个任务，3个并发）

asyncio.run(main())

aiohttp 异步 HTTP

import aiohttp, asyncio

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

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, f"https://httpbin.org/get?id={i}") for i in range(10)]
        results = await asyncio.gather(*tasks)
        print(f"获取了 {len(results)} 个响应")

asyncio.run(main())

常见问题

Q1: asyncio vs threading vs multiprocessing？asyncio 单线程协程，I/O 密集并发首选。threading 多线程，受 GIL 限制但简单。multiprocessing 多进程，CPU 密集并行首选。

Q2: async 函数里能调普通同步代码吗？能，但同步代码会阻塞整个事件循环。需要调同步代码时用 await loop.run_in_executor(None, sync_function)。

Q3: asyncio 程序怎么调试？用 Python 3.7+ 的 asyncio.run()，异常会正确传播。用 aiotask contextvars 做请求追踪。