如何对FastAPI端点进行内存分析？求排查API内存泄漏的简易中间件

从你提供的内存趋势图来看：

内存随时间持续增长，符合典型的内存泄漏特征。结合你用fastapi-cprofile排查性能的经验，以下是用中间件检测内存泄漏的可行方案：

一、基础内存监控中间件（快速定位可疑接口） #

这个中间件会记录每个请求前后的内存变化，帮你找出每次调用都会导致内存增长的接口：

import psutil
from fastapi import Request, FastAPI
from time import time

app = FastAPI()

def get_memory_usage():
    # 获取当前进程内存使用（单位：MB）
    process = psutil.Process()
    return process.memory_info().rss / (1024 * 1024)

@app.middleware("http")
async def memory_monitor_middleware(request: Request, call_next):
    start_mem = get_memory_usage()
    response = await call_next(request)
    end_mem = get_memory_usage()
    mem_diff = end_mem - start_mem

    # 仅记录内存增长超过阈值的请求（可根据业务调整）
    if mem_diff > 1:
        print(f"[内存告警] 路径: {request.url.path} | 增长: {mem_diff:.2f}MB")
    
    return response

使用说明：

• 依赖psutil，需安装：pip install psutil
• 运行后观察日志，反复出现内存增长的接口就是泄漏可疑点

二、进阶内存快照中间件（定位具体泄漏代码） #

针对可疑接口，用tracemalloc生成内存快照对比，直接定位到代码行：

import tracemalloc
from fastapi import Request, FastAPI

app = FastAPI()

@app.middleware("http")
async def leak_trace_middleware(request: Request, call_next):
    # 仅针对可疑接口启用跟踪（替换成你的目标路径）
    target_path = "/suspected-leaky-endpoint"
    if request.url.path == target_path:
        tracemalloc.start()
        snapshot_before = tracemalloc.take_snapshot()

    response = await call_next(request)

    if request.url.path == target_path:
        snapshot_after = tracemalloc.take_snapshot()
        tracemalloc.stop()

        # 对比快照，输出Top10内存增长来源
        print(f"\n[接口 {target_path} 内存增长详情]")
        top_stats = snapshot_after.compare_to(snapshot_before, "lineno")
        for stat in top_stats[:10]:
            print(stat)
    
    return response

使用说明：

• tracemalloc是Python3.4+自带库，无需额外安装
• 快照对比会显示哪些代码行创建了最多的对象，直接指向泄漏根源

三、K8s环境下的辅助排查 #

结合你的部署环境，补充两个实用手段：

• 集群层面监控：用Prometheus采集container_memory_working_set_bytes指标，关联请求量与内存增长趋势，确认泄漏是否和特定流量模式相关
• 手动触发快照接口：新增一个内部接口，调用时生成内存快照（比如用objgraph.show_growth()），方便按需排查：

  import objgraph
  from fastapi import FastAPI
  
  app = FastAPI()
  
  @app.get("/internal/memory-snapshot")
  def get_memory_snapshot():
      objgraph.show_growth(limit=10)
      return {"status": "snapshot generated"}
  

注意事项 #

• 内存监控会带来一定性能开销，建议仅在测试/预发环境启用，或添加开关控制（比如通过环境变量ENABLE_MEM_MONITOR）
• Docker容器中需确保安装对应依赖：psutil、objgraph（后者需pip install objgraph）

内容的提问来源于stack exchange，提问作者Adam