Last week, my boss asked me to speed up an old web scraping project. I thought, "No problem — I’ll just throw asyncio at it, fetch concurrently, and theoretically slash 200 requests from 40 seconds to around 2." But after writing the first version, while it was indeed faster, half the data was missing and the console was flooded with RuntimeWarning: coroutine was never awaited. Even worse, the program would randomly freeze with 0% CPU and eventually time out. After three hours of poring over docs and source code, I finally filled in all the pitfalls. This post captures those gotchas and the best practices I ended up with, so you don’t have to go down the same rabbit hole.

The Scenario: 200 API data fetches, from 40s to 2s

The original synchronous code looked like this — simple but painfully slow:

import time
import requests

def fetch_all(urls):
    results = []
    for url in urls:
        resp = requests.get(url, timeout=5)
        results.append(resp.json())
    return results

urls = [f"https://api.example.com/item/{i}" for i in range(200)]
start = time.time()
data = fetch_all(urls)
print(f"耗时: {time.time() - start:.2f}s")
# 输出: 耗时: 41.23s

200 sequential requests taking over 40 seconds — a terrible experience. Brimming with confidence, I set out to rewrite it with asyncio.

Core Concept: Event Loop + Coroutines = Non-blocking Concurrency

asyncio works completely differently from multithreading. It runs a single-threaded event loop where all coroutines are scheduled within the same thread. When a coroutine hits an IO wait (network, disk), it voluntarily yields control back to the event loop via await. The loop immediately switches to other coroutines that are ready to run. This way, the CPU never idly spins waiting for IO.

The most basic pattern:

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

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

urls = [f"https://api.example.com/item/{i}" for i in range(200)]
asyncio.run(main())

asyncio.gather schedules all coroutines concurrently, so the total time is roughly the duration of the slowest request, not the sum. In theory, wonderful — but as soon as I ran it, the pitfalls began.

Pitfall 1: Sneaky Synchronous Blocking Calls Inside Coroutines

At first, I took a shortcut and kept using requests.get inside the coroutine, thinking simply wrapping it in async def would work. But the event loop got completely stuck on requests.get — concurrency went out the window.

# 错误示范
import requests

async def fetch_bad(url):
    resp = requests.get(url)   # 同步阻塞！事件循环被堵死
    return resp.json()

The requests library does synchronous I/O. Once called, the entire thread blocks waiting for the network response, and the event loop loses all control. For asyncio, you must use libraries that support async/await — like aiohttp for HTTP and aiomysql for database queries.

Solution: Replace all I/O with libraries from the async/await ecosystem. For unavoidable synchronous code, offload it to a thread pool using loop.run_in_executor:

import concurrent.futures

def sync_heavy_work(data):
    # 这是一个没法改写的同步 CPU 计算
    return sum(i * i for i in range(data))

async def run_in_thread(data):
    loop = asyncio.get_running_loop()
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, sync_heavy_work, data)
    return result

This prevents blocking the event loop, but don't overuse it — thread switching overhead still exists.

Pitfall 2: gather Raises on First Exception, Discarding Other Results

With 200 requests, a couple of timeouts or 500s are expected. The first time I used gather, if a single task raised an exception, the entire gather would immediately throw, and the other 190+ successful responses were discarded. I was so frustrated I slapped my thigh.

results = await asyncio.gather(*tasks)  # 一个挂了，全部白干

Solution: gather has a return_exceptions=True parameter. Instead of raising, it places exception objects directly into the result list, so you can handle them like regular return values:

results = await asyncio.gather(*tasks, return_exceptions=True)

for i, res in enumerate(results):
    if isinstance(res, Exception):
        print(f"任务 {i} 失败: {res}")
    else:
        process(res)

This way, even if 10 tasks timeout, you still get the data from the remaining 190. I also like to add timeout and retry logic inside fetch:

from aiohttp import ClientTimeout

async def fetch(session, url, retries=2):
    for attempt in range(retries):
        try:
            async with session.get(url, timeout=Clien