2256 字
11 分钟

Python 异步编程完全指南 2026：asyncio 核心 + aiohttp 并发请求 + 实战陷阱规避

2026-07-25
Python教程
Python
/
asyncio
/
异步编程
/
aiohttp
/
并发
/
协程
/
高性能

Python 的 asyncio 让单线程代码能同时处理数千个并发 IO 操作。但很多开发者在从同步代码迁移到异步时容易踩坑——比如在协程里用了阻塞的 time.sleep，或者误以为 async def 就会自动并行执行。

本文从概念到实战，厘清 asyncio 核心，重点讲解并发控制和常见陷阱。

一、核心概念：协程、Task、Future 三者关系#

import asyncio


# ── 协程（Coroutine）──────────────────────────────────────────
# async def 定义的函数，调用时不会立即执行，返回协程对象
async def fetch_data(url: str) -> str:
    await asyncio.sleep(1)   # await 表示在这里挂起，等待 IO
    return f"data from {url}"


# 调用协程函数只是创建了协程对象，不会运行！
coro = fetch_data("https://api.example.com")  # 返回 coroutine 对象
# 要运行，需要：
# 1. await 它（在另一个协程中）
# 2. 包装成 Task（asyncio.create_task / asyncio.ensure_future）
# 3. 顶层：asyncio.run(coro)




# ── Task ──────────────────────────────────────────────────────
# Task 是对协程的包装，会被事件循环立即调度执行
# 创建 Task 后，协程就会开始运行（不需要 await 它）
async def main():
    # 方式一：asyncio.create_task（推荐）
    task1 = asyncio.create_task(fetch_data("url1"))
    task2 = asyncio.create_task(fetch_data("url2"))


    # 此时 task1 和 task2 已经在后台运行了！
    # 等待两个任务都完成
    result1 = await task1
    result2 = await task2
    print(result1, result2)




# ── 事件循环 ──────────────────────────────────────────────────
asyncio.run(main())   # Python 3.7+，创建事件循环并运行顶层协程

二、asyncio 核心 API 速查#

2.1 并发运行：gather vs as_completed vs TaskGroup#

import asyncio
import time


async def task(n: int, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"task {n} done (delay={delay}s)"


# ── asyncio.gather：等待所有，统一返回 ─────────────────────────
async def use_gather():
    start = time.perf_counter()
    results = await asyncio.gather(
        task(1, 2.0),   # 3 个协程并发运行
        task(2, 1.0),
        task(3, 0.5),
        return_exceptions=True,   # True：异常作为结果返回，不抛出
    )
    elapsed = time.perf_counter() - start
    print(f"总耗时：{elapsed:.2f}s")   # 约 2.0s（取最长）
    print(results)
    # ['task 1 done (delay=2.0s)', 'task 2 done (delay=1.0s)', 'task 3 done (delay=0.5s)']
    # 顺序与输入一致！


# ── asyncio.as_completed：谁先完成先处理谁 ────────────────────
async def use_as_completed():
    tasks = [task(1, 2.0), task(2, 1.0), task(3, 0.5)]
    async for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"已完成：{result}")
    # 输出顺序：task 3 → task 2 → task 1（按完成时间）


# ── TaskGroup（Python 3.11+，推荐）────────────────────────────
async def use_taskgroup():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(task(1, 2.0))
        t2 = tg.create_task(task(2, 1.0))
    # with 块结束时，所有 task 都已完成
    print(t1.result(), t2.result())
    # 优势：任何一个 task 异常，其他 task 会被自动取消（结构化并发）

2.2 超时控制#

async def slow_operation():
    await asyncio.sleep(10)
    return "result"


# ── asyncio.wait_for：超时取消 ────────────────────────────────
async def with_timeout():
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=3.0)
    except asyncio.TimeoutError:
        print("操作超时（3秒内未完成）")


# ── asyncio.timeout（Python 3.11+，更优雅）────────────────────
async def with_timeout_ctx():
    try:
        async with asyncio.timeout(3.0):
            result = await slow_operation()
    except TimeoutError:
        print("超时")

2.3 并发限制：Semaphore（信号量）#

不加限制的并发可能导致服务器过载或被限流。Semaphore 限制同时运行的协程数量：

import asyncio
import aiohttp


async def fetch_url(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore) -> dict:
    async with sem:   # 最多同时 10 个请求
        async with session.get(url) as resp:
            return {"url": url, "status": resp.status}


async def fetch_all(urls: list[str], max_concurrent: int = 10):
    sem = asyncio.Semaphore(max_concurrent)  # 最大并发数
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url, sem) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

三、aiohttp：高并发 HTTP 请求#

3.1 基础用法（Session 复用）#

import asyncio
import aiohttp


# ❌ 错误：每次请求都创建新的 Session（不复用 TCP 连接，性能差）
async def bad_example():
    for url in urls:
        async with aiohttp.ClientSession() as session:  # 每次循环创建！
            async with session.get(url) as resp:
                pass


