asyncio 还是 threading？用 3 个问题选对了

1. 概述

许多新学 Python 并发的人常问“asyncio 还是 threading 更好？”——这个问题一开始就错了，因为这两个工具解决两个不同的问题。本文通过 3 个具体问题让你自己回答，并附上可运行的代码示例。读完这篇文章，你将准确知道什么时候用 threading，什么时候用 asyncio，以及什么时候两者都无用（必须用 multiprocessing）。

2. 为什么“哪个更快”这个问题是错的

Python 有全局解释器锁（GIL）——无论机器有多少 CPU 核心，同一时刻只有一个 Python 线程在执行字节码。threading 和 asyncio 都无法打破 GIL。两者只能帮助 CPU 在等待 I/O（读文件、网络调用、数据库查询）时不被“阻塞”——它们无法加速计算。

import time

def cpu_bound_task():
    total = 0
    for i in range(10_000_000):
        total += i * i
    return total

start = time.perf_counter()
cpu_bound_task()
cpu_bound_task()
print(f"Sequential: {time.perf_counter() - start:.2f}s")

Sequential: 1.85s

无论你用 threading.Thread 还是 asyncio.gather 运行两次 cpu_bound_task()，总时间不会减少，因为 GIL 同时只允许一个线程运行 Python 代码。这就是下面第一个问题最重要的原因。

**注意：**如果任务是 CPU 绑定的（重型数值计算、图像处理、加密等），threading 和 asyncio 都帮不了。你需要 multiprocessing 或用 C/Rust 扩展（NumPy、Cython）编写耗时部分。

3. 问题 1：你的任务是 I/O 绑定还是 CPU 绑定？

这是决定你是否应该使用“等待”类型的并发（I/O 绑定）的问题。

3.1. I/O 绑定：threading 和 asyncio 都很高效

import time
import requests
from concurrent.futures import ThreadPoolExecutor

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

def fetch(url: str) -> int:
    return requests.get(url).status_code

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as pool:
    results = list(pool.map(fetch, URLS))
print(f"Threading: {time.perf_counter() - start:.2f}s, results={results}")

Threading: 1.12s, results=[200, 200, 200, 200, 200]

ThreadPoolExecutor 用 5 个操作系统线程同时运行 5 个请求。因为 requests.get 是阻塞式调用等待网络，当线程在 C 层等待 I/O 时，GIL 被释放——所以 5 个请求真正并行运行，总时间约 1 秒而不是 5 秒。

import asyncio
import aiohttp
import time

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

async def fetch(session: aiohttp.ClientSession, url: str) -> int:
    async with session.get(url) as resp:
        return resp.status

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

start = time.perf_counter()
results = asyncio.run(main())
print(f"Asyncio: {time.perf_counter() - start:.2f}s, results={results}")

Asyncio: 1.08s, results=[200, 200, 200, 200, 200]

结果与 threading 相同，但机制不同：只有一个线程，事件循环主动 await 并在等待 I/O 时让出控制权，无需 OS 创建新线程。

3.2. CPU 绑定：两者都不如 multiprocessing

from concurrent.futures import ProcessPoolExecutor

def cpu_bound_task():
    return sum(i * i for i in range(10_000_000))

start = time.perf_counter()
with ProcessPoolExecutor(max_workers=2) as pool:
    results = list(pool.map(cpu_bound_task, [None, None]))
print(f"Multiprocessing: {time.perf_counter() - start:.2f}s")

Multiprocessing: 1.02s

ProcessPoolExecutor 创建独立的进程，每个进程有自己的 GIL，因此可以真正利用多个 CPU 核心。如果你在 CPU 绑定问题上还在纠结’threading 还是 asyncio’，正确答案是两者都错——使用 multiprocessing。

4. 问题 2：你使用的生态系统支持 async 吗？

这是在真实代码库中工作时最实用的问题，因为你不是从零开始编写所有东西。

import psycopg2  # 同步驱动，阻塞式

