本番環境での非同期 I/O パターン

イベントループのベストプラクティス

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

接続プール

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

asyncio.Queueによるプロデューサー・コンシューマー

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

セマフォによるレート制限

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

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

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

指数バックオフによる再試行

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

グレースフルシャットダウン

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

ストリーミングレスポンス

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

FastAPIのバックグラウンドタスク

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

ハートビートパターン

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

非同期コードのプロファイリング

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

よくある落とし穴の回避

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

確認問題

まとめ