关于AI蓝绿部署的几点记录

蓝绿部署 零停机 部署策略
作者: 醉月思 发布日期: 2023-08-19 分类: 技术实践

让我先说一下我们之前是怎么部署的：

  1. 停止运行中的服务进程
  2. 部署新代码和模型文件
  3. 启动新服务进程
  4. 等待服务就绪（有时候要等很久，因为模型需要加载）
  5. 恢复流量

这个过程看似简单，但实际上充满了坑：

为什么写这篇文章

说实话，每次发布 AI 应用的时候，我都在担心一个问题：服务会不会挂？

去年我们公司上线了一个基于大模型的客服系统，每次部署新版本的时候，用户总是会遇到"服务暂时不可用"的提示。虽然我们在凌晨3点发布，但那些凌晨还来咨询问题的用户，依然会失望地离开。

更糟糕的是有一次，新版本上线后，模型的响应时间从2秒飙升到了8秒，用户体验直接崩了。我们紧急回滚，但回滚过程又花了20分钟，这段时间整个服务都不可用。

这让我下定决心：一定要找到一个真正能实现零停机部署的方案。

经过半年的实践和多次踩坑，我终于在 AI 服务部署中完整实现了蓝绿部署。这篇文章就是要把这个过程中的血泪经验分享出来，让大家少走弯路。

背景：AI 服务的部署困境

传统的部署方式有多痛

让我先说一下我们之前是怎么部署的：

  1. 停止运行中的服务进程
  2. 部署新代码和模型文件
  3. 启动新服务进程
  4. 等待服务就绪（有时候要等很久，因为模型需要加载）
  5. 恢复流量

这个过程看似简单，但实际上充满了坑：

  • 停机时间不可控：模型加载时间不固定，小的模型可能10秒，大的模型可能需要几分钟
  • 回滚困难：一旦新版本有问题，回滚需要重新走一遍部署流程
  • 依赖关系复杂：AI 服务的依赖很多（Python 环境、CUDA、模型文件），任何一个环节出问题都会导致部署失败

为什么蓝绿部署是答案

蓝绿部署的核心思想很简单：同时维护两套完全相同的环境，一个叫 Blue（当前生产环境），一个叫 Green（新版本环境）。

部署的时候，我们这样做：

  1. 在 Green 环境部署新版本
  2. 等待 Green 环境完全就绪
  3. 切换流量到 Green 环境
  4. 如果有问题，快速切换回 Blue 环境

这个方案的优点是：

  • 零停机：流量切换是瞬间完成的
  • 快速回滚：只需要切换流量，不需要重新部署
  • 风险可控：新版本在接收真实流量之前已经完全就绪

为了更直观地理解传统部署和蓝绿部署的区别，我用一个时序图来对比：

gantt title 传统部署 vs 蓝绿部署时序对比 dateFormat HH:mm:ss axisFormat %H:%M:%S section 传统部署 停止服务 :a1, 10:00:00, 5s 部署新版本 :a2, after a1, 30s 启动服务 :a3, after a2, 10s 等待模型加载 :a4, after a3, 2m 恢复流量 :a5, after a4, 5s 服务不可用期 :crit, 10:00:00, 2m50s section 蓝绿部署 部署到Green环境 :b1, 10:00:00, 30s 启动Green服务 :b2, after b1, 10s 等待模型加载 :b3, after b2, 2m 切换流量 :b4, 10:02:40, 5s Blue服务持续运行 :b5, 10:00:00, 2m45s

实现方案：从理论到落地

系统架构设计

我设计的架构是这样的：

graph TD A[用户请求] --> B[Nginx 负载均衡] B --> C{流量路由} C -->|当前流量| D[Blue 环境] C -->|切换后流量| E[Green 环境] D --> F[AI 服务 Blue] E --> G[AI 服务 Green] F --> H[共享存储模型文件] G --> H F --> I[数据库 Redis] G --> I

具体实现步骤

1. 环境准备

首先需要准备两套完全相同的环境：

# 目录结构
/opt/ai-service/
├── blue/           # 蓝色环境
│   ├── venv/       # Python 虚拟环境
│   ├── models/     # 模型文件（软链接到共享存储）
│   └── app.py      # 应用代码
├── green/          # 绿色环境
│   ├── venv/
│   ├── models/
│   └── app.py
└── shared/         # 共享存储
    └── models/     # 模型文件

关键点：

  • Python 虚拟环境需要独立，避免依赖冲突
  • 模型文件使用软链接到共享存储，节省磁盘空间
  • 两个环境的配置文件完全相同，除了端口号

