实战篇 / 第 25 章

第 25 章 · 部署与运维实战

部署全景：一条请求的完整旅程

在动手敲命令之前，先建立一个心智模型。当一个用户访问 https://api.example.com/orders 时，这条请求会依次穿过下面这些「站点」，每层都有自己该负责的事，也有自己最容易出事的地方。把这条链路记住，后面所有技术选型就都有了依据。

部署方式的四种形态

依赖固化：把「我这里能跑」变成可复现

可复现的第一步是把依赖从「我 pip install 了一堆」变成「一个锁文件」。现代 Python 推荐用 pyproject.toml 声明 + 锁定文件记录精确版本。工具链上，uv 已经成为速度最快的选择（Rust 编写，比 pip 快一到两个数量级），pip-tools、poetry 仍是稳妥选择。

pyproject.toml · 声明「允许什么」

[project]
name = "shop-api"
version = "1.4.0"
description = "订单服务"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "uvicorn[standard]>=0.32",
    "sqlalchemy>=2.0",
    "psycopg[binary]>=3.2",
    "pydantic-settings>=2.6",
    "structlog>=24.4",
    "redis>=5.2",
]

[project.optional-dependencies]
dev = ["pytest", "ruff", "mypy", "httpx"]

[project.scripts]
shop-admin = "app.cli:main"        # 装包后可直接在命令行调用

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["app"]

生成锁文件（本地）

# 方式一：uv（最快，推荐）
pip install uv
uv lock                 # 生成 uv.lock，精确到每个依赖的哈希
uv sync                 # 按锁文件建虚拟环境
uv sync --group dev     # 连开发依赖一起装

# 方式二：pip-tools（老项目兼容）
pip install pip-tools
pip-compile pyproject.toml -o requirements.txt
pip-compile pyproject.toml --extra dev -o requirements-dev.txt

# 方式三：直接导出当前环境（应急）
pip freeze > requirements.lock     # 缺点：包含所有间接依赖，来源不明

依赖的四类与处理策略

requirements.lock · 生产镜像里实际执行的文件

# 本文件由 uv lock 后同步导出，只含运行时依赖，带哈希
fastapi==0.115.6
uvicorn[standard]==0.32.1
sqlalchemy==2.0.36
psycopg[binary]==3.2.3
pydantic-settings==2.7.0
structlog==24.4.0
redis==5.2.1
anyio==4.7.0
idna==3.10

Docker 镜像：从分层原理到多阶段构建

理解 Docker 镜像的关键在于分层（layer）与写时复制（copy-on-write）。镜像由多个只读层叠成容器，每条 Dockerfile 指令产生一层。某一层发生变化时，只有它及其之后的层会重建；未变化的层会从缓存或镜像仓库直接复用。这就是「Dockerfile 顺序决定构建速度」的原因。

指令与缓存行为对照

反例：缓存完全失效的 Dockerfile

FROM python:3.12
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt          # 问题：改一行代码 -> 这一层作废 -> 全部依赖重装
CMD ["python", "main.py"]

正例：分层清晰 + 多阶段构建

# ---------- 阶段 1：构建（带编译工具链）----------
FROM python:3.12-slim AS builder

ENV PIP_NO_CACHE_DIR=1 \
    PIP_DISABLE_PIP_VERSION_CHECK=1

