Python asyncio: Coroutines, Event Loop & async/await

Why Asynchronous Programming

CPython has the GIL: at any moment only one thread executes Python bytecode, so threads do not speed up CPU-bound work. For IO-bound work (HTTP calls, database queries), async lets one thread switch to other tasks while waiting — thousands of connections with a single process.

Scenario	Threads	Async
1000 idle HTTP connections	1000 OS threads — heavy	1 thread, trivial
CPU-heavy computation	No gain (GIL)	No gain, event loop blocks
Many concurrent DB/HTTP waits	Works, but thread switch cost	Best fit

Coroutines, async and await

import asyncio

async def fetch_user(uid: int) -> str:
    await asyncio.sleep(0.5)        # 模拟 IO 等待
    return f"user-{uid}"

# 调用协程返回的是协程对象，不会执行
# 必须由事件循环调度

async def main():
    # 串行：约 1 秒
    a = await fetch_user(1)
    b = await fetch_user(2)

    # 并发：约 0.5 秒
    a, b = await asyncio.gather(
        fetch_user(1), fetch_user(2)
    )

asyncio.run(main())   # 启动事件循环并运行

async def defines a coroutine function; calling it returns a coroutine object.
await suspends the coroutine until the awaited awaitable completes, yielding control back to the event loop.
asyncio.gather runs awaitables concurrently; asyncio.create_task schedules a task without waiting.

Real-world Async Clients

import asyncio
import aiohttp

async def fetch_titles(urls):
    async with aiohttp.ClientSession() as session:
        async def get(url):
            async with session.get(url) as resp:
                html = await resp.text()
                return html
        return await asyncio.gather(*[get(u) for u in urls])

asyncio.run(fetch_titles(["https://example.com" for _ in range(100)]))

Every library in the call chain must be async — requests is synchronous and will block the loop. Use aiohttp/httpx for HTTP, asyncpg/aiomysql or SQLAlchemy async for databases.

Common Pitfalls

Forgetting await — the coroutine never runs and you get a 'coroutine was never awaited' warning.
Mixing blocking calls (time.sleep, open(), requests) inside async code blocks the whole loop.
Creating one asyncio.run per request instead of a long-lived loop.

为什么要异步编程

CPython 有 GIL：任意时刻只有一个线程在执行 Python 字节码，所以多线程无法加速 CPU 密集任务。但对 IO 密集（HTTP 调用、数据库查询），异步让线程在等待时切换到其他任务——单进程即可处理数千连接。

场景	多线程	异步
1000 个空闲 HTTP 连接	1000 个操作系统线程，开销大	1 个线程，毫无压力
CPU 密集计算	无收益（GIL）	无收益，事件循环被阻塞
大量并发数据库/HTTP 等待	可行，但线程切换有成本	最佳场景

协程、async 与 await

import asyncio

async def fetch_user(uid: int) -> str:
    await asyncio.sleep(0.5)        # 模拟 IO 等待
    return f"user-{uid}"

# 调用协程返回的是协程对象，不会执行
# 必须由事件循环调度

async def main():
    # 串行：约 1 秒
    a = await fetch_user(1)
    b = await fetch_user(2)

    # 并发：约 0.5 秒
    a, b = await asyncio.gather(
        fetch_user(1), fetch_user(2)
    )

asyncio.run(main())   # 启动事件循环并运行

async def 定义协程函数，调用它返回协程对象而不执行。
await 暂停协程直到被等待对象完成，期间把控制权交还事件循环。
asyncio.gather 并发执行多个任务；asyncio.create_task 先调度不等待。

真实异步客户端

import asyncio
import aiohttp

async def fetch_titles(urls):
    async with aiohttp.ClientSession() as session:
        async def get(url):
            async with session.get(url) as resp:
                html = await resp.text()
                return html
        return await asyncio.gather(*[get(u) for u in urls])

asyncio.run(fetch_titles(["https://example.com" for _ in range(100)]))

调用链上所有库都必须是异步的——requests 是同步库，会阻塞事件循环。HTTP 用 aiohttp/httpx，数据库用 asyncpg/aiomysql 或 SQLAlchemy 2.0 异步引擎。

常见坑

忘记 await——协程根本没执行，报「coroutine was never awaited」警告。
在异步代码里混用阻塞调用（time.sleep、open()、requests）会卡死整个事件循环。
每次请求建一个 asyncio.run，而不是复用长期运行的事件循环。