关键词：Prometheus、Python、指标暴露、Counter、Gauge、Histogram、WSGI 阅读收获：10 分钟学会 3 种官方推荐方案，告别“print 调试 + 人肉监控”
0. 选型一览：3 种姿势怎么选？
| ① start_http_server 独立端口 | 脚本/快速验证 | 零依赖，1 行代码 | 多进程要手动合流 |
| ② WSGI 嵌入（Flask/Django） | Web 服务 | 端口复用，可鉴权 | 需改路由 |
| ③ Pushgateway 推送 | 短任务/Cron | 拉模式失效时救场 | 需运维 Pushgateway |
下文给出可运行最小示例，按场景直接抄。
1. 姿势一：独立线程暴露 /metrics（最简）
安装
pip install prometheus-client
代码
from prometheus_client import Counter, Histogram, start_http_server
import time, random
# ① 定义指标
req_count = Counter('http_requests_total', 'Total HTTP requests', ['method'])
req_duration = Histogram('http_request_duration_seconds', 'HTTP latency')
# ② 模拟业务
def handle():
req_count.labels(method='GET').inc()
with req_duration.time(): # 自动记录耗时
time.sleep(random.uniform(0.01, 0.5))
# ③ 启动独立线程，监听 8000
start_http_server(8000)
print('Metrics: http://localhost:8000/metrics')
while True:
handle()
time.sleep(1)
访问 http://localhost:8000/metrics 即可看到 Prometheus 标准格式文本。 适合：脚本、定时任务、本地调试 。
2. 姿势二：WSGI 嵌入现有 Web 应用（生产推荐）
无需额外端口，直接复用业务端口，天然享受认证/HTTPS。
Flask 版
from flask import Flask, request
from prometheus_client import Counter, generate_latest, CONTENT_TYPE_LATEST
app = Flask(__name__)
counter = Counter('api_invocations_total', 'Total API calls', ['endpoint'])
@app.route('/')
def index():
counter.labels(endpoint='/').inc()
return 'ok'
@app.route('/metrics')
def metrics():
return generate_latest(), 200, {'Content-Type': CONTENT_TYPE_LATEST}
if __name__ == '__main__':
app.run(port=8080)
Django 版
路由 urls.py：
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from django.http import HttpResponse
def metrics(_):
return HttpResponse(generate_latest(), content_type=CONTENT_TYPE_LATEST)
urlpatterns = [path('metrics', metrics), ...]
适合：已有 Web 框架、需要统一端口与权限管控 。
3. 姿势三：Pushgateway 推送（短任务/Cron）
Prometheus 拉模式对生命周期 < 30s 的脚本不友好，Pushgateway 救场。
步骤
from prometheus_client import CollectorRegistry, Gauge, push_to_gateway
import time
registry = CollectorRegistry()
duration = Gauge('backup_duration_seconds', 'Backup runtime', registry=registry)
start = time.time()
# do backup …
duration.set(time.time() – start)
push_to_gateway('pushgateway:9091', job='mysql_backup', registry=registry)
清理策略
Pushgateway 不自动删除，需设置 TTL 或脚本结束手动 delete_from_gateway()。 适合：Cron 备份、CI/CD、批量计算 。
4. 指标类型速查（官方 4 种）
| Counter | 只增不减（请求数、订单数） | inc() |
| Gauge | 可增可减（温度、内存） | set() / inc() / dec() |
| Histogram | 延迟/大小分布 | observe(value) |
| Summary | 类似 Histogram，带滑动窗 | observe(value) |
示例：
from prometheus_client import Counter, Gauge, Histogram
c = Counter('requests_total', 'Total requests', ['method'])
g = Gauge('temperature_celsius', 'Current temp')
h = Histogram('request_duration_seconds', 'Latency', buckets=(0.1, 0.5, 1, 5))
c.labels(method='GET').inc()
g.set(23.4)
h.observe(0.267)
5. 多进程模式（Gunicorn/uWSGI）必开
单进程指标只统计本 Worker，需合流。
from prometheus_client import multiprocess, CollectorRegistry
import os
# ① 启动前设置环境变量
os.environ['prometheus_multiproc_dir'] = '/tmp/prometheus'
os.makedirs('/tmp/prometheus', exist_ok=True)
# ② 在 worker 里注册
def child_exit(server, worker):
multiprocess.mark_process_dead(worker.pid)
WSGI 入口：
registry = CollectorRegistry()
multiprocess.MultiProcessCollector(registry)
def metrics(request):
return HttpResponse(generate_latest(registry), content_type=CONTENT_TYPE_LATEST)
官方文档称性能损耗 < 2%，可放心开启 。
6. Prometheus 抓取配置示例
prometheus.yml：
scrape_configs:
– job_name: 'python-app'
static_configs:
– targets: ['app:8080']
metrics_path: '/metrics'
scrape_interval: 15s
重启 Prometheus，Status → Targets 看到 UP 即成功。
7. Grafana 面板 30 秒导入
8. 常见坑 & 排查清单
| metrics 404 | 路由未注册 | 确认 /metrics 返回 200 |
| 指标值一直 0 | 标签未匹配 | labels(**kwargs) 必须与定义一致 |
| 多进程缺数据 | 未开 MultiProcessCollector | 加环境变量与合流代码 |
| 标签维度爆炸 | 把用户 ID 当标签 | 用 Summary 统计，勿用高基维标签 |
9. 总结：一张图带走 3 种姿势
独立端口 8000 ← ① 脚本、调试
Web 嵌入 /metrics ← ② 生产、复用端口
Pushgateway 9091 ← ③ 短任务、Cron
下一步：给指标加 SLO 告警（rate(requests_total[5m]) > 1000），让监控真正闭环。
10. 附录：全部代码 GitHub 地址
https://github.com/yourname/prometheus-python-demos 含 Flask/Django/Cron 3 套模板，直接 docker-compose up 体验。