# 系统构建依赖，只存在于 builder 阶段
RUN apt-get update \
 && apt-get install -y --no-install-recommends build-essential \
 && rm -rf /var/lib/apt/lists/*

WORKDIR /build
# 先只拷贝依赖清单 -> 代码改动不会让这一层失效
COPY pyproject.toml requirements.lock ./
RUN pip wheel --wheel-dir /wheels -r requirements.lock

# ---------- 阶段 2：运行（干净、无编译器）----------
FROM python:3.12-slim AS runtime

ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PIP_NO_CACHE_DIR=1 \
    TZ=Asia/Shanghai

# 运行时真正需要的系统库
RUN apt-get update \
 && apt-get install -y --no-install-recommends libpq5 curl \
 && rm -rf /var/lib/apt/lists/* \
 && useradd -r -u 10001 -m appuser

WORKDIR /app
COPY --from=builder /wheels /wheels
COPY requirements.lock ./
RUN pip install --no-index --find-links=/wheels -r requirements.lock \
 && rm -rf /wheels

# 代码放在最后，改代码只重建这一层
COPY --chown=appuser:appuser app ./app
COPY --chown=appuser:appuser gunicorn_conf.py ./

USER appuser
EXPOSE 8000

# 健康检查：容器自己告诉编排系统「我活着吗」
HEALTHCHECK --interval=30s --timeout=3s --start-period=20s --retries=3 \
  CMD curl -fsS http://127.0.0.1:8000/healthz || exit 1

ENTRYPOINT ["gunicorn", "--config", "gunicorn_conf.py", "app.main:app"]

体积与速度的实测对比

.dockerignore：别把垃圾打进镜像

# .dockerignore —— 没有它，.git 和 venv 会被一起 COPY 进镜像
.git
.github
.venv
venv/
__pycache__/
*.pyc
*.pyo
.pytest_cache/
.mypy_cache/
.ruff_cache/
htmlcov/
.coverage
.env
.env.*
!.env.example
data/
*.sqlite3
*.log
Dockerfile
docker-compose.yml
README.md
docs/
node_modules/

常用镜像管理命令

# 构建并打标签（tag 是给人看的版本号，不是 git 仓库地址）
docker build -t shop-api:1.4.0 -t shop-api:latest .

# 构建时避免每次都全量下载基础镜像
docker build --pull=false -t shop-api:1.4.0 .

# 查看分层，定位体积大头
docker history shop-api:1.4.0
docker images --format "table {{.Repository}}\t{{.Tag}}\t{{.Size}}"

# 导出 / 导入离线部署
docker save -o shop-api.tar shop-api:1.4.0
docker load -i shop-api.tar

# 进入正在运行的容器排查（用 --entrypoint 绕过 ENTRYPOINT）
docker exec -it shop-api-1 sh
docker run --rm -it --entrypoint sh shop-api:1.4.0

# 清理磁盘：按「悬空镜像」清理最安全
docker image prune           # 只删无标签的悬空镜像
docker system prune -a --volumes   # 危险：会删掉所有未被使用的镜像和卷

Docker Compose：把多服务拼成一套系统

Compose 用一个 YAML 文件描述一组容器的关系：谁依赖谁、谁在哪台机器上、数据挂在哪里、网络怎么连。开发环境里它替代「手动开五个终端跑五个 python 进程」，生产环境的小规模部署它也够用。

docker-compose.yml

name: shop

services:
  api:                                   # 主应用
    build:
      context: .
      dockerfile: Dockerfile
      target: runtime
    image: shop-api:1.4.0
    restart: unless-stopped                # 容器退出就自动拉起
    env_file: .env                        # 非敏感配置（可用 env_file）
    environment:
      DATABASE_URL: ${DATABASE_URL}
      REDIS_URL: redis://cache:6379/0
    depends_on:
      db:
        condition: service_healthy         # 等数据库健康再启动，而不是"启动了"
      cache:
        condition: service_started
    ports:
      - "127.0.0.1:8000:8000"              # 只绑本机回环，外部流量一律走 Nginx
    deploy:
      resources:
        limits:
          cpus: "2.0"
          memory: 1G
    healthcheck:
      test: ["CMD", "curl", "-fsS", "http://localhost:8000/healthz"]
      interval: 30s
      timeout: 3s
      retries: 3
      start_period: 20s
    logging:
      driver: json-file
      options:
        max-size: "20m"                    # 防止日志把磁盘写满
        max-file: "5"
    networks: [appnet]

  nginx:
    image: nginx:1.27-alpine
    restart: unless-stopped
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
      - ./nginx/certs:/etc/nginx/certs:ro
      - static:/var/www/static:ro
    depends_on:
      api:
        condition: service_healthy
    logging:
      driver: json-file
      options: {max-size: "20m", max-file: "5"}
    networks: [appnet]

  db:
    image: postgres:16-alpine
    restart: unless-stopped
    environment:
      POSTGRES_USER: shop
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      POSTGRES_DB: shop
    volumes:
      - pgdata:/var/lib/postgresql/data       # 命名卷持久化，容器删了数据还在
    command:
      - postgres
      - -c
      - max_connections=200
      - -c
      - shared_buffers=512MB
      - -c
      - log_min_duration_statement=200       # 超过 200ms 的查询写慢日志
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U shop -d shop"]
      interval: 10s
      timeout: 5s
      retries: 5
    networks: [appnet]

  cache:
    image: redis:7-alpine
    restart: unless-stopped
    command:
      - redis-server
      - --requirepass
      - ${REDIS_PASSWORD}
      - --maxmemory
      - 256mb
      - --maxmemory-policy
      - allkeys-lru          # 内存满了淘汰最近最少使用的键
    volumes:
      - redisdata:/data
    healthcheck:
      test: ["CMD", "redis-cli", "-a", "${REDIS_PASSWORD}", "ping"]
      interval: 10s
      timeout: 3s
      retries: 5
    networks: [appnet]

volumes:                    # 命名卷：数据生命周期独立于容器
  pgdata:
  redisdata:
  static:

networks:
  appnet:
    driver: bridge

Compose 常用命令速查

docker compose up -d              # 后台启动整套服务
docker compose up -d --build     # 改了代码重新构建再起
docker compose up -d --scale api=4   # 同一服务起 4 个容器（注意端口冲突）
docker compose ps                # 看状态与健康状况
docker compose logs -f --tail=200 api
docker compose exec api bash      # 进容器调试（已进对用户与目录）
docker compose exec -u root api bash   # 提权排查（生产慎用）
docker compose config             # 渲染并校验最终配置（提交前必跑）
docker compose restart api        # 只重启这一个服务
docker compose down              # 停止并删除容器（命名卷默认保留）
docker compose down -v           # 连数据卷一起删除 —— 生产绝对别手抖
docker compose run --rm api python manage.py shell   # 一次性任务容器

Nginx 反向代理与 gunicorn

生产环境绝不要让 Flask 自带的开发服务器（app.run()）直接对外。标准组合是：Nginx 在最前面做 TLS 与静态资源，gunicorn 在后面管理多个 Python worker 进程。

为什么不能直接跑 Flask 开发服务器

gunicorn 配置文件（Python 写法，可读性最好）

gunicorn_conf.py

"""Gunicorn 配置：把部署参数从命令行挪到可版本管理的文件里。"""
import multiprocessing
import os

bind = "0.0.0.0:8000"
worker_class = "gthread"                 # 线程 worker，适合 IO 密集（调数据库/外部 API）
workers = int(os.getenv("WEB_CONCURRENCY", multiprocessing.cpu_count() * 2 + 1))
threads = 4                              # 每个 worker 4 个线程 -> 总并发 workers×threads
worker_connections = 1000
timeout = 60                             # worker 处理超过 60 秒被杀掉重启
graceful_timeout = 30                    # 优雅停机的最长等待
keepalive = 5

max_requests = 2000                      # 单进程处理 2000 个请求后重启
max_requests_jitter = 200                # 加随机抖动，避免所有 worker 同时重启
preload_app = False                      # True 会把应用 fork 之前载入内存，内存翻倍慎用

accesslog = "-"                          # "-" = 输出到 stdout，由 Docker 收集
errorlog = "-"
loglevel = "info"
access_log_format = '%(h)s "%(r)s" %(s)s %(b)s %(M)sms "%(f)s" "%(a)s"'

# 反代之后要正确识别真实 IP
forwarded_allow_ips = "127.0.0.1"

# 预加载应用并延迟启动，让容器 healthcheck 有时间窗口
def when_ready(server):
    server.log.info("应用已就绪，开始接收请求")

def worker_int(worker):
    worker.log.info("worker 收到中断，开始优雅退出")


def post_fork(server, worker):
    # 每个 worker 独立初始化随机种子与连接池，避免 fork 后状态共享
    import random
    random.seed()
    server.log.info("worker 启动 pid=%s", worker.pid)

worker_class 怎么选

FastAPI/ASGI 的启动方式（与 gunicorn 的区别）

# 方式一：uvicorn 自管多进程（简单，适合中等规模）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4 --proxy-headers

# 方式二：gunicorn 做进程管理 + uvicorn worker 做协议解析（性能更佳，推荐）
gunicorn app.main:app \
    -k uvicorn.workers.UvicornWorker \
    -w 4 \
    -b 0.0.0.0:8000 \
    --timeout 60 \
    --graceful-timeout 30 \
    --access-logfile -

# 方式三：不用 gunicorn，直接用 Granian（Rust 实现，启动快、省内存）
granian --interface asgi app.main:app --workers 4 --port 8000

Nginx 配置：TLS、静态、代理、限流、日志

nginx/nginx.conf

worker_processes auto;

events {
    worker_connections 2048;
    multi_accept on;
}

http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    # 结构化 JSON 访问日志，便于日志系统解析
    log_format json_combined escape=json
        '{"ts":"$time_iso8601","ip":"$remote_addr",'
        '"method":"$request_method","uri":"$uri","status":$status,'
        '"bytes":$body_bytes_sent,"rt":$request_time,'
        '"ua":"$http_user_agent","rid":"$request_id"}';

    access_log /var/log/nginx/access.log json_combined;

    sendfile on;
    tcp_nopush on;
    keepalive_timeout 65;
    client_max_body_size 20m;
    server_tokens off;             # 隐藏 Nginx 版本号

    gzip on;
    gzip_types text/plain text/css application/json application/javascript text/css;
    gzip_min_length 1024;

    # ---------- 限流区：按客户端 IP ----------
    limit_req_zone $binary_remote_addr zone=api_zone:10m rate=20r/s;
    limit_conn_zone $binary_remote_addr zone=conn_zone:10m;

    # ---------- HTTP -> HTTPS ----------
    server {
        listen 80;
        server_name api.example.com;
        location /.well-known/acme-challenge/ { root /var/www/certbot; }
        location / { return 301 https://$host$request_uri; }
    }

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

        ssl_certificate     /etc/nginx/certs/fullchain.pem;
        ssl_certificate_key /etc/nginx/certs/privkey.pem;
        ssl_protocols       TLSv1.2 TLSv1.3;
        ssl_ciphers         HIGH:!aNULL:!MD5;
        ssl_session_cache   shared:SSL:10m;

        add_header Strict-Transport-Security "max-age=31536000" always;
        add_header X-Content-Type-Options    "nosniff"           always;
        add_header X-Frame-Options           "SAMEORIGIN"        always;

        # 静态资源交给 Nginx：命中就不进 Python，CPU 直接省一大截
        location /static/ {
            alias /var/www/static/;
            expires 30d;
            access_log off;
            gzip_static on;
        }

        location /media/ {
            alias /var/www/static/media/;
            expires 7d;
        }

        location / {
            limit_req zone=api_zone burst=40 nodelay;   # 削峰
            limit_conn conn_zone 50;

            proxy_pass http://127.0.0.1:8000;
            proxy_http_version 1.1;
            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_set_header Connection        "";
            proxy_set_header X-Request-Id      $request_id;   # 贯穿全链路的追踪 ID

            proxy_connect_timeout 5s;
            proxy_send_timeout    60s;
            proxy_read_timeout    60s;
            proxy_buffering on;
            proxy_buffers 8 16k;
        }

        # 接口错误单独记一份日志，方便快速定位
        error_log /var/log/nginx/error.log warn;

        # Nginx 自带的健康检查端点
        location = /nginx-health { access_log off; return 200 "ok\n"; }
    }
}

日志、监控与健康检查

运维的核心命题只有一个：出问题时，你能不能在 5 分钟内回答「发生了什么、影响了谁、怎么恢复」。这需要三件套：结构化日志、关键指标、主动告警。

日志的四个级别怎么用

结构化日志：用 structlog 输出 JSON

app/logging_conf.py

"""统一的结构化日志配置：本地可读，线上可被采集。"""
import logging
import logging.config
import os
import sys

import structlog

SERVICE = os.getenv("SERVICE_NAME", "shop-api")


def setup_logging(level: str | None = None) -> None:
    level = level or os.getenv("LOG_LEVEL", "INFO")
    shared = [
        structlog.contextvars.merge_contextvars,      # 自动带上 request_id 等上下文
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,        # 异常转成字符串，不丢栈
        structlog.processors.UnicodeDecoder(),
    ]
    if sys.stdout.isatty() and os.getenv("LOG_FORMAT") != "json":
        renderer = structlog.dev.ConsoleRenderer(colors=True)
    else:
        renderer = structlog.processors.JSONRenderer()

    structlog.configure(
        processors=[*shared, renderer],
        wrapper_class=structlog.make_filtering_bound_logger(
            logging.getLevelName(level)
        ),
        logger_factory=structlog.PrintLoggerFactory(),
        cache_logger_on_first_use=True,
    )

    # 第三方库（uvicorn、sqlalchemy）的日志也统一成 JSON
    logging.config.dictConfig({
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {"json": {"()": structlog.stdlib.ProcessorFormatter,
                                "processor": renderer}},
        "handlers": {"stdout": {"class": "logging.StreamHandler",
                                "stream": "ext://sys.stdout",
                                "formatter": "json"}},
        "root": {"handlers": ["stdout"], "level": level},
        "loggers": {
            "uvicorn.error": {"handlers": ["stdout"], "level": level, "propagate": False},
            "uvicorn.access": {"handlers": ["stdout"], "level": level, "propagate": False},
            "sqlalchemy.engine": {"handlers": ["stdout"], "level": "WARNING", "propagate": False},
        },
    })

app/main.py · 中间件注入上下文 + 记录耗时

import time
import uuid

import structlog
from flask import Flask, g, request
from prometheus_client import Histogram, Counter, generate_latest, CONTENT_TYPE_LATEST

from .logging_conf import setup_logging

setup_logging()
log = structlog.get_logger("shop.api")

REQUEST_COUNT = Counter(
    "http_requests_total", "请求总数",
    ["method", "endpoint", "status"],
)
REQUEST_LATENCY = Histogram(
    "http_request_duration_seconds", "请求耗时",
    ["method", "endpoint"],
)


def create_app() -> Flask:
    app = Flask(__name__)

    @app.before_request
    def _bind_context():
        # 优先复用 Nginx 传来的 ID，没有就自己生成
        g.request_id = request.headers.get("X-Request-Id") or uuid.uuid4().hex
        g.started_at = time.perf_counter()
        structlog.contextvars.clear_contextvars()
        structlog.contextvars.bind_contextvars(
            request_id=g.request_id,
            method=request.method,
            path=request.path,
        )

    @app.after_request
    def _log_response(resp):
        elapsed = time.perf_counter() - g.started_at
        structlog.contextvars.unbind_contextvars("path")
        log.info(
            "request_finished",
            status=resp.status_code,
            duration_ms=round(elapsed * 1000, 2),
            user_id=getattr(g, "user_id", None),
        )
        endpoint = request.endpoint or "unknown"
        REQUEST_COUNT.labels(request.method, endpoint, resp.status_code).inc()
        REQUEST_LATENCY.labels(request.method, endpoint).observe(elapsed)
        resp.headers["X-Request-Id"] = g.request_id
        return resp

    @app.get("/healthz")
    def healthz():
        """存活探针：进程还在就返回 200。不要在这里查数据库。"""
        return {"status": "ok"}

    @app.get("/readyz")
    def readyz():
        """就绪探针：依赖不可用时返回 503，让负载均衡暂时别往这打流量。"""
        checks = {}
        try:
            import redis
            redis.from_url(app.config["REDIS_URL"]).ping()
            checks["redis"] = "ok"
        except Exception as exc:
            checks["redis"] = f"fail: {exc}"
        healthy = all(v == "ok" for v in checks.values())
        return {"checks": checks, "ready": healthy}, 200 if healthy else 503

    @app.get("/metrics")
    def metrics():
        """Prometheus 抓取端点"""
        return generate_latest(), 200, {"Content-Type": CONTENT_TYPE_LATEST}

    return app


app = create_app()

必须监控的四类指标

连接池饱和度：最容易被忽略的瓶颈

"""把数据库连接池的饱和度暴露成 Prometheus 指标。"""
from prometheus_client import Gauge

POOL_SIZE = Gauge(
    "db_pool_size", "连接池上限",
    ["pool"],
)
POOL_CHECKED_IN = Gauge(
    "db_pool_checked_in", "已借出的连接数",
    ["pool"],
)
POOL_OVERFLOW = Gauge(
    "db_pool_overflow", "溢出创建的连接数",
    ["pool"],
)


def instrument_pool(engine, pool_name: str = "default") -> None:
    """包装 SQLAlchemy 连接池的已借出计数。"""
    POOL_SIZE.labels(pool_name).set(engine.pool.size())
    original = engine.pool._checked_out  # noqa: SLF001 —— 原理演示用途

    class _GaugeProxy:
        def __get__(self, instance, owner=None):
            return original.__get__(instance, owner)

    engine.pool._checked_out = _GaugeProxy()      # noqa: SLF001
    POOL_CHECKED_IN.labels(pool_name).set(0)
    POOL_OVERFLOW.labels(pool_name).set(engine.pool._overflow or 0)


# 判定规则：连接池长期打满 -> 数据库或下游变慢，而不是应用「太忙」

配置、环境变量与密钥管理

代码里出现任何一个硬编码的密码，都是一次安全事故的倒计时。配置管理的铁律是：同一份代码在开发、测试、生产三个环境只差配置，不差代码。

配置分级

用 pydantic-settings 做类型安全的配置

app/config.py

"""集中式配置：环境变量 -> 带类型、可校验、可打印摘要的对象。"""
from functools import lru_cache
from typing import Literal

from pydantic import Field, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",            # 本地开发读文件；生产不挂这个文件
        env_file_encoding="utf-8",
        extra="ignore",
    )

    # ---- 基础 ----
    app_name: str = "shop-api"
    env: Literal["dev", "test", "prod"] = "dev"
    debug: bool = False
    host: str = "0.0.0.0"
    port: int = 8000

    # ---- 数据层 ----
    database_url: SecretStr = Field(
        default=SecretStr("postgresql+psycopg://shop:shop@localhost/shop")
    )
    db_pool_size: int = 10
    db_max_overflow: int = 10
    db_pool_timeout: int = 10        # 拿不到连接就等 10 秒，之后抛 QueuePool 超时
    db_echo: bool = False             # 打印 SQL，生产务必关闭（有性能开销+泄露数据）

    # ---- 缓存 ----
    redis_url: str = "redis://localhost:6379/0"
    cache_ttl_seconds: int = 300

    # ---- 安全 ----
    jwt_secret: SecretStr = SecretStr("dev-only-insecure-secret")
    jwt_expire_minutes: int = 60 * 24
    cors_origins: list[str] = Field(default_factory=list)

    # ---- 业务 ----
    log_level: str = "INFO"
    order_expire_seconds: int = 1800
    feature_new_checkout: bool = False

    @field_validator("cors_origins", mode="before")
    @classmethod
    def _split_origins(cls, v):
        """支持 CORS_ORIGINS=https://a.com,https://b.com 这种写法"""
        if isinstance(v, str):
            return [item.strip() for item in v.split(",") if item.strip()]
        return v

    def safe_summary(self) -> str:
        """打印配置时必须脱敏：SecretStr 只暴露前后 4 个字符。"""
        return (
            f"app={self.app_name} env={self.env} debug={self.debug} "
            f"db={self.database_url.get_secret_value().split('@')[-1]} "
            f"jwt={self.jwt_secret.get_secret_value()[:4]}**** "
            f"cors={self.cors_origins}"
        )


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    """整个进程只解析一次配置。"""
    return Settings()


if __name__ == "__main__":
    print(get_settings().safe_summary())

.env.example · 提交进仓库的只有模板

# 复制成 .env 后填写真实值；.env 必须在 .gitignore 和 .dockerignore 里
ENV=dev
DEBUG=true
PORT=8000

DATABASE_URL=postgresql+psycopg://shop:CHANGE_ME@localhost/shop
DB_POOL_SIZE=5
REDIS_URL=redis://localhost:6379/0

JWT_SECRET=CHANGE_ME_TO_A_LONG_RANDOM_STRING
CORS_ORIGINS=http://localhost:5173

LOG_LEVEL=DEBUG

生成安全密钥的正确姿势

# 用 secrets 而不是 random 生成密钥
python -c "import secrets; print(secrets.token_urlsafe(48))"
# 输出示例：3Qv9-xJ2mKp7dLw5nRt8yZ1aBcDeFgHiJkLmNoPqRsTuVwXyZ012

# JWT / 签名密钥建议至少 32 字节熵
python -c "import secrets; print(secrets.token_hex(32))"

密钥管理方案对比

性能优化：从「感觉慢」到可测量

性能优化必须建立在测量之上。先用工具定位，再动手改，否则只是把代码改得看不懂而速度没变。

定位工具三件套

# 1) cProfile：找出耗时函数（火焰图的输入）
python -m cProfile -o prof.out manage.py runserver
python -m pstats prof.out
# 进入 pstats 后：sort cumtime | head -30；也可敲 "help" 看交互命令

# 2) py-spy：无侵入采样，生产环境可直接 attach
py-spy top --pid 12345                    # 实时看占用 CPU 的栈
py-spy dump --pid 12345                   # 抓一份栈快照（卡顿时最有价值）
py-spy record --pid 12345 -o flame.svg --duration 30   # 生成火焰图
py-spy top --pid 12345 --subprocesses     # 含子进程

# 3) line_profiler：精确到行
pip install line_profiler
kernprof -l -o prof.out script.py
python -m line_profiler prof.out | head -40

热点优化的六个方向

① 算法复杂度：换算法胜过换硬件

"""嵌套循环查存在性 -> 哈希集合，复杂度 O(n*m) 变 O(n+m)。"""
import time

# 反例：列表的 in 是线性查找
def find_pairs_bad(users: list[dict], orders: list[dict]) -> list[tuple]:
    out = []
    for u in users:                    # n 次
        for o in orders:                # × m 次，每次都遍历整个 orders
            if u["id"] == o["user_id"]:
                out.append((u["id"], o["id"]))
    return out

# 正例：先建索引
def find_pairs_good(users: list[dict], orders: list[dict]) -> list[tuple]:
    order_index: dict[int, list[int]] = {}
    for o in orders:
        order_index.setdefault(o["user_id"], []).append(o["id"])
    return [
        (u["id"], oid)
        for u in users
        for oid in order_index.get(u["id"], [])
    ]

import random
random.seed(0)
users = [{"id": i} for i in range(3000)]
orders = [{"id": j, "user_id": random.randrange(5000)} for j in range(3000)]

t0 = time.perf_counter(); find_pairs_bad(users, orders); t_bad = time.perf_counter() - t0
t0 = time.perf_counter(); find_pairs_good(users, orders); t_good = time.perf_counter() - t0
print(f"反例 {t_bad*1000:.1f} ms / 正例 {t_good*1000:.1f} ms / 提速 {t_bad/max(t_good,1e-9):.0f} 倍")

② 数据库：索引 + 批量化 + 避免 N+1

"""N+1 查询是 Web 应用最常见的性能杀手。"""
from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

# 反例：1 次查用户 + N 次查订单
def load_orders_naive(session: Session, user_ids: list[int]):
    orders = []
    for uid in user_ids:                      # 每次循环打一次数据库
        stmt = select(Order).where(Order.user_id == uid)
        orders.extend(session.scalars(stmt).all())
    return orders

# 正例：一条 IN 查询搞定，关联对象用 selectinload 预加载
def load_orders_fast(session: Session, user_ids: list[int]):
    stmt = (
        select(Order)
        .where(Order.user_id.in_(user_ids))   # 一条 SQL
        .options(selectinload(Order.items))   # 关联对象也是批量一条，而不是逐条
    )
    return session.scalars(stmt).all()

# 反例：循环里 commit()，每次都是一次独立事务 + fsync
def save_naive(records: list[dict]):
    for r in records:
        session.add(Model(**r))
        session.commit()                     # 1000 条 = 1000 次磁盘同步
    return len(records)

# 正例：一次事务批量提交
def save_fast(session: Session, records: list[dict]) -> int:
    session.add_all([Model(**r) for r in records])
    session.commit()                         # 一次事务
    return len(records)

# 大批量用 COPY / executemany，比 ORM 快一个数量级
# psycopg 走 copy：
#   with session.connection() as conn, conn.cursor().copy(
#       "COPY orders(user_id, amount) FROM STDIN"
#   ) as cp:
#       for r in records:
#           cp.write_row((r["user_id"], r["amount"]))

③ 缓存：三层结构与失效策略

"""多级缓存 + 显式失效，避免缓存三板斧。"""
import functools
import hashlib
import json
import time
from typing import Any, Callable

from redis import Redis

_redis = Redis.from_url("redis://localhost:6379/0", decode_responses=True)


def cache_key(namespace: str, *parts: Any) -> str:
    raw = json.dumps([str(p) for p in parts], ensure_ascii=False, sort_keys=True)
    digest = hashlib.sha256(raw.encode()).hexdigest()[:16]
    return f"{namespace}:{digest}"


def cached(namespace: str, ttl: int = 300,
           version: int = 1) -> Callable:
    """装饰器式缓存：命名空间 + 版本号，版本号是批量失效的救星。"""
    def deco(fn: Callable) -> Callable:
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            key = f"{namespace}:v{version}:{cache_key(namespace, *args, **kwargs)[len(namespace)+3:]}"
            hit = _redis.get(key)
            if hit is not None:
                return json.loads(hit)
            value = fn(*args, **kwargs)
            try:
                _redis.setex(key, ttl, json.dumps(value, ensure_ascii=False))
            except Exception:                 # 缓存故障绝不能影响主流程
                pass
            return value
        wrapper.invalidate_all = lambda: _redis.delete_pattern(f"{namespace}:v{version}:*")
        wrapper.bump_version = lambda: _redis.incr(f"{namespace}:version")
        return wrapper
    return deco


# 用法
@cached("product", ttl=600, version=3)
def get_product(product_id: int) -> dict:
    print(f"[DB] 查询商品 {product_id}")      # 缓存命中时不会打印
    return {"id": product_id, "name": "机械键盘", "price": 399.0}


for i in range(3):
    print(get_product(7))
# 输出：
# [DB] 查询商品 7
# {'id': 7, 'name': '机械键盘', 'price': 399.0}
# {'id': 7, 'name': '机械键盘', 'price': 399.0}
# {'id': 7, 'name': '机械键盘', 'price': 399.0}   （只查了一次库）


# 进程内二级缓存：省掉网络往返，适合极热的小数据
_local: dict[str, tuple[float, Any]] = {}


def local_cache(key: str, ttl: float = 10.0):
    now = time.monotonic()
    if (item := _local.get(key)) and now - item[0] < ttl:
        return item[1]
    return None

④ 序列化：JSON 换成 orjson / msgpack

"""序列化常常占接口耗时的 10%~30%，换库收益立竿见影。"""
import json
import time

import orjson

payload = {
    "items": [{"id": i, "name": f"商品{i}", "tags": ["a", "b"]} for i in range(500)]
}

t0 = time.perf_counter()
for _ in range(200):
    std = json.dumps(payload, ensure_ascii=False)
t_std = time.perf_counter() - t0

t0 = time.perf_counter()
for _ in range(200):
    fast = orjson.dumps(payload)           # 默认输出 bytes，不转义非 ASCII
t_fast = time.perf_counter() - t0

print(f"json  {t_std*1000:.1f} ms / orjson {t_fast*1000:.1f} ms")
# 预期输出：json 1200.0 ms / orjson 30.0 ms  （orjson 通常快 5~10 倍）

# FastAPI 里直接换掉序列化器
# from fastapi.responses import ORJSONResponse
# app = FastAPI(default_response_class=ORJSONResponse)

⑤ 启动与运行时开销

import os
import time

t0 = time.perf_counter()
import numpy as np                     # 单独测量重型库导入
print(f"numpy 导入耗时 {time.perf_counter()-t0:.2f}s")   # 常见 0.2~0.8s

# 结论一：不要在模块顶层导入只在校验路径用到的重库（如 pandas、torch）
# 结论二：真正的优化手段是 gunicorn preloading + 容器内不做重复导入

# 多进程下的 GIL：CPU 密集要靠多进程，不能靠加线程
import multiprocessing as mp


def cpu_heavy(n: int) -> int:
    return sum(i * i for i in range(n))


if __name__ == "__main__":
    with mp.Pool(processes=mp.cpu_count()) as pool:
        t0 = time.perf_counter()
        pool.map(cpu_heavy, range(4) * os.cpu_count())
        print(f"多进程耗时 {time.perf_counter()-t0:.2f}s")

⑥ 容器层面的性能选项

# 启动快一点：设置容器可见 CPU 数与亲和性（降低调度开销）
docker run -e WEB_CONCURRENCY=4 --cpuset-cpus=0-3 shop-api:1.4.0

# 减少重启带来的冷启动：uvloop + httptools 必须在 pip 安装时就带上
pip install "uvicorn[standard]"

# 明确 Python 内存回收阈值（Gunicorn 官方推荐的环境变量组合）
export MALLOC_ARENA_MAX=2          # 限制 glibc arena 数量，显著降低多进程 RSS
export PYTHONMALLOC=malloc

# 查看实际内存占用
docker stats --no-stream
docker exec shop-api-1 cat /sys/fs/cgroup/memory.current

CI/CD：让发布变成一条命令

持续集成（Continuous Integration，CI）解决「代码能不能合」，持续交付/部署（Continuous Delivery / Deployment，CD）解决「能不能自动上线」。对 Python 项目，一条合格的流水线至少包含：静态检查 → 单元测试 → 覆盖率 → 构建镜像 → 安全扫描 → 推仓库 → 部署 → 冒烟验证。

.github/workflows/ci.yml

name: CI

on:
  push:
    branches: [main]
  pull_request:

concurrency:                        # 同一分支的新提交自动取消旧流水线
  group: ci-${{ github.ref }}
  cancel-in-progress: true

jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: pip                     # GitHub 侧缓存 pip 下载
      - name: 安装依赖
        run: |
          python -m pip install --upgrade pip uv
          uv sync --group dev
      - name: 代码规范（ruff）
        run: uv run ruff check .
      - name: 格式化检查
        run: uv run ruff format --check .
      - name: 静态类型（mypy）
        run: uv run mypy app

  test:
    runs-on: ubuntu-latest
    services:                          # GitHub Actions 的服务容器
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_USER: shop
          POSTGRES_PASSWORD: shop
          POSTGRES_DB: shop_test
        ports: ["5432:5432"]
        options: >-                       # 等健康再跑测试
          --health-cmd pg_isready
          --health-interval 5s
          --health-timeout 5s
          --health-retries 10
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.12", cache: pip}
      - run: python -m pip install uv && uv sync --group dev
      - name: 运行测试并出覆盖率
        env:
          DATABASE_URL: postgresql+psycopg://shop:shop@localhost:5432/shop_test
          JWT_SECRET: test-secret-for-ci-only
        run: |
          uv run pytest -q --cov=app --cov-report=term-missing \
                          --cov-report=xml --junitxml=report.xml
      - uses: actions/upload-artifact@v4
        if: always()
        with:
          name: test-report
          path: |
            coverage.xml
            report.xml

  build-and-push:
    needs: [quality, test]              # 前两步都过才构建
    if: github.ref == "refs/heads/main"
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write                   # 推 GHCR 所需
    steps:
      - uses: actions/checkout@v4
      - uses: docker/setup-buildx-action@v3
      - uses: docker/login-action@v3
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/metadata-action@v5      # 自动生成标签与镜像说明
        id: meta
        with:
          images: ghcr.io/${{ github.repository }}/api
          tags: |
            type=sha,format=long
            type=ref,event=branch
            type=semver,pattern={{version}}
            type=raw,value=latest,enable={{is_default_branch}}
      - uses: docker/build-push-action@v6
        with:
          context: .
          target: runtime
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha
          cache-to: type=gha,mode=max       # 跨流水线复用构建缓存

  security:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: 依赖漏洞扫描
        run: |
          python -m pip install pip-audit
          pip-audit --strict || echo "存在漏洞，需要评估"
      - uses: aquasecurity/trivy-action@0.28.0
        with:
          scan-type: image
          image-ref: ghcr.io/${{ github.repository }}/api:latest
          severity: CRITICAL,HIGH
          exit-code: "1"

本地与 CI 结果不一致怎么办

定时任务与后台作业

不是所有事情都适合放在 Web 请求里。「每天凌晨汇总一次报表」「每 5 分钟同步外部数据」这类需求，如果写成用户请求触发，会遇到并发重复执行、请求超时、失败无法重试三个问题。

三种方案的选型

crontab 版：最小可用示例

# app/jobs/daily_report.py —— 每天生成日报
import asyncio
import logging
from datetime import datetime
from pathlib import Path

log = logging.getLogger("jobs.daily")


async def build_daily_report() -> str:
    """模拟：查询昨日订单 -> 聚合 -> 写文件并（真实场景）推送到企业微信。"""
    from app.db import session_scope
    from app.models import Order

    yesterday = datetime.now().date()
    with session_scope() as session:
        rows = session.execute(
            select(Order.amount).where(Order.created_at < yesterday)
        ).scalars().all()

    total = sum(rows)
    content = f"日期: {yesterday}\n订单数: {len(rows)}\n总金额: {total:.2f}\n"
    out = Path("reports") / f"{yesterday}.txt"
    out.parent.mkdir(exist_ok=True)
    out.write_text(content, encoding="utf-8")
    log.info("日报生成完成 path=%s orders=%d", out, len(rows))
    return str(out)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(build_daily_report())

# 部署（容器内跑 crontab 的注意点）
# 1) 容器主进程不是 cron 时，用一个常驻进程 + 子进程的方式
# 2) 环境变量在 cron 里默认没有，必须显式声明
# 3) 容器重启会重置 crontab（除非挂载文件）
# 4) 用 flock 防止上一次没跑完就叠加下一次

# crontab 文件（挂载进容器 /etc/cron.d/jobs）
SHELL=/bin/bash
PATH=/usr/local/bin:/usr/local/sbin:/usr/sbin:/usr/bin:/sbin:/bin
ENV_FILE=/app/.env

# 每天 02:17 执行（避开整点的资源高峰）
17 2 * * * root set -a && . $ENV_FILE && set +a && \
    /usr/local/bin/python -m app.jobs.daily_report >> /proc/1/fd/1 2>&1

# 每 5 分钟同步一次，带锁
*/5 * * * * root flock -n /tmp/sync.lock -c 'python -m app.jobs.sync_data' >> /proc/1/fd/1 2>&1

