如何写代码适配容器运行环境

wen 实用脚本 2026-06-26 250

本文目录导读：

如何写代码适配容器运行环境

环境变量配置
服务发现与连接
优雅关闭与信号处理
Dockerfile 优化
日志与监控
环境适配最佳实践

我来为你详细介绍如何编写适配容器运行环境的代码,包括最佳实践和代码示例。

环境变量配置

配置文件示例

# 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

通过以上代码示例和最佳实践，你的应用将能够很好地适应容器运行环境,关键点包括：