2. 服务管理脚本

我写了一个 Python 脚本来管理蓝绿部署：

#!/usr/bin/env python3
import os
import sys
import time
import requests
import subprocess
from pathlib import Path

class BlueGreenDeployer:
    def __init__(self):
        self.base_path = Path('/opt/ai-service')
        self.blue_path = self.base_path / 'blue'
        self.green_path = self.base_path / 'green'
        self.shared_models = self.base_path / 'shared' / 'models'
        self.blue_port = 8001
        self.green_port = 8002
        self.health_check_url = 'http://localhost:{port}/health'

    def get_current_active_env(self):
        """检查当前哪个环境是活跃的"""
        try:
            response = requests.get('http://localhost/api/status', timeout=2)
            data = response.json()
            return data.get('active_env', 'blue')
        except:
            return 'blue'  # 默认返回 blue

    def deploy_to_inactive_env(self, new_version):
        """部署到非活跃环境"""
        current_active = self.get_current_active_env()
        target_env = 'green' if current_active == 'blue' else 'blue'
        target_path = self.green_path if target_env == 'green' else self.blue_path

        print(f"准备部署到 {target_env} 环境...")

        # 1. 停止目标环境的服务
        self.stop_service(target_path)

        # 2. 更新代码
        print(f"更新 {target_env} 环境代码...")
        subprocess.run([
            'git', '-C', str(target_path), 'fetch', 'origin'
        ], check=True)
        subprocess.run([
            'git', '-C', str(target_path), 'checkout', new_version
        ], check=True)

        # 3. 更新依赖
        print(f"更新 {target_env} 环境依赖...")
        subprocess.run([
            str(target_path / 'venv' / 'bin' / 'pip'), 'install', '-r',
            str(target_path / 'requirements.txt')
        ], check=True)

        # 4. 确保模型文件链接正确
        models_link = target_path / 'models'
        if models_link.exists():
            models_link.unlink()
        models_link.symlink_to(self.shared_models)

        # 5. 启动服务
        print(f"启动 {target_env} 环境服务...")
        self.start_service(target_path, target_env)

        # 6. 健康检查
        port = self.green_port if target_env == 'green' else self.blue_port
        if not self.wait_for_service_ready(port):
            print(f"❌ {target_env} 环境服务启动失败")
            return False

        print(f"✅ {target_env} 环境部署成功")
        return True

    def switch_traffic(self, target_env):
        """切换流量到指定环境"""
        port = self.green_port if target_env == 'green' else self.blue_port

        print(f"切换流量到 {target_env} 环境...")

        # 更新 Nginx 配置
        nginx_config = f"""
upstream ai_service {{
    server localhost:{port};
}}

server {{
    listen 80;
    server_name ai.example.com;

    location / {{
        proxy_pass http://ai_service;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }}
}}
"""

        # 写入临时配置文件
        temp_config = Path('/tmp/ai_service_nginx.conf')
        temp_config.write_text(nginx_config)

        # 测试配置
        result = subprocess.run(
            ['nginx', '-t', '-c', str(temp_config)],
            capture_output=True
        )

        if result.returncode != 0:
            print(f"❌ Nginx 配置测试失败: {result.stderr.decode()}")
            return False

        # 复制配置到 Nginx 配置目录
        subprocess.run([
            'sudo', 'cp', str(temp_config),
            '/etc/nginx/sites-available/ai_service'
        ], check=True)

        # 重新加载 Nginx
        subprocess.run(['sudo', 'nginx', '-s', 'reload'], check=True)

        print(f"✅ 流量已切换到 {target_env} 环境")
        return True

    def wait_for_service_ready(self, port, timeout=300):
        """等待服务就绪"""
        start_time = time.time()
        url = self.health_check_url.format(port=port)

        while time.time() - start_time < timeout:
            try:
                response = requests.get(url, timeout=5)
                if response.status_code == 200:
                    data = response.json()
                    if data.get('status') == 'ready':
                        print(f"✅ 服务在端口 {port} 已就绪")
                        return True
            except:
                pass

            print(f"等待服务就绪... ({int(time.time() - start_time)}s)")
            time.sleep(5)

        return False

    def stop_service(self, env_path):
        """停止指定环境的服务"""
        pid_file = env_path / 'app.pid'
        if pid_file.exists():
            with open(pid_file) as f:
                pid = int(f.read().strip())
            try:
                os.kill(pid, 15)  # SIGTERM
                time.sleep(5)
                # 如果还在运行，强制杀掉
                try:
                    os.kill(pid, 0)  # 检查进程是否存在
                    os.kill(pid, 9)  # SIGKILL
                except:
                    pass
            except:
                pass
            pid_file.unlink()

    def start_service(self, env_path, env_name):
        """启动指定环境的服务"""
        port = self.green_port if env_name == 'green' else self.blue_port
        venv_python = env_path / 'venv' / 'bin' / 'python'
        app_path = env_path / 'app.py'
        pid_file = env_path / 'app.pid'
        log_file = env_path / f'{env_name}.log'

        # 启动服务进程
        process = subprocess.Popen(
            [str(venv_python), str(app_path), '--port', str(port)],
            stdout=open(log_file, 'w'),
            stderr=subprocess.STDOUT,
            start_new_session=True
        )

        # 写入 PID 文件
        with open(pid_file, 'w') as f:
            f.write(str(process.pid))

        print(f"✅ {env_name} 环境服务已启动 (PID: {process.pid})")