# 重启容器时的补跑（@reboot 只在开机跑一次）
@reboot root sleep 60 && python -m app.jobs.reconcile >> /proc/1/fd/1 2>&1

APScheduler 版：单进程内的定时任务

"""APScheduler：适合单实例服务，注意多副本重复执行的问题。"""
import logging
from datetime import datetime, timedelta

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from redis import Redis

from app.jobs.daily_report import build_daily_report

log = logging.getLogger("scheduler")
_redis = Redis.from_url("redis://localhost:6379/1")
LOCK_KEY = "lock:daily-report"


def with_redis_lock(func):
    """分布式锁：多副本时只让一个实例真正执行任务。"""
    def wrapper(*args, **kwargs):
        acquired = _redis.set(LOCK_KEY, "1", nx=True, ex=1800)
        if not acquired:
            log.info("任务已在其它实例执行，本次跳过")
            return None
        try:
            return func(*args, **kwargs)
        finally:
            _redis.delete(LOCK_KEY)
    return wrapper


def start_scheduler() -> BackgroundScheduler:
    sched = BackgroundScheduler(
        timezone="Asia/Shanghai",
        job_defaults={"coalesce": True,   # 错过的多次触发只补一次
                       "max_instances": 1, # 同一任务不并发
                       "misfire_grace_time": 300},
    )
    sched.add_job(
        with_redis_lock(build_daily_report),
        CronTrigger(hour=2, minute=17),
        id="daily_report",
        replace_existing=True,
    )
    sched.add_job(
        sync_partner_prices,
        IntervalTrigger(minutes=5),
        id="sync_prices",
        replace_existing=True,
    )
    sched.start()
    log.info("调度器已启动，共 %d 个任务", len(sched.get_jobs()))
    return sched


