DEV Community

BAOFUFAN
BAOFUFAN

Posted on

asyncio Pitfalls: The Mistake That Cost Me 3 Hours

#python #asyncio #爬虫 #性能优化

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())
Enter fullscreen mode Exit fullscreen mode

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())
Enter fullscreen mode Exit fullscreen mode

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")
Enter fullscreen mode Exit fullscreen mode

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 的结果也丢了")
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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.

Top comments (0)