if __name__ == '__main__':
    deployer = BlueGreenDeployer()

    if len(sys.argv) < 2:
        print("Usage: python deploy.py <version> [switch]")
        sys.exit(1)

    version = sys.argv[1]

    # 部署到非活跃环境
    if deployer.deploy_to_inactive_env(version):
        # 如果指定了 switch 参数，则切换流量
        if len(sys.argv) > 2 and sys.argv[2] == 'switch':
            current_active = deployer.get_current_active_env()
            target_env = 'green' if current_active == 'blue' else 'blue'
            deployer.switch_traffic(target_env)
    else:
        print("❌ 部署失败")
        sys.exit(1)

3. Nginx 配置

Nginx 作为反向代理，负责流量路由：

# /etc/nginx/sites-available/ai_service
upstream ai_service {
    server localhost:8001;  # 默认指向 blue 环境
}

server {
    listen 80;
    server_name ai.example.com;

    location / {
        proxy_pass http://ai_service;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

        # AI 服务的特殊设置
        proxy_read_timeout 300;  # AI 推理可能需要较长时间
        proxy_connect_timeout 60;
        proxy_send_timeout 60;
    }

    # 健康检查端点
    location /health {
        proxy_pass http://ai_service/health;
        access_log off;
    }
}

4. 健康检查端点

在 AI 服务中实现健康检查：

from fastapi import FastAPI
import torch
from pathlib import Path

app = FastAPI()

# 全局变量，用于标记服务状态
model_loaded = False
model = None

@app.on_event("startup")
async def startup_event():
    global model, model_loaded
    # 加载模型
    model_path = Path("models/base_model.pt")
    print(f"Loading model from {model_path}...")
    model = torch.load(model_path)
    model_loaded = True
    print("Model loaded successfully")

@app.get("/health")
async def health_check():
    """健康检查端点"""
    if model_loaded:
        return {
            "status": "ready",
            "model_loaded": True,
            "active_env": os.getenv("ENV_NAME", "unknown")
        }
    else:
        return {
            "status": "loading",
            "model_loaded": False,
            "active_env": os.getenv("ENV_NAME", "unknown")
        }

@app.post("/predict")
async def predict(text: str):
    """预测接口"""
    if not model_loaded:
        return {"error": "Service not ready"}
    # ... 实际的预测逻辑
    return {"result": "prediction"}

部署流程

完整的部署流程是这样的：

sequenceDiagram participant Dev as 开发者 participant Script as 部署脚本 participant Blue as Blue环境 participant Green as Green环境 participant Nginx as Nginx participant User as 用户 Dev->>Script: 执行部署命令 Script->>Script: 检查当前活跃环境 Script->>Blue: 停止Blue服务（如果Blue是非活跃的） Script->>Blue: 拉取新代码 Script->>Blue: 更新依赖 Script->>Blue: 启动Blue服务 Script->>Blue: 健康检查 Blue-->>Script: 服务就绪 Script->>Dev: 部署成功 Note over Dev,Script: 用户决定切换流量 Dev->>Script: 执行切换命令 Script->>Nginx: 更新配置 Script->>Nginx: 重新加载 Nginx->>Green: 路由流量到Green User->>Green: 发送请求 Green-->>User: 返回结果

踩坑记录：那些让我秃头的问题

问题 1：模型加载时间超长

症状：健康检查一直超时，部署脚本以为服务启动失败。

原因：模型文件太大（8GB），加载时间超过了5分钟的等待时间。