# 优雅停机：先等正在跑的任务结束再退出
if __name__ == "__main__":
    scheduler = start_scheduler()
    try:
        import time
        while True:
            time.sleep(60)
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown(wait=True)

Celery Beat 版：可观测、可重试的任务队列

# tasks.py
from celery import Celery
from celery.schedules import crontab

celery_app = Celery("shop", broker="redis://cache:6379/2", backend="redis://cache:6379/3")
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="Asia/Shanghai",
    task_acks_late=True,             # 任务执行完成才 ack：worker 崩溃会重投
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,    # 长任务必须设为 1，否则一个 worker 囤积任务
    task_track_started=True,
    task_time_limit=600,             # 硬超时
    task_soft_time_limit=540,        # 软超时（抛 SoftTimeLimitExceeded）
    result_expires=3600,
    task_default_retry_delay=30,
    task_routes={"shop.tasks.report.*": {"queue": "heavy"}},
    beat_schedule={
        "daily-report": {
            "task": "shop.tasks.report.daily",
            "schedule": crontab(hour=2, minute=17),
        },
        "clear-sessions": {
            "task": "shop.tasks.maintenance.clear_sessions",
            "schedule": crontab(minute=0, hour="*/6"),
        },
    },
)


@celery_app.task(name="shop.tasks.report.daily", bind=True, max_retries=3)
def daily_report(self) -> dict:
    """报告类任务：网络易失败，必须可重试且幂等。"""
    import logging
    log = logging.getLogger("celery.daily")
    try:
        path = build_daily_report_sync()
        log.info("日报完成 path=%s", path)
        return {"status": "ok", "path": path, "attempt": self.request.retries}
    except (ConnectionError, TimeoutError) as exc:
        log.warning("日报失败，第 %d 次重试: %s", self.request.retries + 1, exc)
        raise self.retry(exc=exc, countdown=2 ** self.request.retries * 30)
    except Exception:
        log.exception("日报遇到不可重试的错误")
        raise


