The fundamental pitfall: blocking calls inside the loop
asyncio concurrency rests on a single thread with cooperative scheduling. The loop runs one task at a time and only yields control when a task hits a real await on asynchronous I/O. Put a blocking call inside a coroutine and the whole loop stalls — every other coroutine waits, and your optimistic concurrency degrades into serial execution.
There are three classes of blocking sources: synchronous database drivers (psycopg2, pymysql), synchronous HTTP clients (requests), and anything calling time.sleep, touching files, or doing CPU-heavy work. Writing them directly into an async def parks the event loop at that line, which shows up as "all requests slow, CPU mostly idle".
There are two isolation options. The lightweight one is asyncio.to_thread(fn, *args) (Python 3.9+), which uses the default executor and keeps the code minimal. to_thread is a newer convenience method whose default executor is loop.run_in_executor(None, ...).
When you need to size the pool or supply a custom executor, use run_in_executor with an explicit ThreadPoolExecutor. Remember that max_workers is shared globally: if 1000 concurrent tasks call to_thread against a default executor with 32 slots, the rest queue up and it looks like the async API hangs. Either create an executor sized for the workload, or bound concurrency with a semaphore.
import asyncio
async def fetch_one(client, url):
# 正确：异步 HTTP 客户端，await 时会让出控制权
resp = await client.get(url)
return await resp.text()
async def read_legacy_csv(path):
# 同步库：隔离到线程池，避免阻塞事件循环
def _read():
with open(path, encoding="utf-8") as f:
return f.read()
return await asyncio.to_thread(_read)
# 需要自定义容量时
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=64)
data = await loop.run_in_executor(executor, blocking_fn, arg1, arg2)gather versus TaskGroup: how exceptions differ
Both asyncio.gather and asyncio.TaskGroup (Python 3.11+) run coroutines concurrently, but their failure semantics differ in a way that catches people. gather returns on the first exception by default yet does **not** cancel the still-running tasks — they continue in the background while you lose control over them, producing "ghost tasks".
To fix that, pass return_exceptions=True: every task then runs to completion and exceptions are collected into the returned results list for you to handle. The cost is that in the failure path you still wait for the slowest task.
TaskGroup embodies structured concurrency: when any child task fails, the group cancels all remaining tasks automatically and raises an ExceptionGroup (or a single exception) on exit. No leaked background tasks, no half-completed states, and no silently forgotten work.
Recommendation: prefer TaskGroup in new code — it enforces the invariant that concurrent tasks cannot outlive their enclosing scope. Fall back to gather only for older Python compatibility or when you want to inspect results individually instead of failing as a group. If you use gather without group-failure semantics, remember return_exceptions=True.
# 旧写法：一个失败，其他仍在后台跑（幽灵任务）
results = await asyncio.gather(*tasks)
# 改进：全部跑完，异常作为结果返回
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
if isinstance(r, Exception):
log.warning("task failed: %s", r)
# 推荐写法：TaskGroup 自动取消其余任务
async with asyncio.TaskGroup() as tg:
for url in urls:
tg.create_task(fetch_one(client, url))
# 失败时抛出 ExceptionGroup，其余任务已被取消Creating the loop and the forgotten await
The most basic mistake is never starting the loop. Defining async def main() and calling main() directly produces a coroutine object that does nothing and raises nothing — unless you enable the "coroutine was never awaited" warning. The correct entry point is asyncio.run(main()), which creates, runs and closes the loop.
The second common error is forgetting await. Calling asyncio.gather(...) without await yields a Future rather than results, and for many APIs no request is even issued; async with session.get(url) without await never calls __aenter__. The signature is "the code ran and nothing happened", with only a RuntimeWarning as evidence.
Turning on warnings is the best-value defence available. Python 3.12+ emits an explicit warning for un-awaited coroutines; on older versions enable -W error::RuntimeWarning or set the PYTHONASYNCIODEBUG=1 environment variable, which also reports slow callbacks exceeding 100ms.
A third worthwhile development-time check is whether a sync build of a library got installed instead of the async one — requests rather than httpx, psycopg2 rather than psycopg. Static analysis rarely catches this, but in production it shows up immediately as concurrency that does nothing at all.
import asyncio
async def main():
# 记得每个协程都要 await
data = await asyncio.gather(fetch_a(), fetch_b())
return data
if __name__ == "__main__":
asyncio.run(main())
# 开发时开启调试：慢回调与未 await 协程告警
# PYTHONASYNCIODEBUG=1 python app.pyTimeout control and concurrency limiting
The most dangerous property of asynchronous code is that failures propagate silently. An await that never returns leaves the request hanging forever. An async service without timeout control has effectively converted fast failures into slow leaks.
Use asyncio.wait_for for per-call timeouts. Note the pre-3.11 behaviour difference: on timeout it cancels the inner task and raises TimeoutError, but if the cancelled code has a broad except Exception or a cleanup that ignores cancellation, the real return can come noticeably later than the deadline. Python 3.11+ offers the asyncio.timeout() context manager with clearer semantics.
For a deadline covering a group of operations, wrap them in asyncio.timeout() (3.11+); it converts expiry into TimeoutError and handles cancellation propagation correctly. If you need partial results after a deadline, asyncio.wait is the better tool — it waits without cancelling.
Rate and concurrency limiting use asyncio.Semaphore. Its meaning is "at most N at a time"; queued waiters still all complete eventually. That differs from many rate-limiting middlewares and is worth thinking through at design time: a semaphore bounds concurrency, not rate. If you also need a velocity limit such as 10 requests per second, implement a token bucket separately.
import asyncio
async def call_with_limits(client, url, sem):
# 并发度限制：同时最多 20 个在飞
async with sem:
# 单次超时（Python 3.11+ 推荐写法）
async with asyncio.timeout(5):
resp = await client.get(url)
return resp.status
async def main(urls):
sem = asyncio.Semaphore(20)
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(call_with_limits(c, u, sem)) for u in urls]
return [t.result() for t in tasks]Connection pools: sizing and lifecycle
Pool parameters are easy to get wrong. httpx Limits(max_connections=N) bounds total connections and max_keepalive_connections bounds idle retention; aiohttp TCPConnector(limit=N) is the same idea. Defaults are often small, so a few dozen concurrent requests already queue.
Size the pool by dividing concurrency by average response time to get required throughput, then take the smaller of that and what the upstream can actually accept. Aligning the limit with the upstream application-level cap matters: firing 500 concurrent requests at a service that permits 50 only amplifies timeouts and retry storms.
Regarding lifecycle, the pool belongs to the application process, not to the request. Create it at startup, share it globally, and close it during shutdown. Building a pool per request makes handshake latency dominate and drives the TIME_WAIT count through the roof.
The last recurring problem is connection leakage: calling client.get() without closing the response body keeps the connection checked out until the pool is exhausted. Prefer async with client.stream(...) or fully read every response. A pool exhaustion symptom looks like "the first few thousand requests are fine, then everything times out".
import httpx
limits = httpx.Limits(
max_connections=100, # 总连接数上限，按上游承受能力设置
max_keepalive_connections=20, # 空闲连接保留数
keepalive_expiry=30.0,
)
# 绑定到应用生命周期，全局复用，关闭时释放
client = httpx.AsyncClient(limits=limits, timeout=httpx.Timeout(10.0))
# 读取响应体必须完整，否则连接不会归还池
async def ok(url):
async with client.stream("GET", url) as resp:
chunks = [c async for c in resp.aiter_bytes()]
return b"".join(chunks)