解决方案：

  1. 实现渐进式健康检查
  2. 先检查进程是否启动，再检查模型是否加载完成
  3. 增加超时时间到30分钟

为了更好地理解模型加载的过程，我用一个状态机图来展示服务启动的各个阶段：

stateDiagram-v2 [*] --> 进程启动: 启动命令 进程启动 --> HTTP服务就绪: FastAPI启动完成 HTTP服务就绪 --> 模型加载中: 开始加载模型 模型加载中 --> 模型加载中: 加载权重文件 模型加载中 --> 模型加载中: 初始化推理引擎 模型加载中 --> 服务就绪: 模型加载完成 服务就绪 --> [*]: 健康检查通过 note right of 模型加载中 可能需要2-5分钟 取决于模型大小 end note

改进后的健康检查代码：

def wait_for_service_ready(self, port, timeout=1800):
    """等待服务就绪 - 改进版"""
    start_time = time.time()
    url = self.health_check_url.format(port=port)

    # 先检查进程是否启动
    while time.time() - start_time < 60:
        try:
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                break
        except:
            pass
        time.sleep(2)

    # 再检查模型是否加载完成
    while time.time() - start_time < timeout:
        try:
            response = requests.get(url, timeout=5)
            if response.status_code == 200:
                data = response.json()
                if data.get('status') == 'ready':
                    return True
                elif data.get('status') == 'loading':
                    print(f"模型加载中... ({int(time.time() - start_time)}s)")
                else:
                    print(f"未知状态: {data.get('status')}")
        except:
            print(f"健康检查失败，重试中... ({int(time.time() - start_time)}s)")
        time.sleep(10)

    return False

问题 2：端口冲突

症状：启动新环境服务时，提示端口已被占用。

原因：旧服务没有正确停止，端口还被占用。

解决方案：

  1. 在启动前检查端口是否被占用
  2. 如果被占用，强制杀掉占用端口的进程
  3. 增加端口清理逻辑
def check_and_free_port(self, port):
    """检查并释放端口"""
    try:
        # 查找占用端口的进程
        result = subprocess.run(
            ['lsof', '-t', '-i', f':{port}'],
            capture_output=True,
            text=True
        )
        if result.stdout.strip():
            pids = result.stdout.strip().split('\n')
            for pid in pids:
                try:
                    os.kill(int(pid), 9)  # 强制杀掉进程
                    print(f"已杀掉占用端口 {port} 的进程 {pid}")
                except:
                    pass
            time.sleep(2)  # 等待端口释放
    except:
        pass  # lsof 命令可能不存在

问题 3：依赖包版本冲突

症状：新环境启动后，报错 “ModuleNotFoundError” 或依赖版本不匹配。

原因：两个环境的虚拟环境没有完全隔离，或者依赖版本不一致。

解决方案：

  1. 使用 pip freeze > requirements.txt 锁定版本
  2. 在部署时清空虚拟环境后重新安装
  3. 使用 Docker 容器化（终极解决方案）
def rebuild_venv(self, env_path):
    """重建虚拟环境"""
    venv_path = env_path / 'venv'
    requirements = env_path / 'requirements.txt'

    # 删除旧虚拟环境
    if venv_path.exists():
        shutil.rmtree(venv_path)

    # 创建新虚拟环境
    subprocess.run([
        sys.executable, '-m', 'venv', str(venv_path)
    ], check=True)

    # 升级 pip
    pip_path = venv_path / 'bin' / 'pip'
    subprocess.run([str(pip_path), 'install', '--upgrade', 'pip'], check=True)

    # 安装依赖
    subprocess.run([
        str(pip_path), 'install', '-r', str(requirements)
    ], check=True)

问题 4：流量切换后发现严重 bug

症状：切换流量后，发现新版本有严重问题，但用户已经受到了影响。

原因：没有对新版本进行充分测试就切换了流量。

解决方案：

  1. 实现金丝雀发布，先切换 10% 的流量
  2. 添加监控告警，及时发现异常
  3. 准备快速回滚脚本
def canary_deploy(self, target_env, percentage=10):
    """金丝雀发布"""
    port = self.green_port if target_env == 'green' else self.blue_port

    # 更新 Nginx 配置，只路由部分流量
    nginx_config = f"""
upstream ai_service {{
    server localhost:{port} weight={percentage};
    server localhost:{8001 if port == 8002 else 8002} weight={100 - percentage};
}}
"""
    # ... 更新配置的代码

实际效果：从停机20分钟到零停机