# 幂等性：任务可能重复执行，写操作必须自带去重
from redis import Redis
_r = Redis.from_url("redis://cache:6379/2")


@celery_app.task(name="shop.tasks.mail.welcome")
def send_welcome_mail(user_id: int) -> None:
    dedup_key = f"task:welcome:{user_id}"
    if not _r.set(dedup_key, "1", nx=True, ex=86400):
        return                       # 24 小时内已发过，直接跳过
    _r.hincrby("metrics:welcome_sent", "count", 1)
    send_mail(user_id)

# worker 与 beat 启动
celery -A tasks.celery_app worker -l info -Q default,heavy --concurrency=4
celery -A tasks.celery_app beat --loglevel=info

# 常用运维命令
celery -A tasks.celery_app inspect active        # 看正在执行的任务
celery -A tasks.celery_app inspect registered    # 看注册了哪些任务
celery -A tasks.celery_app inspect active_queues # 看队列积压
celery -A tasks.celery_app purge -f              # 清空队列（危险）
celery -A tasks.celery_app call shop.tasks.report.daily   # 手动触发一次

# 监控积压：队列长度是必配告警项
redis-cli -n 2 llen celery          # 待处理任务数，应 < 1000
redis-cli -n 2 info | grep -E "used_memory_human|evicted_keys"

