FastAPI 监控

AsyncIO 调试模式

快速定位阻塞事件循环的端点：

PYTHONASYNCIODEBUG=1 python main.py

任何耗时超过 100ms 的任务会打印警告，帮你找到 async def 里的阻塞调用。

线程池监控

实时监控线程池使用情况，排查线程耗尽问题：

import anyio
from anyio.to_thread import current_default_thread_limiter

async def monitor_thread_limiter():
    limiter = current_default_thread_limiter()
    threads_in_use = limiter.borrowed_tokens
    while True:
        if threads_in_use != limiter.borrowed_tokens:
            print(f"Threads in use: {limiter.borrowed_tokens}")
            threads_in_use = limiter.borrowed_tokens
        await anyio.sleep(0)

可在 lifespan 里启动：

from contextlib import asynccontextmanager
import anyio

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with anyio.create_task_group() as tg:
        tg.start_soon(monitor_thread_limiter)
        yield

请求耗时追踪

用纯 ASGI 中间件记录每个请求的处理时间：

import time
from loguru import logger

class TimingMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        start = time.perf_counter()
        await self.app(scope, receive, send)
        elapsed = time.perf_counter() - start

        if elapsed > 1.0:
            logger.warning(f"Slow request: {scope['method']} {scope['path']} {elapsed:.3f}s")
        else:
            logger.info(f"{scope['method']} {scope['path']} {elapsed:.3f}s")

Sentry 集成

生产环境接入 Sentry 捕获异常：

import sentry_sdk
from src.config import settings

if settings.SENTRY_DSN:
    sentry_sdk.init(dsn=settings.SENTRY_DSN)