---
description: "FastAPI 监控的约定：AsyncIO 调试模式、线程池监控、请求耗时追踪、Sentry 集成"
---

# FastAPI 监控

## AsyncIO 调试模式

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

```bash
PYTHONASYNCIODEBUG=1 python main.py
```

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

## 线程池监控

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

```python
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 里启动：

```python
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 中间件记录每个请求的处理时间：

```python
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 捕获异常：

```python
import sentry_sdk
from src.config import settings

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