优雅停机、健康检查与发布策略

服务的可用性往往不是靠「永不崩溃」，而是靠「崩溃了能快速自愈，且重启过程对用户无感」。

优雅停机的完整链路

1. 收到 SIGTERM：docker stop、K8s 驱逐、systemd 停止都会发这个信号。
2. Nginx 摘流量：反代侧停止把新请求转进来（K8s 里体现为从 endpoints 移除，有 preStop 延迟）。
3. 应用停止接受新请求：gunicorn 收到 SIGTERM 后立刻停止 accept，等待存量请求处理完。
4. 超时保护：超过 graceful_timeout（如 30 秒）仍未结束则强制杀掉。
5. 容器退出：因为写了 STOPSIGNAL SIGTERM，Docker 才发对信号（默认发 SIGQUIT 会导致 core dump）。

# 关键：让 gunicorn 收到 SIGTERM 而不是 SIGKILL
# Dockerfile 里显式声明
STOPSIGNAL SIGTERM

# compose 里给足停止宽限期（默认 10 秒往往不够）
services:
  api:
    stop_signal: SIGTERM
    stop_grace_period: 40s

# 代码里注册退出钩子：关闭连接池、flush 日志、落盘状态
import signal
import threading

_shutdown_event = threading.Event()


def register_shutdown(app):
    def handler(signum, frame):
        app.logger.info("收到信号 %s，开始优雅停机", signum)
        _shutdown_event.set()

    signal.signal(signal.SIGTERM, handler)
    signal.signal(signal.SIGINT, handler)