def get_user(user_id: int) -> dict:
    conn = psycopg2.connect("dbname=app")
    cur = conn.cursor()
    cur.execute("SELECT id, name FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    conn.close()
    return {"id": row[0], "name": row[1]}

psycopg2 是阻塞式库——它没有 await 版本。想要用这个驱动并行运行多个查询，threading（或 asyncio 的 run_in_executor）是自然的选择，因为你无需重写任何代码。

import asyncpg  # 原生异步驱动

async def get_user(conn: asyncpg.Connection, user_id: int) -> dict:
    row = await conn.fetchrow(
        "SELECT id, name FROM users WHERE id = $1", user_id
    )
    return {"id": row["id"], "name": row["name"]}

如果你的生态系统已经有原生异步驱动（asyncpg、aiohttp、MongoDB 的 motor 等），asyncio 能发挥最大优势：在一个线程上有数千个并发连接，没有创建 OS 线程的开销。

注意：在 async def 中混入阻塞式代码但不正确地 await（例如在协程中直接调用 requests.get）会阻塞整个事件循环——其他所有任务都必须等待，甚至不相关的任务也一样。这是从 threading 转向 asyncio 时最常见的错误。

5. 问题 3：你需要多少个并发任务？

当任务数量增加时，线程和协程之间的资源成本差异非常大。

import threading
import resource

def noop():
    threading.Event().wait(2)

threads = [threading.Thread(target=noop) for _ in range(2000)]
for t in threads:
    t.start()

peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
print(f"Peak RSS with 2000 threads: {peak_kb / 1024:.1f} MB")

for t in threads:
    t.join()

Peak RSS with 2000 threads: 1850.3 MB

每个 OS 线程默认占用数百 KB 到数 MB 的栈，所以 2000 个线程容易消耗近 2 GB 的 RAM，并给 OS 调度器带来巨大的上下文切换压力。

import asyncio
import resource

async def noop():
    await asyncio.sleep(2)

async def main():
    tasks = [asyncio.create_task(noop()) for _ in range(2000)]
    await asyncio.gather(*tasks)
    peak_kb = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    print(f"Peak RSS with 2000 coroutines: {peak_kb / 1024:.1f} MB")

asyncio.run(main())

Peak RSS with 2000 coroutines: 95.7 MB

协程是纯 Python 对象，不需要独立的 OS 线程——创建 2000 个任务的成本仅为几十 MB。如果你需要处理数千个并发连接（Web 服务器、聊天服务器、大规模爬虫），asyncio 几乎是必选；threading 会在用尽带宽前先耗尽资源。

6. 快速决策表

问题	答案 → 选择
1. I/O 绑定还是 CPU 绑定？	CPU 绑定 → multiprocessing。I/O 绑定 → 继续问题 2
2. 你使用的库有原生异步支持吗？	否（旧的阻塞驱动）→ threading。是（aiohttp、asyncpg 等）→ asyncio
3. 需要多少个并发任务？	几十个 → threading 足够简单。数百/数千 → asyncio

在实践中，许多系统两者都用：asyncio 用于大规模网络 I/O 绑定层，结合 run_in_executor 在单独的线程池中运行阻塞式代码段（例如旧库），以避免阻塞事件循环。

import asyncio

def legacy_blocking_call(x: int) -> int:
    import time
    time.sleep(1)  # 模拟旧的阻塞式库
    return x * 2

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, legacy_blocking_call, 21)
    print(result)

asyncio.run(main())

42

run_in_executor 将阻塞式函数推送到默认线程池中的一个线程，事件循环仍可自由处理等待期间的其他协程。

7. 用 pytest 测试行为

例如测试确认 asyncio.gather 真正并行运行任务，而不是顺序运行：

import asyncio
import time
import pytest

async def sleepy_task(seconds: float) -> float:
    await asyncio.sleep(seconds)
    return seconds

@pytest.mark.asyncio
async def test_gather_runs_concurrently():
    start = time.perf_counter()
    results = await asyncio.gather(
        sleepy_task(0.5), sleepy_task(0.5), sleepy_task(0.5)
    )
    elapsed = time.perf_counter() - start

    assert results == [0.5, 0.5, 0.5]
    # 如果顺序运行需要约 1.5 秒，并行运行只需约 0.5 秒
    assert elapsed < 0.7

测试所需的依赖：

pytest
pytest-asyncio

这个测试确认了 asyncio 的本质：三个 sleep(0.5) 任务在一个线程上并发运行只需约 0.5 秒，而不是 1.5 秒——如果 elapsed < 0.7 断言失败，说明你的代码是按顺序 await 而不是 gather。

8. 常见错误

• 在协程中使用 time.sleep()：必须使用 await asyncio.sleep()。time.sleep() 是阻塞式调用，会阻止整个事件循环。
• 忘记 await：调用 fetch(session, url) 而不 await 只会创建一个协程对象，什么都不会运行——Python 会警告 RuntimeWarning: coroutine was never awaited。
• 在不加锁的情况下在线程间共享状态：threading 有真正的竞态条件（多个线程写同一个变量），需要 threading.Lock。asyncio 在单个线程上没有这种竞态条件，但如果在应该是“原子”的代码段中间 await 仍可能有逻辑 bug。
• 为 CPU 绑定问题使用 ThreadPoolExecutor 然后失望于没有更快——因为 GIL 仍在阻止，使用 ProcessPoolExecutor。

**注意：**如果你的团队已经熟悉 threading 且代码库全是阻塞式库，不要仅因为它“更新”就急着重构到 asyncio。重写所有驱动、处理不同的异常、调试复杂的协程堆栈跟踪的成本，相比于在并发任务数量不大时获得的好处要大得多。

9. 结论

在 asyncio 和 threading 之间选择不是关于’哪个更快’，而是三个具体问题：任务是 I/O 绑定还是 CPU 绑定、库生态是否支持原生异步、以及你需要处理的并发规模。CPU 绑定总是需要 multiprocessing。I/O 绑定搭配旧库和小任务量则 threading 简单易集成。I/O 绑定大规模搭配原生异步库则 asyncio 是性能和资源上的正确选择。