問題: Python はシングルスレッドです

Python には Global Interpreter Lock (GIL) があります。バイトコードを実行するのは 1 つの Python スレッドだけです 一度に。これはよく誤解されます。 GIL は競争を妨げない — を防ぎます CPU並列処理 スレッドの。 I/O バウンドのワークロードの場合 (API は通常そうなります)、asyncio は効率の点でスレッドよりも優れています。 OS コンテキスト切り替えのオーバーヘッドを排除します。

何を学ぶか

  • イベントループ: 非同期実行を制御するループ
  • コルーチン: 一時停止および再開できる関数
  • async/await: コルーチンを読み取り可能にする構文
  • タスクと将来: asyncio プリミティブ
  • I/O バウンドと CPU バウンド: 非同期が役立つ場合とそうでない場合
  • 構造化された同時実行のための asyncio.gather と asyncio.TaskGroup
  • 実際のワークロードでの FastAPI 同期と非同期のベンチマーク

イベント ループ: asyncio の核心

イベント ループは、コールバックのキューを管理する無限ループであり、 コルーチン。コルーチンが I/O 操作に遭遇したとき (await）、 一時停止して制御をイベント ループに戻し、実行できるようにします。 その間に他のコルーチンを実行します。 I/O が完了すると、コルーチンはリセットされます 撮影するために並んでいます。 

# Visualizzazione concettuale dell'event loop (pseudocodice)
#
# Event Loop Iteration:
# 1. Guarda la coda delle callback pronte
# 2. Esegui la prima callback/coroutine
# 3. Se incontra un await su I/O:
#    - Registra l'operazione I/O con il sistema operativo (epoll/kqueue/IOCP)
#    - Metti la coroutine in "sospensione" (waiting)
#    - Torna al passo 1 (esegui la prossima callback disponibile)
# 4. Quando l'I/O completa (notifica OS):
#    - Rimetti la coroutine nella coda "pronta"
# 5. Ripeti

# In Python reale:
import asyncio

async def fetch_data(url: str) -> str:
    # Simulazione di una richiesta HTTP asincrona
    # await sospende questa coroutine finche la risposta non arriva
    # L'event loop nel frattempo puo eseguire altre coroutine
    await asyncio.sleep(1)  # Simula latenza di rete
    return f"Data from {url}"

async def main():
    # Esecuzione sequenziale: 3 secondi totali
    result1 = await fetch_data("https://api1.example.com")
    result2 = await fetch_data("https://api2.example.com")
    result3 = await fetch_data("https://api3.example.com")
    return [result1, result2, result3]

# asyncio.run() crea l'event loop e lo esegue
asyncio.run(main())

コルーチンと通常の関数

通常の関数 (def) なしで最初から最後まで実行されます 中断。コルーチン (async def) は、 特定の時点で一時停止できます (await）そしてあきらめます イベントループ制御。 

import asyncio
import time

# --- FUNZIONE SINCRONA ---
def fetch_sync(url: str) -> str:
    time.sleep(1)  # BLOCCA l'intero thread per 1 secondo
    return f"Data from {url}"

def main_sync():
    start = time.time()
    results = [
        fetch_sync("https://api1.example.com"),  # aspetta 1s
        fetch_sync("https://api2.example.com"),  # aspetta 1s
        fetch_sync("https://api3.example.com"),  # aspetta 1s
    ]
    print(f"Sync: {time.time() - start:.2f}s")  # ~3.00s
    return results

# --- COROUTINE ASINCRONA ---
async def fetch_async(url: str) -> str:
    await asyncio.sleep(1)  # SOSPENDE la coroutine, NON il thread
    return f"Data from {url}"

async def main_async():
    start = time.time()
    # gather esegue le tre coroutine CONCORRENTEMENTE
    results = await asyncio.gather(
        fetch_async("https://api1.example.com"),
        fetch_async("https://api2.example.com"),
        fetch_async("https://api3.example.com"),
    )
    print(f"Async: {time.time() - start:.2f}s")  # ~1.00s
    return results

# La differenza: 3s vs 1s per lo stesso workload I/O

タスクとasyncio.gather

asyncio.gather() これは最も一般的な実行方法です コルーチンを同時に実行します。すべてのコルーチンが完了すると戻ります (またはデフォルトで失敗した場合)。 

import asyncio
from typing import Any

# asyncio.gather: esecuzione concorrente di piu coroutine
async def concurrent_fetches():
    # Tutte e tre iniziano quasi simultaneamente
    results = await asyncio.gather(
        fetch_async("https://api1.example.com"),
        fetch_async("https://api2.example.com"),
        fetch_async("https://api3.example.com"),
        return_exceptions=True,  # Errori restituiti come valori invece di eccezioni
    )

    for url, result in zip(["api1", "api2", "api3"], results):
        if isinstance(result, Exception):
            print(f"{url}: Error - {result}")
        else:
            print(f"{url}: {result}")

# asyncio.create_task: esecuzione in background
async def background_tasks():
    # Task 1 inizia subito
    task1 = asyncio.create_task(fetch_async("https://api1.example.com"))

    # Fai altro mentre task1 e in background
    await asyncio.sleep(0.5)  # Simula altro lavoro

    # task2 parte dopo 0.5s
    task2 = asyncio.create_task(fetch_async("https://api2.example.com"))

    # Aspetta entrambi
    result1 = await task1
    result2 = await task2
    return result1, result2

# asyncio.TaskGroup (Python 3.11+): concorrenza strutturata
async def structured_concurrency():
    results = []
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_async("https://api1.example.com"))
        task2 = tg.create_task(fetch_async("https://api2.example.com"))
        task3 = tg.create_task(fetch_async("https://api3.example.com"))
    # Qui tutti i task sono completati (o c'e stata un'eccezione)
    return [task1.result(), task2.result(), task3.result()]

FastAPI の async def: いつ使用するか

FastAPI は両方をサポートします async def それ def 彼らにとっては普通のこと ルート。選択は関数の動作によって異なります。 

# FastAPI: async def vs def
from fastapi import FastAPI
import asyncio
import httpx  # Client HTTP asincrono

app = FastAPI()

# USA async def quando:
# - Fai operazioni I/O con librerie async (httpx, asyncpg, aioredis, etc.)
# - Chiami altre coroutine con await
@app.get("/async-example")
async def async_endpoint():
    # httpx.AsyncClient e la versione async di requests
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
        return response.json()

# USA def normale quando:
# - La funzione e puramente CPU (calcoli, elaborazione in memoria)
# - Usi librerie sincrone che non supportano async
# FastAPI esegue le funzioni sync in un thread pool separato
# per non bloccare l'event loop
@app.get("/sync-example")
def sync_endpoint():
    import json
    # Operazione CPU-bound: OK in def normale
    data = {"numbers": list(range(1000))}
    return json.dumps(data)

# CASO CRITICO: MAI fare I/O sincrono bloccante in async def
@app.get("/bad-example")
async def bad_endpoint():
    import requests  # SBAGLIATO: requests e sincrono
    # Questo BLOCCA l'event loop per la durata della richiesta HTTP!
    response = requests.get("https://api.example.com")  # NON FARE QUESTO
    return response.json()

# VERSIONE CORRETTA:
@app.get("/good-example")
async def good_endpoint():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com")
        return response.json()

Async における同期ライブラリの危険性

同期 I/O ライブラリ (次のような) を使用します。 requests, psycopg2、 または従来の DB クライアント) コルーチン内 async def ブロック イベント ループ全体: まで他のリクエストは処理できません。 操作は完了しません。データベース用途向け asyncpg o SQLAlchemy 2.0 async。 HTTP で使用する場合 httpx o aiohttp。 Redis を使用する場合 redis.asyncio.

