本文目录导读：
我来为你详细介绍如何编写适配容器运行环境的代码,包括最佳实践和代码示例。
环境变量配置
配置文件示例
# config.yaml - 容器环境配置
app:
name: my-app
version: 1.0.0
database:
host: ${DB_HOST:-localhost}
port: ${DB_PORT:-5432}
username: ${DB_USER:-admin}
password: ${DB_PASSWORD}
database: ${DB_NAME:-myapp}
redis:
host: ${REDIS_HOST:-localhost}
port: ${REDIS_PORT:-6379}
password: ${REDIS_PASSWORD:-}
logging:
level: ${LOG_LEVEL:-info}
format: ${LOG_FORMAT:-json}
Python 示例
import os
import logging
from dotenv import load_dotenv
class Config:
"""容器环境配置管理"""
def __init__(self):
# 尝试加载 .env 文件（本地开发）
load_dotenv()
# 应用配置
self.app_name = os.getenv('APP_NAME', 'my-app')
self.debug = os.getenv('DEBUG', 'false').lower() == 'true'
# 数据库配置
self.db_config = {
'host': os.getenv('DB_HOST', 'localhost'),
'port': int(os.getenv('DB_PORT', '5432')),
'username': os.getenv('DB_USER', 'admin'),
'password': os.getenv('DB_PASSWORD', ''),
'database': os.getenv('DB_NAME', 'myapp')
}
# Redis 配置
self.redis_config = {
'host': os.getenv('REDIS_HOST', 'localhost'),
'port': int(os.getenv('REDIS_PORT', '6379')),
'password': os.getenv('REDIS_PASSWORD', '')
}
# 服务发现配置
self.service_discovery = os.getenv('SERVICE_DISCOVERY', 'dns').lower()
def get_database_url(self):
"""获取数据库连接URL"""
return (f"postgresql://{self.db_config['username']}:"
f"{self.db_config['password']}@"
f"{self.db_config['host']}:{self.db_config['port']}/"
f"{self.db_config['database']}")
# 初始化配置
config = Config()
logger = logging.getLogger(__name__)
服务发现与连接
Docker Compose 服务发现示例
# docker-compose.yml
version: '3.8'
services:
app:
build: .
ports:
- "8080:8080"
environment:
- DB_HOST=postgres
- REDIS_HOST=redis
- SERVICE_DISCOVERY=dns
depends_on:
postgres:
condition: service_healthy
redis:
condition: service_healthy
postgres:
image: postgres:15
environment:
POSTGRES_USER: admin
POSTGRES_PASSWORD: ${DB_PASSWORD}
healthcheck:
test: ["CMD-SHELL", "pg_isready"]
interval: 10s
timeout: 5s
retries: 5
redis:
image: redis:7-alpine
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 10s
timeout: 5s
retries: 5
健康检查与重试机制
import time
import random
from functools import wraps
import logging
logger = logging.getLogger(__name__)
def retry_connection(max_retries=5, base_delay=1, backoff=2):
"""连接重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt < max_retries - 1:
delay = base_delay * (backoff ** attempt)
jitter = random.uniform(0, delay * 0.1)
total_delay = delay + jitter
logger.warning(
f"Connection attempt {attempt + 1}/{max_retries} failed: {e}. "
f"Retrying in {total_delay:.2f}s"
)
time.sleep(total_delay)
logger.error(f"All {max_retries} connection attempts failed")
raise last_exception
return wrapper
return decorator
class DatabaseConnector:
"""数据库连接器（支持容器环境）"""
def __init__(self, config):
self.config = config
self.connection = None
@retry_connection(max_retries=5)
def connect(self):
"""连接到数据库"""
import psycopg2
self.connection = psycopg2.connect(
host=self.config.db_config['host'],
port=self.config.db_config['port'],
user=self.config.db_config['username'],
password=self.config.db_config['password'],
database=self.config.db_config['database'],
connect_timeout=5 # 设置连接超时
)
# 验证连接
with self.connection.cursor() as cursor:
cursor.execute("SELECT 1")
result = cursor.fetchone()
logger.info("Database connection established successfully")
return self.connection
def disconnect(self):
"""断开连接"""
if self.connection and not self.connection.closed:
self.connection.close()
logger.info("Database connection closed")
# 健康检查端点
def health_check():
"""健康检查函数"""
health_status = {
'status': 'healthy',
'checks': {}
}
# 检查数据库连接
try:
if db_connector.connection and not db_connector.connection.closed:
health_status['checks']['database'] = 'connected'
else:
health_status['checks']['database'] = 'disconnected'
health_status['status'] = 'unhealthy'
except Exception as e:
health_status['checks']['database'] = f'error: {str(e)}'
health_status['status'] = 'unhealthy'
return health_status
优雅关闭与信号处理
import signal
import sys
import asyncio
class GracefulShutdown:
"""优雅关闭处理器"""
def __init__(self):
self.shutdown_requested = False
self.shutdown_tasks = []
# 注册信号处理器
signal.signal(signal.SIGTERM, self.handle_signal)
signal.signal(signal.SIGINT, self.handle_signal)
def register_shutdown_task(self, task, name=None):
"""注册关闭任务"""
self.shutdown_tasks.append({
'task': task,
'name': name or f'task_{len(self.shutdown_tasks)}'
})
def handle_signal(self, signum, frame):
"""处理关闭信号"""
signal_name = signal.Signals(signum).name
logger.info(f"Received signal: {signal_name}")
if self.shutdown_requested:
logger.warning("Force shutdown requested")
sys.exit(1)
self.shutdown_requested = True
asyncio.create_task(self.graceful_shutdown())
async def graceful_shutdown(self, timeout=30):
"""优雅关闭"""
logger.info("Starting graceful shutdown...")
# 设置关闭超时
shutdown_timeout = time.time() + timeout
# 执行关闭任务
for task in reversed(self.shutdown_tasks):
remaining_time = shutdown_timeout - time.time()
if remaining_time <= 0:
logger.warning(f"Shutdown timeout reached for task: {task['name']}")
break
try:
await asyncio.wait_for(
task['task'](),
timeout=remaining_time
)
logger.info(f"Task '{task['name']}' completed shutdown")
except asyncio.TimeoutError:
logger.warning(f"Task '{task['name']}' shutdown timeout")
except Exception as e:
logger.error(f"Task '{task['name']}' shutdown error: {e}")
logger.info("Graceful shutdown completed")
sys.exit(0)
# 使用示例
def init_app():
"""初始化应用"""
shutdown_handler = GracefulShutdown()
# 注册关闭任务
shutdown_handler.register_shutdown_task(
db_connector.disconnect,
name='database_disconnect'
)
# 启动应用
logger.info("Application started in container environment")
# 保持运行
try:
asyncio.get_event_loop().run_forever()
except KeyboardInterrupt:
pass
Dockerfile 优化
# Dockerfile 多阶段构建
FROM python:3.11-slim as builder
WORKDIR /app
# 安装构建依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc \
&& rm -rf /var/lib/apt/lists/*
# 复制并安装依赖
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt
# 第二阶段：运行环境
FROM python:3.11-slim
WORKDIR /app
# 创建非root用户
RUN groupadd -r appuser && useradd -r -g appuser appuser
# 复制依赖和代码
COPY --from=builder /root/.local /home/appuser/.local
COPY --chown=appuser:appuser . .
# 设置环境变量
ENV PATH=/home/appuser/.local/bin:$PATH \
PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1
# 暴露端口
EXPOSE 8080
# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD python -c "import requests; requests.get('http://localhost:8080/health')"
# 切换为非root用户
USER appuser
# 启动命令
CMD ["python", "app.py"]
日志与监控
import json
import logging
from datetime import datetime
class ContainerLogger:
"""容器日志处理器"""
def __init__(self, name='app'):
self.logger = logging.getLogger(name)
self.logger.setLevel(logging.INFO)
# JSON 格式日志（适合容器环境）
handler = logging.StreamHandler()
handler.setFormatter(ContainerJsonFormatter())
self.logger.addHandler(handler)
def info(self, message, extra=None):
self._log(logging.INFO, message, extra)
def error(self, message, extra=None):
self._log(logging.ERROR, message, extra)
def _log(self, level, message, extra=None):
log_data = {
'message': message,
'timestamp': datetime.utcnow().isoformat(),
'level': logging.getLevelName(level),
}
if extra:
log_data.update(extra)
self.logger.log(level, json.dumps(log_data))
class ContainerJsonFormatter(logging.Formatter):
"""JSON 格式日志格式化器"""
def format(self, record):
log_data = {
'timestamp': datetime.utcnow().isoformat(),
'level': record.levelname,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno
}
# 添加异常信息
if record.exc_info:
log_data['exception'] = self.formatException(record.exc_info)
return json.dumps(log_data)
# 指标收集
class ContainerMetrics:
"""容器指标收集"""
def __init__(self):
self.metrics = {}
def increment_counter(self, name, value=1):
"""增加计数器"""
if name not in self.metrics:
self.metrics[name] = 0
self.metrics[name] += value
def record_gauge(self, name, value):
"""记录指标值"""
self.metrics[name] = value
def get_metrics(self):
"""获取所有指标"""
return {
'metrics': self.metrics,
'timestamp': datetime.utcnow().isoformat()
}
环境适配最佳实践
配置文件模板
# application.yml
server:
port: ${SERVER_PORT:8080}
graceful_shutdown_timeout: 30
spring:
profiles:
active: ${SPRING_PROFILES_ACTIVE:default}
database:
url: jdbc:postgresql://${DB_HOST}:${DB_PORT}/${DB_NAME}
pool:
min: ${DB_POOL_MIN:2}
max: ${DB_POOL_MAX:10}
logging:
level:
root: ${LOG_LEVEL:INFO}
pattern:
console: ${LOG_PATTERN:json}
management:
endpoints:
web:
exposure:
include: health,metrics,info
health:
readinessstate:
enabled: true
通过以上代码示例和最佳实践，你的应用将能够很好地适应容器运行环境,关键点包括：
- 使用环境变量配置
- 实现服务发现和健康检查
- 处理优雅关闭
- 优化容器镜像
- 使用结构化日志
- 实现指标监控
这些实践将确保你的应用在 Kubernetes、Docker Swarm 等容器编排平台上运行良好。
