我做过好几个从0到1的小工具，真正被做成“产品”的只有一个：一个把分散在各处的数据自动拉齐、计算指标并对外提供稳定接口的小服务。它用了两样朴素但强有力的东西——Python 和 Docker。本文不谈华丽词藻，尽量把这件事复原：是什么问题逼出来这个产品，我怎么用 Python 和 Docker 把问题解决到足够“可复制”，以及过程中真实踩到的坑。希望读完，你不仅知道方案，更知道背后的取舍。

一、从问题出发：不是写个脚本，而是做个能“活”的东西
我在一个互联网团队里做数据和平台，一开始我们只是靠几段 Python 脚本加手工操作，把运营需要的指标每早十点刷出来。用着用着问题成群出现：

• 数据延迟不稳定。十点有时候能出，有时候拖到十点半。
• 环境不一致。A 同事机器是 Python 3.10，B 同事是 3.11；依赖库版本各不相同，时不时报“我这边复现不了”。
• 脚本难以托管。谁启动的脚本？出错谁看日志？是否重试？怎么告警？
• 接口不可复用。每次给新系统接入都要改脚本，重复劳动多。

这不是一两个“更聪明的脚本”能解决的，它需要被产品化：可重复交付、标准化配置、可观测、可维护。这就是我把它变成一个小型 Python 服务并用 Docker 固化的原因。

二、产品定位：小而稳，先把“可依赖”做出来
我给自己写了一个简化版 PRD：

• 核心价值：每日按时产出指标，提供可查询接口，并在数据异常时主动告警。
• 使用者画像：运营、数据分析、后端同事。
• 最小可行能力：
  1. 定时抽取多源数据，完成计算。
  2. 暴露 REST API 和 /metrics 监控端点。
  3. 出错自动重试 + 严格超时 + 结构化日志。
  4. 一条命令可部署，环境一致，十分钟内可回滚。
• 非目标：不追求“大而全”的数据平台，不搞复杂编排，先跑稳一个域。

这份“PRD”反过来约束了技术实现，避免我一上来就做成“大工程”。

三、技术选型与架构骨架
语言与库：

• Python 3.11：异步性能提升，生态成熟。
• FastAPI：轻量 REST 框架，Pydantic 校验够用。
• SQLAlchemy + asyncpg：操作 PostgreSQL。
• httpx：异步 HTTP 客户端，易于超时和重试。
• APScheduler：定时任务、灵活调度。
• structlog 或 logging + JSON formatter：结构化日志，方便 ELK。
• pydantic-settings：环境变量驱动配置。
• Alembic：数据库迁移。
• backoff：指数退避重试。

组件关系（简述）：

• Collector 模块：从第三方接口/内部服务拉原始数据。
• Compute 模块：指标计算，幂等，支持批量。
• Store 模块：读写 PostgreSQL，必要时配合 Redis 做缓存或分布式锁。
• API 模块：查询指标、健康检查、Prometheus 指标。
• Scheduler 模块：定时调度 Collector/Compute，带单实例锁防重入。
• Observability：日志、metrics、trace 三件套。

四、目录结构与关键代码片段
应用目录（简化）：

配置与启动：

app/core/config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
ENV: str = “prod”
DB_DSN: str
REDIS_URL: str | None = None
API_TOKEN: str
SENTRY_DSN: str | None = None
TZ: str = “Asia/Shanghai”
LOG_LEVEL: str = “INFO”
REQUEST_TIMEOUT: float = 10.0

class Config:
    env_file = ".env"

settings = Settings()

结构化日志：

app/core/logging.py

import logging, json, sys, os

class JsonFormatter(logging.Formatter):
def format(self, record):
payload = {
“level”: record.levelname,
“msg”: record.getMessage(),
“logger”:
“time”: self.formatTime(record, “%Y-%m-%dT%H:%M:%S”),
}
if record.exc_info:
payload[“exc_info”] = self.formatException(record.exc_info)
return json.dumps(payload, ensure_ascii=False)

def setup_logging():
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JsonFormatter())
root = logging.getLogger()
root.setLevel(os.getenv(“LOG_LEVEL”,“INFO”))
root.handlers.clear()
root.addHandler(handler)

