正在加载,请稍候…

FastAPI 生产部署:Docker、Gunicorn 与 Nginx

将 FastAPI 应用部署到生产环境——异步工作器、Gunicorn 配置、Docker 多阶段构建、Nginx 反向代理、健康检查与零停机部署。

FastAPI 生产部署:Docker、Gunicorn 与 Nginx

FastAPI 生产环境

FastAPI 已为生产就绪,但需要正确配置——合适的 ASGI 服务器、进程管理和基础设施设置。

FastAPI 生产部署:Docker、Gunicorn 与 Nginx 示意图

ASGI 服务器选项

Uvicorn(开发环境)

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Gunicorn + Uvicorn 工作器(生产环境)

pip install gunicorn uvicorn[standard]

gunicorn main:app \
  --workers 4 \
  --worker-class uvicorn.workers.UvicornWorker \
  --bind 0.0.0.0:8000 \
  --timeout 30 \
  --keep-alive 5 \
  --log-level info

工作器数量公式(2 × CPU 核心数) + 1

应用结构

fastapi-app/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用工厂
│   ├── config.py        # 设置(pydantic-settings)
│   ├── database.py      # 数据库连接
│   ├── dependencies.py  # 共享依赖
│   ├── models/
│   ├── schemas/
│   ├── routers/
│   └── services/
├── tests/
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
└── gunicorn.conf.py

FastAPI 生产部署:Docker、Gunicorn 与 Nginx 示意图

使用 pydantic-settings 进行配置

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "My API"
    debug: bool = False
    database_url: str
    redis_url: str = "redis://localhost:6379"
    secret_key: str
    allowed_hosts: list[str] = ["*"]
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

应用工厂模式

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from contextlib import asynccontextmanager

from app.config import get_settings
from app.database import init_db
from app.routers import users, products, orders

settings = get_settings()

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动
    await init_db()
    yield
    # 关闭
    # 清理连接

def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.app_name,
        debug=settings.debug,
        lifespan=lifespan,
        docs_url="/docs" if settings.debug else None,
    )
    
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.allowed_hosts,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    app.add_middleware(TrustedHostMiddleware, allowed_hosts=["*"])
    
    app.include_router(users.router, prefix="/api/v1")
    app.include_router(products.router, prefix="/api/v1")
    
    return app

app = create_app()

Dockerfile(多阶段构建)

# 构建阶段
FROM python:3.12-slim as builder

WORKDIR /build
RUN pip install poetry
COPY pyproject.toml poetry.lock ./
RUN poetry export -f requirements.txt --output requirements.txt --without-hashes

# 生产阶段
FROM python:3.12-slim

WORKDIR /app

# 安全:非 root 用户
RUN addgroup --system app && adduser --system --group app

# 安装依赖
COPY --from=builder /build/requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY --chown=app:app ./app ./app

USER app

EXPOSE 8000

CMD ["gunicorn", "app.main:app", "--workers", "4", 
     "--worker-class", "uvicorn.workers.UvicornWorker",
     "--bind", "0.0.0.0:8000"]

Nginx 配置

upstream fastapi_backend {
    server app:8000;
    keepalive 32;
}

server {
    listen 80;
    server_name api.example.com;
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl http2;
    server_name api.example.com;

    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;

    client_max_body_size 10M;

    location / {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_read_timeout 60s;
        proxy_connect_timeout 10s;
    }

    location /health {
        proxy_pass http://fastapi_backend/health;
        access_log off;
    }
}

FastAPI 生产部署:Docker、Gunicorn 与 Nginx 示意图

健康检查端点

from fastapi import APIRouter
from sqlalchemy import text
from app.database import get_db

router = APIRouter()

@router.get("/health")
async def health_check(db=Depends(get_db)):
    try:
        await db.execute(text("SELECT 1"))
        db_status = "healthy"
    except Exception:
        db_status = "unhealthy"
    
    return {
        "status": "healthy" if db_status == "healthy" else "degraded",
        "database": db_status,
        "version": "1.0.0",
    }

Gunicorn 配置文件

# gunicorn.conf.py
import multiprocessing

bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
timeout = 30
keepalive = 5
preload_app = True
accesslog = "-"
errorlog = "-"
loglevel = "info"

零停机部署

# 使用 Docker Swarm 滚动更新
docker service update \
  --image myapp:v2 \
  --update-delay 10s \
  --update-parallelism 1 \
  myapp

# 或使用 Kubernetes
kubectl set image deployment/fastapi-app app=myapp:v2
kubectl rollout status deployment/fastapi-app

性能优化建议

  • 启用 响应压缩(GZipMiddleware)
  • 使用 异步数据库驱动(asyncpg、motor)
  • 为昂贵查询添加 Redis 缓存
  • 使用 py-spypyinstrument 进行性能分析
  • 设置合适的 连接池大小