# ✅ 正确：一个 Session 处理所有请求（复用 TCP 连接池）
async def good_example(urls: list[str]) -> list[str]:
    connector = aiohttp.TCPConnector(
        limit=100,           # 全局最大连接数
        limit_per_host=10,   # 每个主机最大连接数（防止对单一服务器过度请求）
        ssl=False,           # 是否验证 SSL（生产环境保持 True）
    )
    timeout = aiohttp.ClientTimeout(total=30, connect=5)  # 总超时 30s，连接超时 5s


    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers={"User-Agent": "MyBot/1.0"},
    ) as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)


async def fetch_one(session: aiohttp.ClientSession, url: str) -> str:
    try:
        async with session.get(url) as resp:
            resp.raise_for_status()         # 4xx/5xx 抛出异常
            return await resp.text()
    except aiohttp.ClientError as e:
        return f"Error: {e}"

3.2 实战：批量下载 URL 内容（带限速 + 重试）#

import asyncio
import aiohttp
from typing import NamedTuple


class FetchResult(NamedTuple):
    url: str
    content: str | None
    error: str | None


async def fetch_with_retry(
    session: aiohttp.ClientSession,
    url: str,
    sem: asyncio.Semaphore,
    max_retries: int = 3,
) -> FetchResult:
    async with sem:
        for attempt in range(max_retries):
            try:
                async with session.get(url) as resp:
                    if resp.status == 200:
                        content = await resp.text()
                        return FetchResult(url=url, content=content, error=None)
                    elif resp.status == 429:  # Too Many Requests
                        retry_after = int(resp.headers.get("Retry-After", 5))
                        await asyncio.sleep(retry_after)
                    else:
                        return FetchResult(url=url, content=None, error=f"HTTP {resp.status}")
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt == max_retries - 1:
                    return FetchResult(url=url, content=None, error=str(e))
                await asyncio.sleep(2 ** attempt)  # 指数退避


async def batch_fetch(urls: list[str], max_concurrent: int = 20) -> list[FetchResult]:
    sem = asyncio.Semaphore(max_concurrent)
    connector = aiohttp.TCPConnector(limit=max_concurrent * 2)
    timeout = aiohttp.ClientTimeout(total=15)


    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [fetch_with_retry(session, url, sem) for url in urls]
        results = await asyncio.gather(*tasks)


    success = sum(1 for r in results if r.error is None)
    print(f"成功：{success}/{len(urls)}")
    return list(results)


# 使用
urls = [f"https://httpbin.org/delay/{i % 3}" for i in range(50)]
results = asyncio.run(batch_fetch(urls))

四、在 asyncio 中处理同步阻塞代码#

4.1 run_in_executor（关键！）#

import asyncio
import time
import requests  # 同步库


# ❌ 错误：直接在协程中调用阻塞函数
async def bad():
    time.sleep(2)          # 阻塞整个事件循环！
    requests.get("...")     # 阻塞整个事件循环！


# ✅ 正确：用 run_in_executor 在线程池中运行
async def good():
    loop = asyncio.get_event_loop()


    # 方式一：run_in_executor（Python 3.6+）
    await loop.run_in_executor(None, time.sleep, 2)
    result = await loop.run_in_executor(None, requests.get, "https://example.com")


    # 方式二：asyncio.to_thread（Python 3.9+，更简洁）
    result = await asyncio.to_thread(requests.get, "https://example.com")


    # 方式三：使用自定义线程池（控制并发数）
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=4) as pool:
        result = await loop.run_in_executor(pool, cpu_intensive_task, data)

4.2 异步文件 IO（aiofiles）#

import aiofiles
import asyncio


async def read_file(path: str) -> str:
    async with aiofiles.open(path, 'r', encoding='utf-8') as f:
        return await f.read()


async def write_file(path: str, content: str) -> None:
    async with aiofiles.open(path, 'w', encoding='utf-8') as f:
        await f.write(content)


# 并发读取多个文件
async def read_many(paths: list[str]) -> list[str]:
    tasks = [read_file(p) for p in paths]
    return await asyncio.gather(*tasks)

五、异步数据库（asyncpg / SQLAlchemy async）#

# asyncpg：直接操作 PostgreSQL（最快）
import asyncpg
import asyncio


async def db_example():
    # 创建连接池（整个应用复用）
    pool = await asyncpg.create_pool(
        "postgresql://user:password@localhost/mydb",
        min_size=5,
        max_size=20,
    )


    # 查询
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users WHERE active = $1", True)
        for row in rows:
            print(row["id"], row["name"])


        # 插入
        await conn.execute(
            "INSERT INTO users (name, email) VALUES ($1, $2)",
            "Alice", "alice@example.com"
        )


        # 事务
        async with conn.transaction():
            await conn.execute("UPDATE accounts SET balance = balance - $1 WHERE id = $2", 100, 1)
            await conn.execute("UPDATE accounts SET balance = balance + $1 WHERE id = $2", 100, 2)


    await pool.close()

六、10 个常见异步陷阱#

陷阱 1：忘记 await#

# ❌ 忘记 await，协程不会执行
async def bad():
    asyncio.sleep(1)   # 没有 await！这行什么都没做
    print("这里立刻执行")


# ✅ 正确
async def good():
    await asyncio.sleep(1)
    print("1秒后执行")