API 入口：

app/main.py

import uvicorn
from fastapi import FastAPI, Depends, Header, HTTPException
from app.core.logging import setup_logging
from app.core.config import settings
from app.services.store import query_metric
from prometheus_client import Counter, generate_latest

setup_logging()
app = FastAPI(title=“Metrics Service”)

REQUESTS = Counter(“api_requests_total”, “Total API requests”)

def auth(x_token: str = Header(…)):
if x_token != settings.API_TOKEN:
raise HTTPException(401, “Unauthorized”)

@app.get(“/metrics/raw”)
def metrics_raw():
return Response(generate_latest(), media_type=“text/plain”)

@app.get(“/metric/{name}”, dependencies=[Depends(auth)])
def metric(name: str, ts: str | None = None):
REQUESTS.inc()
return {“name”: name, “value”: query_metric(name, ts)}

if name == “main”:
uvicorn.run(app, host=“0.0.0.0”, port=8000)

定时任务与单实例锁：

app/jobs/scheduler.py

import os, asyncio, socket, time
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from redis.asyncio import Redis
from app.jobs.tasks import run_pipeline

async def try_lock(redis: Redis, key: str, ttl=3000):
# 简化分布式锁，避免重复执行
val = f"{socket.gethostname()}:{os.getpid()}:{time.time()}"
ok = await redis.set(key, val, px=ttl, nx=True)
return ok

async def job_daily(redis: Redis):
if await try_lock(redis, “lock:pipeline:daily”, ttl=60_000):
await run_pipeline()
else:
# 另一实例在跑，跳过
pass

def start_scheduler():
scheduler = AsyncIOScheduler(timezone=“Asia/Shanghai”)
scheduler.add_job(lambda: asyncio.create_task(job_daily(Redis.from_url(settings.REDIS_URL))),
“cron”, hour=9, minute=30)
scheduler.start()

核心任务：

app/jobs/tasks.py

import asyncio, httpx, logging
from app.services.cmpute import compute_metrics
from app.services.store import save_metrics

log = logging.getLogger(“tasks”)

async def run_pipeline():
async with httpx.AsyncClient(timeout=10.0) as client:
# 采集

    # 计算
    metrics = compute_metrics(data1, data2)
    # 入库
    await save_metrics(metrics)
    log.info("pipeline done", extra={"count": len(metrics)})

数据库迁移：
使用 Alembic 管理表结构演进，每次发布先执行 migrate，确保正向兼容一段时间，避免“先发代码后改表”导致回滚困难。

五、容器化：让环境一致且可复制
我对 Docker 的要求只有两点：镜像小、行为可预期。很多人把 Docker 打成 docke，但打错无所谓，重要的是两件事：构建过程可重复，运行过程可观测。

Dockerfile（多阶段构建，开启 BuildKit 缓存）：

syntax=docker/dockerfile:1.7

FROM python:3.11-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1
PYTHONUNBUFFERED=1
PIP_DISABLE_PIP_VERSION_CHECK=1
PIP_NO_CACHE_DIR=1

