本文目录导读：
我来介绍几种实现Python健康检查端点的方法：
使用 Flask 框架
基础实现
from flask import Flask, jsonify
import psutil
import time
app = Flask(__name__)
@app.route('/health')
def health_check():
return jsonify({
'status': 'healthy',
'timestamp': time.time()
})
@app.route('/health/ready')
def readiness_check():
"""检查应用是否准备好接收流量"""
# 这里可以检查数据库连接、缓存等
try:
# 模拟检查
return jsonify({
'status': 'ready',
'services': {
'database': 'connected',
'cache': 'connected'
}
})
except Exception as e:
return jsonify({
'status': 'not ready',
'error': str(e)
}), 503
@app.route('/health/live')
def liveness_check():
"""检查应用是否运行"""
return jsonify({
'status': 'alive',
'uptime': time.time()
})
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
详细健康检查（包含资源监控）
from flask import Flask, jsonify
import psutil
import time
import os
app = Flask(__name__)
class HealthChecker:
@staticmethod
def check_disk():
disk = psutil.disk_usage('/')
return {
'total_gb': disk.total / (1024**3),
'used_gb': disk.used / (1024**3),
'free_gb': disk.free / (1024**3),
'percent_used': disk.percent
}
@staticmethod
def check_memory():
memory = psutil.virtual_memory()
return {
'total_gb': memory.total / (1024**3),
'available_gb': memory.available / (1024**3),
'percent_used': memory.percent
}
@staticmethod
def check_cpu():
return {
'percent': psutil.cpu_percent(interval=1),
'count': psutil.cpu_count()
}
@app.route('/health/detailed')
def detailed_health():
start_time = time.time()
health_status = {
'status': 'healthy',
'timestamp': start_time,
'application': {
'name': 'my-app',
'version': '1.0.0',
'pid': os.getpid()
},
'system': {
'disk': HealthChecker.check_disk(),
'memory': HealthChecker.check_memory(),
'cpu': HealthChecker.check_cpu()
}
}
# 检查是否有任何健康状况不佳
if health_status['system']['disk']['percent_used'] > 90:
health_status['status'] = 'degraded'
health_status['response_time_ms'] = (time.time() - start_time) * 1000
return jsonify(health_status)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000)
使用 FastAPI（异步框架）
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse
import psutil
import time
from typing import Dict
app = FastAPI()
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"timestamp": time.time()
}
@app.get("/health/ready")
async def readiness_check():
# 检查依赖服务
dependencies_healthy = True
if dependencies_healthy:
return {"status": "ready"}
else:
return JSONResponse(
status_code=503,
content={"status": "not_ready"}
)
@app.get("/health/live")
async def liveness_check():
return {"status": "alive"}
# 详细检查
@app.get("/health/detailed")
async def detailed_health():
start_time = time.time()
health_data = {
"status": "healthy" if psutil.cpu_percent() < 90 else "degraded",
"timestamp": start_time,
"system": {
"cpu": psutil.cpu_percent(interval=0.5),
"memory": psutil.virtual_memory()._asdict(),
"disk": psutil.disk_usage('/')._asdict()
}
}
return health_data
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Django 实现
urls.py
from django.urls import path
from . import views
urlpatterns = [
path('health/', views.health_check, name='health_check'),
path('health/ready/', views.readiness_check, name='readiness'),
path('health/live/', views.liveness_check, name='liveness'),
]
views.py
from django.http import JsonResponse
from django.db import connections
from django.core.cache import cache
import time
def health_check(request):
return JsonResponse({
'status': 'healthy',
'timestamp': time.time()
})
def readiness_check(request):
"""检查Django应用是否就绪"""
try:
# 检查数据库
db_conn = connections['default']
db_conn.ensure_connection()
db_status = 'connected'
except Exception:
db_status = 'disconnected'
# 检查缓存
try:
cache.set('_health_check', 'ok', 5)
cache.get('_health_check')
cache_status = 'connected'
except Exception:
cache_status = 'disconnected'
if db_status == 'connected' and cache_status == 'connected':
return JsonResponse({
'status': 'ready',
'database': db_status,
'cache': cache_status
})
else:
return JsonResponse({
'status': 'not_ready',
'database': db_status,
'cache': cache_status
}, status=503)
def liveness_check(request):
return JsonResponse({'status': 'alive'})
轻量级 HTTP 服务器实现
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import time
class HealthHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path == '/health' or self.path == '/health/live':
self._send_response({
'status': 'healthy',
'timestamp': time.time()
})
elif self.path == '/health/ready':
# 这里进行准备检查
self._send_response({
'status': 'ready',
'database': 'connected'
})
else:
self.send_response(404)
self.end_headers()
def _send_response(self, data):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(data).encode())
if __name__ == '__main__':
server = HTTPServer(('0.0.0.0', 8080), HealthHandler)
print('Health check server running on port 8080...')
server.serve_forever()
Kubernetes 配置示例
apiVersion: apps/v1
kind: Deployment
metadata:
name: my-app
spec:
replicas: 3
selector:
matchLabels:
app: my-app
template:
metadata:
labels:
app: my-app
spec:
containers:
- name: my-app
image: my-app:latest
ports:
- containerPort: 5000
readinessProbe:
httpGet:
path: /health/ready
port: 5000
initialDelaySeconds: 5
periodSeconds: 10
livenessProbe:
httpGet:
path: /health/live
port: 5000
initialDelaySeconds: 15
periodSeconds: 20
startupProbe:
httpGet:
path: /health
port: 5000
initialDelaySeconds: 3
periodSeconds: 5
最佳实践建议
-
区分探针类型：
- 存活检查（Liveness）：检测应用是否运行
- 就绪检查（Readiness）：检测应用是否准备好处理请求
- 启动检查（Startup）：检测应用是否启动完成
-
返回适当的HTTP状态码：
- 200：健康
- 503：不健康或未就绪
-
添加必要的依赖检查：
- 数据库连接
- 缓存服务
- 外部API
- 消息队列
-
添加性能指标：
- 响应时间
- CPU/内存使用率
- 磁盘空间
-
安全性考虑：
- 限制访问IP
- 添加认证（如果需要）
- 不暴露敏感信息
选择哪种实现方式取决于你的技术栈和需求，Flask/FastAPI适合微服务，Django适合全栈应用,轻量级HTTP服务器适合简单的监控场景。