I/O バウンドと CPU バウンド: 主な違い

非同期はワークロードにのみ役立ちます I/Oバウンド: 操作 プログラムは外部リソース (データベース、HTTP API、ファイルシステム) を待ちます。ワークロード別 CPU バウンド (機械学習、ビデオエンコーディング、集中的な計算) asyncio は役に立ちません – 役に立ちます multiprocessing または遺言執行者。 

# Workload I/O-bound: asyncio aiuta molto
async def io_bound_handler():
    # Fa 3 chiamate API in ~1 secondo invece di ~3 secondi
    results = await asyncio.gather(
        fetch_user_from_db(user_id=1),      # ~50ms
        fetch_user_orders(user_id=1),        # ~80ms
        fetch_user_preferences(user_id=1),   # ~40ms
    )
    return results  # Pronto in ~80ms (il piu lento), non 170ms

# Workload CPU-bound: asyncio NON aiuta, usa ProcessPoolExecutor
from concurrent.futures import ProcessPoolExecutor
import asyncio

executor = ProcessPoolExecutor(max_workers=4)

def cpu_intensive_task(data: list) -> list:
    # Sorting O(n log n), computazione pura
    return sorted(data, key=lambda x: x ** 2)

@app.post("/process")
async def process_data(data: list):
    loop = asyncio.get_event_loop()
    # run_in_executor esegue la funzione in un processo separato
    # senza bloccare l'event loop
    result = await loop.run_in_executor(
        executor,
        cpu_intensive_task,
        data,
    )
    return {"processed": result}