# 在 lifespan 里等待停机信号，把内存中的待写数据落盘
async def lifespan(app):
    register_shutdown(app)
    yield                     # 应用正常启动完成
    # ---- 以下是关闭阶段 ----
    engine.dispose()          # 归还数据库连接
    await cache_client.aclose()
    app.logger.info("资源已释放，准备退出")

四种发布策略

一份可直接用的发布脚本

#!/usr/bin/env bash
# deploy.sh —— 零停机发布的最小可用实现
set -Eeuo pipefail                       # 任何命令失败立即退出

APP=shop-api
COMPOSE_FILE="compose.yaml -f compose.prod.yaml"
IMAGE="ghcr.io/acme/shop-api:${GIT_SHA}"

log()  { printf '\033[36m[%s] %s\033[0m\n' "$(date +%H:%M:%S)" "$*"; }
fail() { printf '\033[31m[%s] 失败: %s\033[0m\n' "$(date +%H:%M:%S)" "$*" >&2; exit 1; }
trap 'fail "第 $LINENO 行出错"' ERR

log "1/5 拉取镜像 ${IMAGE}"
docker compose ${COMPOSE_FILE} pull api || fail "拉取失败"

log "2/5 记录当前版本（用于回滚）"
CURRENT=$(docker inspect -f '{{.Config.Image}}' "${APP}-api-1" 2>/dev/null || echo "none")
echo "${CURRENT}" > /tmp/shop-api.previous
log "当前版本: ${CURRENT}"

log "3/5 启动新版本（此时流量仍在旧实例）"
IMAGE_TAG="${GIT_SHA}" docker compose ${COMPOSE_FILE} up -d --no-deps api

log "4/5 等待新实例健康"
for i in $(seq 1 30); do
    status=$(docker inspect -f '{{.State.Health.Status}}' "${APP}-api-1" 2>/dev/null || echo "starting")
    case "${status}" in
        healthy)   log "健康检查通过"; break ;;
        unhealthy) fail "新版本不健康，自动回滚"; ;;
    esac
    [[ $i -eq 30 ]] && fail "等待超时（30 次 × 5 秒）"
    log "  状态=${status}，${i}/30"
    sleep 5
done

log "5/5 冒烟测试"
code=$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:8000/healthz)
[[ "${code}" == "200" ]] || fail "健康端点返回 ${code}"
docker compose ${COMPOSE_FILE} exec -T api python -m app.jobs.verify_db

log "发布完成，新版本 ${IMAGE}"

#!/usr/bin/env bash
# rollback.sh —— 一条命令回退
set -Eeuo pipefail
PREV=$(cat /tmp/shop-api.previous 2>/dev/null) || { echo "没有回滚记录"; exit 1; }
echo "回滚到: ${PREV}"
IMAGE_TAG="${PREV##*:}" docker compose compose.yaml -f compose.prod.yaml up -d --no-deps api
sleep 10
curl -fsS http://127.0.0.1:8000/healthz && echo "回滚完成"

备份、安全与故障演练

备份：三个 3 策略

# 数据库备份与恢复
pg_dump -h db -U shop -Fc -f /backup/shop-$(date +%F).dump shop
pg_dumpall -h db -U postgres -f /backup/globals-$(date +%F).sql   # 角色与权限

# 保留策略：7 天内每日、4 周内每周
find /backup -name "shop-*.dump" -mtime +7 -delete

# 恢复到新库（先建库再恢复）
createdb -h db -U postgres shop_restore
pg_restore -h db -U shop -d shop_restore --clean --if-exists /backup/shop-2026-01-15.dump

# 全库恢复演练脚本（每月自动跑一次，验证备份真的能用）
set -e
TS=$(date +%s)
createdb -h db -U postgres "restore_test_${TS}"
pg_restore -h db -U shop -d "restore_test_${TS}" /backup/latest.dump
psql -h db -U shop -d "restore_test_${TS}" -c "SELECT count(*) FROM orders;"
dropdb -h db -U postgres "restore_test_${TS}"
echo "备份恢复演练通过"

安全基线清单

一次完整的故障演练（混沌工程入门）

# 演练 1：杀掉一个 worker，看是否影响服务
docker compose exec api sh -c 'kill -9 $(pgrep -f "gunicorn" | tail -1)'
# 预期：Nginx 对这一个 worker 的在途请求可能 502 一次，但整体可用

# 演练 2：断开数据库，看就绪探针是否正确降级
docker compose stop db
curl -i http://127.0.0.1:8000/readyz     # 预期 503，且 /healthz 仍 200
curl -s http://127.0.0.1:8000/healthz
docker compose start db

# 演练 3：内存泄漏验证 max_requests 阀门
for i in $(seq 1 5000); do curl -s http://127.0.0.1:8000/ >/dev/null; done
docker stats --no-stream    # 观察内存是否稳定

# 演练 4：磁盘写满（最容易忽略的故障）
df -h
docker system df
# 防护：日志 max-size + 定期清理 + 监控磁盘使用率 > 80% 告警

# 演练 5：证书过期
openssl x509 -enddate -noout -in /etc/nginx/certs/fullchain.pem
# 防护：到期前 30/14/7 天告警 + certbot 自动续期钩子

本章小结

• 部署链路：DNS/CDN → Nginx → gunicorn → 应用 → 数据库/Redis → 观测。先在纸上画清每一层，再决定技术选型。
• 依赖固化：用 pyproject.toml 声明、锁文件精确固定；构建镜像用锁文件，区分运行时/构建时/开发时三类依赖。
• Docker 分层：指令顺序决定缓存命中；依赖清单先于代码 COPY；多阶段构建把编译工具链留在 builder 层，运行镜像瘦到 150 MB 级别；.dockerignore 必写。
• 镜像标签不可变：用 Git SHA 或语义化版本，永不覆盖 latest；镜像上打 OCI 标签记录来源。
• Compose：depends_on 只管顺序，配合 healthcheck + service_healthy 才管可用性；命名卷持久化数据；down -v 会删数据。
• Nginx + gunicorn：TLS 与静态资源在 Nginx，动态请求走 gunicorn；gthread 适合 IO 密集，max_requests 防内存泄漏；X-Request-Id 贯穿全链路。
• 可观测性：结构化 JSON 日志（含 request_id 与 traceback）+ 四类黄金指标（流量/错误/延迟/饱和）+ 就绪与存活探针分离。
• 配置与密钥：pydantic-settings 集中管理、SecretStr 防误打印、.env.example 入库而 .env 不入库；密钥用 secrets 生成。
• 性能优化：先测量（cProfile / py-spy）再改；六个方向——算法复杂度、N+1 与批量提交、多级缓存、序列化库、启动开销、容器参数。
• CI/CD：lint → 类型 → 测试 → 构建推送 → 安全扫描；并发组取消旧流水线；构建缓存跨流水线复用。
• 定时任务：crontab 最简、APScheduler 需分布式锁、Celery Beat 可观测可重试；长任务 prefetch=1，重试任务必须幂等。
• 发布：STOPSIGNAL SIGTERM + 足够 stop_grace_period 实现优雅停机；滚动/蓝绿/金丝雀按回滚要求选型。
• 备份与演练：3-2-1 原则，恢复演练自动化；没验证过的备份等于没有备份。