改进前的指标

  • 平均停机时间：15-20分钟
  • 部署成功率：85%（经常因为各种原因失败）
  • 回滚时间：15-25分钟
  • 用户投诉：每次部署后都有投诉

改进后的指标

  • 平均停机时间：0秒（真正零停机）
  • 部署成功率：98%（失败的主要原因是网络问题）
  • 回滚时间：5秒（只是切换 Nginx 配置）
  • 用户投诉：0次（用户完全没有感知）

性能对比图

我用 matplotlib 绘制了改进前后的性能对比图：

import matplotlib.pyplot as plt
import numpy as np

# 设置中文字体
plt.rcParams['font.sans-serif'] = ['DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

# 数据
metrics = ['停机时间(分钟)', '部署成功率(%)', '回滚时间(分钟)']
before = [17.5, 85, 20]
after = [0, 98, 0.083]  # 5秒 = 0.083分钟

x = np.arange(len(metrics))
width = 0.35

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))

# 左图：时间指标
time_metrics = ['停机时间', '回滚时间']
time_before = [17.5, 20]
time_after = [0, 0.083]

x1 = np.arange(len(time_metrics))
ax1.bar(x1 - width/2, time_before, width, label='改进前', color='#ff6b6b')
ax1.bar(x1 + width/2, time_after, width, label='改进后', color='#51cf66')
ax1.set_ylabel('时间 (分钟)')
ax1.set_title('时间指标对比')
ax1.set_xticks(x1)
ax1.set_xticklabels(time_metrics)
ax1.legend()
ax1.grid(axis='y', alpha=0.3)

# 右图：成功率
success_before = 85
success_after = 98
x2 = [0, 1]
ax2.bar(x2, [success_before, success_after], color=['#ff6b6b', '#51cf66'])
ax2.set_ylabel('成功率 (%)')
ax2.set_title('部署成功率对比')
ax2.set_xticks(x2)
ax2.set_xticklabels(['改进前', '改进后'])
ax2.set_ylim(0, 100)
ax2.grid(axis='y', alpha=0.3)

plt.tight_layout()
plt.savefig('/tmp/deployment_comparison.png', dpi=150, bbox_inches='tight')

我用一个更直观的雷达图来展示综合性能的提升：

pie title 改进前后性能提升 "停机时间减少" : 100 "回滚时间减少" : 99 "部署成功率提升" : 15 "用户投诉减少" : 100

这张图清晰地展示了改进前后的对比，时间指标从几十分钟降到了几乎为0，成功率从85%提升到了98%。

经验总结

做对了什么

  1. 健康检查很重要：不能只检查进程是否启动，还要检查服务是否真正就绪
  2. 超时时间要合理：AI 服务的启动时间可能很长，要根据实际情况设置
  3. 监控是必须的：部署后要密切关注服务状态，及时发现问题
  4. 文档要详细：把部署流程、常见问题都记录下来，方便团队使用

如果重来一次

  1. 使用 Docker：容器化可以解决很多环境问题，部署更可靠
  2. 使用 Kubernetes：K8s 原生支持滚动更新和蓝绿部署，不需要自己造轮子
  3. 自动化测试：部署前自动运行测试，减少上线后的问题
  4. 更好的监控：集成 Prometheus、Grafana 等监控工具，实时监控服务状态

写在最后

实现蓝绿部署的过程并不轻松，踩了不少坑，熬了不少夜。但当看到第一次零停机部署成功，用户完全没有感知时，那种成就感是值得的。

对于 AI 服务来说，蓝绿部署不仅仅是一个部署策略，更是一种对用户体验的承诺。毕竟，用户不会关心你用了什么技术，他们只关心服务是否稳定可用。

希望这篇文章能帮助到正在为部署问题烦恼的同学。如果有什么问题，欢迎交流讨论。

相关资源：

版权声明: 本文首发于 指尖魔法屋-关于AI蓝绿部署的几点记录（https://blog.thinkmoon.cn/post/293-ai-blue-green-deployment-downtime-zero-downtime/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章

Harness Engineering 的介绍与编程实践
这篇文章从 AI 编程的真实工程场景出发，梳理 Harness Engineering 如何通过上下文、工具、沙盒、验证和人工闸门，让 Agent 从“会写代码”走向“能稳定交付”。
📁 技术实践
📅 2026-04-23
AI长度惩罚折腾手记
文档生成项目里有个反差：用户问「Python 快速排序怎么实现」，模型回一行「平均复杂度 O(n log n)」就收工；换到 API 文档场景，…
📁 技术实践
📅 2025-12-31