RUN --mount=type=cache,target=/var/cache/apt
apt-get update && apt-get install -y --no-install-recommends
build-essential curl ca-certificates tzdata &&
rm -rf /var/lib/apt/lists/*

WORKDIR /app

FROM base AS builder
RUN --mount=type=cache,target=/root/.cache/pip pip install -U pip wheel setuptools
COPY pyproject.toml poetry.lock ./
RUN --mount=type=cache,target=/root/.cache/pip pip install poetry && poetry export -f requirements.txt --output requirements.txt --without-hashes
RUN --mount=type=cache,target=/root/.cache/pip pip wheel -r requirements.txt -w /wheels

FROM base AS runtime

安全：非 root 用户

RUN useradd -r -u 10001 appuser
COPY --from=builder /wheels /wheels
RUN --mount=type=cache,target=/root/.cache/pip pip install --no-index --find-links=/wheels /wheels/*
COPY app ./app
ENV TZ=Asia/Shanghai
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=3s --retries=3 CMD curl -fsS
ENTRYPOINT [“python”,“-m”,“app.main”]

几点解释：

• 多阶段构建把编译依赖留在 builder，runtime 更干净。
• 使用 BuildKit 的 cache mount 提升依赖构建速度。
• 关闭 root，降低容器逃逸风险；配合只读文件系统更稳。
• 明确健康检查，便于编排器探针判定。
• 不用 Alpine。对 Python 而言，musl 带来的各种 C 扩展兼容和性能问题得不偿失，slim 更可预期。

开发环境 docker-compose.yml（简要）：
services:
api:
build: .
image: my-metrics:latest
env_file: .env
ports: [“8000:8000”]
depends_on: [db, redis]
restart: unless-stopped
db:
image: postgres:15
environment:
POSTGRES_PASSWORD: example
volumes:
- pgdata:/var/lib/postgresql/data
redis:
image: redis:7
volumes:
pgdata:

注意事项：

• 本地开发可以把代码挂载成只读卷，热重载用 uvicorn --reload。
• 生产禁用 reload，镜像唯一事实来源，配置通过环境变量注入。
• 时区与编码保持一致，避免时间戳“看着对，算着错”。

六、可靠性与可观测性：把不可控变成可控
重试与超时：

• 所有外部 I/O 必须设置合理超时，httpx 默认超短，按业务调整。
• backoff 实现指数退避，最多尝试 N 次，避免风暴。

幂等与事务：

• 定时任务可能重复执行，设计幂等键。例如按日期和来源构造唯一键，数据库 UPSERT。
• 跨多表写入时用事务；长事务要谨慎，必要时拆分批量。

分布式锁：

• 简单需求用 Redis NX 锁足矣，严格一致性可引入 Redlock 或数据库 advisory lock。
• 锁失效后补偿机制要考虑，避免部分步骤执行两次。

可观测三件套：

• 日志 JSON 化，字段包含 trace_id、user_id、job_name。
• metrics 至少包括处理耗时、失败率、外部依赖延迟、队列堆积。
• trace 把关键 I/O 链路串起来，定位慢点。可用 OpenTelemetry + Jaeger。

告警策略：

• 阈值类告警：定时任务超时、失败率超过 5% 连续三次。
• 异常模式告警：指标出值偏离过去 7 天均值 3σ，怀疑上游变更。
• 告警必须降噪，夜间不推无效信息，重要程度分级。

七、性能与成本：花在哪儿，省在哪儿
Python 性能调优要结合业务形态：

• I/O 密集：尽量异步化，httpx + asyncio；数据库连接池调优，避免 N+1。
• CPU 密集：向量化（pandas/numpy）或用 Cython/numba 做热点；必要时拆出 worker 进程。
• 进程模型：uvicorn + workers，结合机器核数；适当打开 uvloop。
• 缓存：热点指标 5~30 秒短缓存，API 延迟明显下降。
• 数据库：合理索引、只取必要字段；高 QPS 时考虑 PgBouncer。

成本控制：

• 镜像约 200MB，可接受；若要精简，去除不必要系统包。
• 横向扩展比纵向堆配置更经济；结合 HPA 或自动扩容策略。
• 打点采样率不要太高，避免观测成本反噬。

八、CI/CD 与发布：让“上线”成为日常小事
GitHub Actions 简化流程：

• 对 main 分支 push：
  1. 运行测试、lint、type check。
  2. 构建镜像，打 semver 标签和 git sha。
  3. 使用 Trivy 扫描漏洞，阻断高危。
  4. 推送到私有 Registry。
  5. 触发部署（例如 ArgoCD 监控 helm chart 更新）。

零停机与回滚：

• 蓝绿或金丝雀：5/20/100 流量递增观察错误率。
• 数据库迁移前置，兼容新老版本一段时间；回滚时代码先回，再执行 down migration。
• 配置以环境变量为主，不在镜像里固化密钥；生产用 Docker secrets 或 KMS。

九、那些把我坑过的细节

• PID 1 信号处理。直接 python app/main.py 会让应用成为 PID 1，SIGTERM 不转发，优雅停机失败。解决：用 tini 或让 uvicorn 成为 PID 1，确认 lifespan 正确关闭连接。
• 时区和夏令时。容器默认 UTC，本地是 CST，计算“昨天”的窗口错一整天。解决：统一用 UTC 存储，展示时转换；或容器内设 TZ 并严格用 aware datetime。
• Docker 网络 DNS 缓存。上游 API 改了 IP，但容器长连接不刷新。解决：设置合理 DNS TTL 和客户端连接池寿命。
• Alpine musl 与加密库。曾因 cryptography 在 musl 上性能下降，握手延迟飙升。换回 debian-slim 后恢复。
• 卷权限。非 root 用户写日志目录权限不足导致宕掉。解决：预创建目录并 chown 给运行用户，或把日志直接打到 stdout。
• 依赖地狱。不同机器 Poetry 解析版本不一致，最后改为导出 requirements.txt + hash pin，同时锁定 Python 次版本。

十、上线后的反馈与数据
我们按上面方案跑了三个月，记录了几组变化：

• 指标出数延迟从平均 18 分钟下降到 2 分钟内波动。
• 每周因“环境不一致”导致的排障工时基本清零。
• 接口 99.85% 的 24 小时可用性，严重告警每月小于 2 次。
• 新业务接入从“拉群对接 + 反复改脚本”的 3 天，缩短为按模板配置 + 增量开发的一天内。

更重要的是团队的心态变化：大家不再害怕“上线”，因为上线变成了可预期的重复动作；也不再把“脚本”当成一次性的工具，而是当作一个需要被维护的产品。

十一、把它产品化的那份“方法论”
这部分不谈技术细节，只谈做法：

• 定义边界。这个产品解决什么，不解决什么。边界是成本控制的第一道防线。
• 优先可靠性，再谈聪明。先把超时、重试、日志、告警做扎实，之后再优化算法和性能。
• 基础设施即代码。Dockerfile、compose、helm、CI 管道都在仓库里，任何人都能复现。
• 指标驱动。可用性、延迟、失败率、错误修复时长这些都是产品的“北极星指标”。
• 持续消除“跑偏成本”。每次故障复盘，更新检查清单：发布前检查项、数据库迁移流程、回滚预案。
• 小步快跑。两周一个稳定节奏，逐步引入复杂能力，比如异地灾备、更多数据源。

十二、一个可复制的最小模板清单

• 代码层面
  1. FastAPI + pydantic-settings + logging JSON
  2. 业务分为 collector/compute/store/job/api
  3. 单元测试覆盖关键路径，HTTP 和 DB 用 fixture/stub
• 容器层面
  1. 多阶段构建，非 root 用户，健康检查
  2. ENV 明确，镜像不可变，配置注入
  3. docker-compose 本地，生产用编排器（K8s/Swarm/自托管）
• 运维与观测
  1. /metrics, /health, /ready 三个端点
  2. 统一 trace_id，日志落 ELK/ClickHouse
  3. 告警策略：阈值 + 异常检测
• 数据可靠性
  1. 幂等键 + UPSERT
  2. 迁移双向脚本 + 兼容窗口
  3. 备份与恢复演练
• 安全
  1. 依赖扫描与基础镜像定期更新
  2. 最小权限网络与密钥管理
  3. 第三方调用签名与限流

结语
很多人问，为什么用 Python 而不是 Go、为什么用 Docker 而不是直接虚机。我的答案是：先看问题，再选工具。在这个要“快速迭代、稳定上线、便于协作”的场景里，Python 的生产力和丰富生态、Docker 的环境一致性和可复制性几乎是天然契合。只要把产品边界和可靠性建设放在前面，剩下的是扎扎实实的细节活。

写到这里，我想起最初那段散落在同事电脑里的“聪明脚本”。它们并不差，只是缺少把价值“反复交付”的外壳。Python 给了我们解决问题的表达力，Docker 给了我们标准化的交付方式；两者结合，不仅是技术栈，更是一种产品化的方法。愿你的下一个“小工具”，也能长成一个“能活”的产品。