Here’s the story: last week my boss threw a “simple” task at me — pull data from 120 internal APIs simultaneously and compile a report. I thought, “This is just I/O-bound work. I know asyncio like the back of my hand.” So I cranked out the first version in 10 minutes. To my disbelief, it ran even slower than a serial approach, and some endpoints never returned any data. That afternoon, I stared at the terminal output, tweaking and cursing for three full hours — until I spotted one innocuous function call. Then it all clicked.

If you’re doing concurrency with asyncio, the following pitfalls might make you question your life choices.

1. The Culprit: Synchronous Blocking Call Inside a Coroutine

Here’s my first naive implementation — can you spot the problem right away?

import asyncio
import time
import requests # 注意：经典的同步库

async def fetch_api(url):
 """协程函数：获取 API 数据"""
 print(f"Starting {url}")
 # 模拟获取数据 —— 这里埋了一颗大雷
 resp = requests.get(url, timeout=10) # 同步阻塞调用！
 data = resp.json()
 print(f"Finished {url}")
 return data

async def main():
 urls = [f"https://httpbin.org/delay/1?req={i}" for i in range(10)]
 start = time.time()
 results = await asyncio.gather(*[fetch_api(u) for u in urls])
 elapsed = time.time() - start
 print(f"10 请求耗时: {elapsed:.2f}s")
 return results

asyncio.run(main())

The result left me dumbfounded: 10 requests took over 10 seconds — exactly like a serial run. The reason is painfully simple: requests.get() is a synchronous blocking call. While waiting for the network, it completely holds the thread hostage, so the event loop never gets a chance to switch to another coroutine. Mixing synchronous code into an async def is like stuffing a tractor engine into a sports car. The golden rule of asyncio is: every I/O operation must be asynchronous.

Two ways to fix it: swap to an async HTTP library (like aiohttp), or offload the blocking call with loop.run_in_executor. I recommend the former:

import aiohttp
import asyncio
import time

async def fetch_api(session, url):
 async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
 return await resp.json()

async def main():
 urls = [f"https://httpbin.org/delay/1?req={i}" for i in range(10)]
 start = time.time()
 async with aiohttp.ClientSession() as session:
 tasks = [fetch_api(session, u) for u in urls]
 results = await asyncio.gather(*tasks)
 elapsed = time.time() - start
 print(f"10 请求耗时: {elapsed:.2f}s")
 return results

asyncio.run(main())

After the switch, 10 requests finished in about 1.5 seconds. My boss’s frown finally relaxed.

2. Forgetting to await — The Coroutine That Never Ran

This trap has bitten me more times than I’d like to admit. Check out this classic:

async def say_hello():
 await asyncio.sleep(1)
 print("Hello")

async def main():
 # 事故现场：创建协程对象，但忘了 await
 say_hello() # 只会返回一个 coroutine object，不会执行
 await asyncio.sleep(2)
 print("End")

When you run it, the terminal only prints End. The Hello never appears. Python doesn’t raise an error — it silently creates a coroutine object and drops it into the void. The correct approach is await say_hello(), or wrap it with asyncio.create_task(say_hello()) so the event loop manages it. My personal habit: whenever I call an async def function, I either put await in front of it or wrap it with create_task. I never leave a coroutine naked.

3. Exception Handling in gather — One Rotten Task Spoils the Whole Bunch

When I took on that 120‑endpoint task, a few APIs occasionally timed out or returned 500. I used asyncio.gather and quickly learned that if a single task raises an exception, all the other tasks — finished or unfinished — get cancelled, leaving me with zero usable data.

# 错误示范：一个炸，全家炸
async def bad_request():
 await asyncio.sleep(0.5)
 raise ValueError("接口挂了")

async def good_request():
 await asyncio.sleep(1)
 return "正常数据"

async def main():
 try:
 results = await asyncio.gather(bad_request(), good_request())
 except ValueError:
 print("捕获异常，但 good_request 的结果也丢了")

The fix is simple — add return_exceptions=True to gather:

results = await asyncio.gather(
 task1, task2, ...,
 return_exceptions=True
)
for r in results:
 if isinstance(r, Exception):
 log_error(r) # 单独处理异常
 else:
 process(r)

With this pattern, you can gracefully handle partial failures — log the errors and still process all the valid responses. No more wasting 3 hours staring at the terminal!

These pitfalls are sneaky, but once you understand the underlying mechanics, asyncio becomes a powerful ally. Hope this saves you from the same debugging rabbit hole I fell into.