练习题

1. 把第 24 章的全栈应用容器化：为 Flask + PostgreSQL 的订单服务写出完整的 pyproject.toml、锁文件、Dockerfile（多阶段、slim 基础、非 root、healthcheck）、.dockerignore、compose.yaml（含 db、redis、api、nginx 四个服务）。要求：① 镜像体积控制在 250 MB 内；② 代码改动后重建镜像只重装最后一层；③ docker compose up -d 后访问 http://localhost/healthz 返回 200。
   参考答案要点Dockerfile 顺序必须是 WORKDIR → COPY pyproject.toml requirements.lock ./ → RUN pip install → COPY app ./app；db 服务加 healthcheck: pg_isready，api 的 depends_on.db.condition=service_healthy；ports 只把 80 暴露到宿主机，8000 绑 127.0.0.1；logging.options.max-size 必须设。
2. 解决缓存问题：某商品详情接口 P99 达到 2.4 秒，压测显示数据库 QPS 被缓存未命中打满。请用「多级缓存 + 布隆过滤器 + 随机 TTL」改造，要求：① 用 @cached 装饰器实现 Redis 缓存；② 对不存在的商品 ID 用布隆过滤器提前拦截，避免缓存穿透；③ TTL 加 ±10% 随机抖动防雪崩；④ 热点商品用 single-flight 只让一个请求回源。给出代码与压测前后对比结论。
   参考答案要点布隆过滤器：bitarray 或 hashlib.sha256 取前 k 位映射到 bytearray(1<<20)，may_contain(id) 为 False 直接返回「不存在」且不查库；single-flight 用 Redis set(nx=True, ex=5) 抢锁，未抢到则 sleep(0.05) 后重读缓存；TTL random.randint(270, 330)；空结果缓存 ttl=30。结论应包含命中率从 62% → 96%、DB QPS 下降约 3.7 倍。
3. 写出可观测性三件套：给订单服务补齐 ①structlog 结构化日志（每条请求日志含 request_id、user_id、endpoint、status、duration_ms）；②Prometheus 四类指标（QPS Counter、5xx Counter、耗时 Histogram、连接池饱和 Gauge）；③/healthz 与 /readyz 两个端点，并写出一段 SQL/Grafana 查询语句统计 P99。
   参考答案要点Histogram(..., buckets=(0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0)) 才能算出 P95/P99；P99 查询：histogram_quantile(0.99, sum by (le) (rate(http_request_duration_seconds_bucket[5m])))；5xx 比例：sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m]))；/readyz 中查 Redis 与 DB，失败返回 503；打点时用 request.endpoint 而非原始 path，避免路径参数造成标签基数爆炸。
4. 定时任务改造：把「每 10 分钟同步第三方商品价格」从 crontab 迁移到 Celery Beat。要求：① 同步 1000 个商品且单次失败只重试失败的那一批；② 任务整体幂等（重复执行不产生重复数据）；③ 队列积压超过 500 时告警；④ 任务执行超过 4 分钟自动中止。给出代码与 alert 规则表达式。
   参考答案要点celery beat schedule: crontab(minute="*/10")；主任务把商品 id 切成 100 个一批，每批一个子任务 autoretry_for=(ConnectionError,)，max_retries=5，retry_backoff=True；幂等用 INSERT ... ON CONFLICT (sku) DO UPDATE SET price=EXCLUDED.price；task_time_limit=240、task_soft_time_limit=230；worker_prefetch_multiplier=1；告警表达式 redis_len_celery > 500（Prometheus exporter 采集 LLEN celery），for 持续 5m 触发。
5. 零停机发布演练：为你的服务写一份 deploy.sh，实现：拉取新镜像 → 启动 → 等待 healthcheck 通过（最多 90 秒）→ 冒烟测试 → 失败自动回滚到上一版本 → 成功后打印版本号。要求脚本幂等，可在 CI 中直接调用。手工模拟一次失败（用一个必挂的 healthcheck）验证回滚生效。
   参考答案要点发布前把当前镜像 tag 写入 /tmp/shop-api.previous；用 docker inspect -f '{{.State.Health.Status}}' 轮询健康状态，循环 18 次 × 5 秒；set -Eeuo pipefail + trap ... ERR 把任意失败导向回滚函数；回滚函数读 previous 文件并重新 up -d；验证手段：临时把 healthcheck 改成 curl -f http://localhost:8000/__never_ok，观察脚本 90 秒内自动回滚且访问恢复 200。
6. 安全加固：审计并加固你的服务。找出并修复：① 硬编码密钥；② 以 root 运行容器；③ 数据库端口暴露公网；④ 缺失的安全响应头；⑤ 日志打印了用户手机号。给出修复前后的配置对比与验证命令。
   参考答案要点①密钥改环境变量 + SecretStr，并用 git log -p -S 'sk-' 扫描历史，泄露则必须轮转密钥；②Dockerfile 加 RUN useradd -u 10001 appuser 与 USER appuser，验证 docker inspect --format '{{.Config.User}}' img；③Compose 中 db 不写 ports，安全组只放通内网；④Nginx 加 add_header Strict-Transport-Security、X-Content-Type-Options、X-Frame-Options，验证 curl -I https://...；⑤加日志脱敏中间件（正则匹配手机号/身份证/卡号并替换为掩码），或引入 structlog.processors 自定义 scrub。
7. 性能定位实战：给你一个「偶发卡顿 3 秒」的 Flask 接口，要求用 py-spy 抓栈定位到具体函数，输出「火焰图 + 根因结论 + 修复方案」。请写出完整的排查命令序列（含 GIL 等待判定）与三种可能的根因（连接池等待 / 外部 API 超时 / 慢查询）及各自的验证手段。
   参考答案要点命令序列：py-spy top --pid $(pgrep -f gunicorn | head -1) 观察 3 秒；卡顿时 py-spy dump --pid PID 抓快照；多次复现用 py-spy record --pid PID --duration 60 -o flame.svg。根因判别：栈停在 pymysql/psycopg execute → 数据库侧看 pg_stat_statements 或慢日志；栈停在 socket.recv 且 GIL 占比低 → 外部 API 超时，调 timeout + 连接池 + 熔断；栈停在 queue.get 的 pool._do_get → 连接池打满（pool_size=workers×threads 原则），调池大小或给慢查询加缓存。GIL 判定：火焰图中大量 _bootstrap → take_gil 说明 CPU 密集，需多进程。
8. 综合实战：把教程里的四章内容串成一条流水线：把第 20 章的数据分析、第 22 章的测试、第 23 章的 CLI、第 25 章的部署整合成一个「销售数据日报系统」：CLI 负责拉数与计算、产出 CSV 与图、pytest 保证核心算法正确、Docker 镜像可一键运行、GitHub Actions 自动测试并构建镜像、Compose 中配 PostgreSQL 与定时任务生成日报。要求提交：Dockerfile、compose.yaml、ci.yml、定时任务模块、以及一份 README.md（含架构图 ASCII 版、部署步骤、故障排查清单）。
   参考答案要点目录结构建议 src/daily_report/{cli.py,compute.py,plot.py,jobs.py,config.py} + tests/ + Dockerfile + .github/workflows/ci.yml；compute 层纯函数便于单测（给 3 组已知数据断言结果）；plot 层 matplotlib.use("Agg") 保证容器内无显示也能出图；jobs 层 @celery.task 且用 Redis set(nx=True) 幂等；README 的排查清单至少包含「容器起不来 / 连不上数据库 / 内存持续上涨 / 任务不执行 / 磁盘满」五类，每类给出 1~2 条排查命令。