ベンチマーク: FastAPI での同期と非同期

これは、エンドポイント同期と非同期の違いを示す現実的なベンチマークです。 同時リクエストが 100 件ある I/O バウンドのワークロードの場合: 

# Benchmark con httpx e asyncio (script di test)
# pip install httpx
import asyncio
import httpx
import time

async def benchmark(endpoint: str, n_requests: int = 100):
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        start = time.time()
        tasks = [client.get(endpoint) for _ in range(n_requests)]
        responses = await asyncio.gather(*tasks)
        elapsed = time.time() - start

        success = sum(1 for r in responses if r.status_code == 200)
        rps = n_requests / elapsed

        print(f"{endpoint}: {elapsed:.2f}s, {rps:.1f} req/s, {success}/{n_requests} success")

# Endpoint test nel server FastAPI
@app.get("/test/sync")
def sync_test():
    import time
    time.sleep(0.1)  # Simula 100ms DB query
    return {"data": "ok"}

@app.get("/test/async")
async def async_test():
    await asyncio.sleep(0.1)  # Simula 100ms DB query async
    return {"data": "ok"}

# Risultati tipici su un server con 4 worker Uvicorn:
# /test/sync:  10.23s, 9.8 req/s   (quasi sequenziale!)
# /test/async: 1.05s, 95.2 req/s   (quasi perfettamente concorrente)
#
# Con 100ms di latenza simulata:
# Sync:  100 richieste * 100ms = ~10s
# Async: concorrente = ~100ms + overhead

asyncio.run(benchmark("/test/sync"))
asyncio.run(benchmark("/test/async"))

結論

FastAPI における Python async の威力は本物ですが、理解が必要です モデル: イベント ループはシングルスレッドですが、I/O はノンブロッキングです。 コルーチンは他のハンドラーと競合をブロックすることなく一時停止されます。 これは協調的です (OS スレッドのようなプリエンプティブではありません)。次のステップは FastAPI が依存するデータ検証を提供する Pydantic v2 を理解します。

FastAPI シリーズの今後の記事

  • 第3条： Pydantic v2 — 高度な検証、BaseModel および TypeAdapter
  • 第4条： FastAPI での依存関係の挿入: depends() とパターン クリーン
  • 第5条： SQLAlchemy 2.0 と Alembic を使用した非同期データベース