陷阱 2：误以为 async def 会自动并行#

# ❌ 顺序执行（不是并发！）
async def sequential():
    await fetch("url1")   # 等待完成
    await fetch("url2")   # 再等待
    await fetch("url3")   # 再等待
    # 总耗时 = 三者之和


# ✅ 并发执行
async def concurrent():
    await asyncio.gather(
        fetch("url1"),
        fetch("url2"),
        fetch("url3"),
    )
    # 总耗时 = 最慢那个

陷阱 3：在协程中直接调用阻塞函数#

# ❌ 阻塞整个事件循环
async def bad():
    time.sleep(5)           # 阻塞！
    result = requests.get("...")  # 阻塞！


# ✅ 用 to_thread 包装
async def good():
    await asyncio.to_thread(time.sleep, 5)
    result = await asyncio.to_thread(requests.get, "...")

陷阱 4：aiohttp.ClientSession 不复用#

# ❌ 每次请求创建 Session，不复用连接池
async def bad(urls):
    for url in urls:
        async with aiohttp.ClientSession() as s:  # 每次都创建！
            await s.get(url)


# ✅ Session 在整个请求批次中复用
async def good(urls):
    async with aiohttp.ClientSession() as s:
        tasks = [s.get(url) for url in urls]
        await asyncio.gather(*tasks)

陷阱 5：未限制并发导致服务器过载#

# ❌ 10000 个并发请求直接打出去
async def bad(urls):
    async with aiohttp.ClientSession() as s:
        await asyncio.gather(*[s.get(url) for url in urls])  # 一次性 10000 个！


# ✅ 用 Semaphore 控制并发
async def good(urls, limit=50):
    sem = asyncio.Semaphore(limit)
    async with aiohttp.ClientSession() as s:
        async def fetch(url):
            async with sem:
                return await s.get(url)
        await asyncio.gather(*[fetch(url) for url in urls])

陷阱 6：异常被 gather 吞掉#

# gather 默认：任何一个协程异常，gather 立即抛出（其他任务继续运行但结果丢失）
# return_exceptions=True：把异常作为结果返回，不抛出


results = await asyncio.gather(
    good_task(),
    failing_task(),   # 这个会抛出异常
    return_exceptions=True,   # 关键！
)
for r in results:
    if isinstance(r, Exception):
        print(f"出错：{r}")
    else:
        print(f"成功：{r}")

陷阱 7：在同步代码中调用 asyncio.run 两次#

# ❌ asyncio.run 会创建新事件循环，已在循环中时不能调用
async def outer():
    asyncio.run(inner())   # RuntimeError: This event loop is already running


# ✅ 在 async 函数中，直接 await
async def outer():
    await inner()

七、性能对比：同步 vs 多线程 vs asyncio#

以并发发送 100 个 HTTP 请求为例：

import asyncio
import aiohttp
import requests
import time
from concurrent.futures import ThreadPoolExecutor


URLS = ["https://httpbin.org/delay/1"] * 100


# ── 同步：顺序执行 ────────────────────────────────────────────
def sync_test():
    t = time.perf_counter()
    for url in URLS:
        requests.get(url)
    print(f"同步：{time.perf_counter()-t:.1f}s")   # 约 100s


# ── 多线程：并发 IO，但有线程开销 ────────────────────────────
def thread_test():
    t = time.perf_counter()
    with ThreadPoolExecutor(max_workers=20) as pool:
        list(pool.map(requests.get, URLS))
    print(f"多线程：{time.perf_counter()-t:.1f}s")  # 约 5s（20线程）


# ── asyncio：轻量并发，无线程切换开销 ────────────────────────
async def async_test():
    t = time.perf_counter()
    async with aiohttp.ClientSession() as session:
        sem = asyncio.Semaphore(50)
        async def fetch(url):
            async with sem, session.get(url) as r:
                await r.text()
        await asyncio.gather(*[fetch(u) for u in URLS])
    print(f"asyncio：{time.perf_counter()-t:.1f}s")  # 约 2s


asyncio.run(async_test())
方案耗时（100 个请求，每个延迟 1s）内存占用
同步（顺序）~100s低
多线程（20线程）~5s中
asyncio（50并发）~2s低
asyncio（100并发）~1s低

相关文章：

本文基于 Python 3.12 / aiohttp 3.9 / asyncpg 0.29 编写。asyncio.TaskGroup 和 asyncio.timeout 需要 Python 3.11+。

Python 异步编程完全指南 2026：asyncio 核心 + aiohttp 并发请求 + 实战陷阱规避
https://971918.xyz/posts/python-guide/python-async-concurrency-guide/
作者
九所长
发布于
2026-07-25
许可协议
CC BY-NC-SA 4.0
Trojan 协议完全指南：TLS 流量伪装原理、Xray 服务端部署与全平台客户端配置（2026最新）
Python 性能优化完全指南：从性能分析到内存优化与 JIT 加速实战
© 2026 九所长. All Rights Reserved. / 关于我们 / 友情链接 / 免责声明 / 服务条款 / 隐私政策 / RSS / Sitemap
Powered by Astro & Fuwari