当前位置：首页 > 文章列表 > 文章 > python教程 > Jenkins多线程启动Flask避坑全攻略

Jenkins多线程启动Flask避坑全攻略

2026-02-20 16:37:21 0浏览收藏

本文深入剖析了在 Jenkins CI 环境中以多线程方式启动 Flask 测试服务时频繁遭遇的 `OSError: [Errno 98] Address already in use` 顽疾——其根源并非并发冲突，而是 Flask 进程未优雅终止导致端口残留；文章直击三大典型陷阱（调试模式滥用、守护线程误用、生命周期失控），并提供经生产验证的完整解决方案：禁用 reloader/debugger、通过 `/trigger` 路由主动调用 Werkzeug 内置 shutdown、使用非守护线程+显式 `join()` 等待自然退出，辅以动态端口策略、超时防护与结构化日志，让 Flask 测试服务在 Jenkins 中真正实现“启得稳、触得准、停得净”，彻底终结端口占用引发的 CI 不稳定问题。

如何在 Jenkins 中安全启动多线程 Flask 应用避免端口占用错误

本文详解在 Jenkins CI 环境中通过多线程启动 Flask 测试服务时，因端口未及时释放导致 `OSError: [Errno 98] Address already in use` 的根本原因与可靠解决方案，涵盖端口复用预防、优雅关闭机制及线程同步实践。

在 Jenkins 流水线中以线程方式启动 Flask 应用进行集成测试（如触发 pytest 继续执行）是一种常见模式，但本地运行正常、Jenkins 报错 Address already in use 的问题极为典型。其本质并非并发冲突，而是Flask 进程未被正确终止，导致端口（如 5055 或 5005）持续处于 LISTEN 状态——Jenkins 节点通常复用工作空间和系统环境，上一次测试残留的 Python 进程可能仍未退出，netstat -tulpn | grep :5005 显示的 python3.8 进程正是罪魁祸首。

✅ 正确做法：主动关闭 + 同步等待 + 避免调试模式

原方案存在三处关键风险：

使用 run_simple(..., use_debugger=True) 启用了 Werkzeug 调试器，在非交互式 CI 环境中易引发不可控行为；
flask_thread.daemon = True 使线程为守护线程，主线程退出时其被强制终止，stop_flask_app() 永远不会执行，端口无法释放；
wait_for_trigger() 仅依赖 Event.wait()，未对 Flask 服务生命周期做显式管理。

以下是经过 Jenkins 实践验证的健壮实现：

1. Flask 服务模块（test_server.py）

from flask import Flask, request, jsonify
import logging

app = Flask(__name__)
logging.basicConfig(
    filename='flask_thread.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

@app.route('/trigger', methods=['GET'])
def trigger_test_continue():
    logging.info("Trigger received. Signalling pytest to continue.")
    # 关键：收到触发后立即关闭服务器
    shutdown_func = request.environ.get('werkzeug.server.shutdown')
    if shutdown_func is None:
        raise RuntimeError('Not running with the Werkzeug Server')
    shutdown_func()
    logging.info("Flask server shut down gracefully.")
    return jsonify({'status': 'success', 'message': 'Continuing test execution'}), 200

def run_flask_app():
    logging.info("Starting Flask test server on 0.0.0.0:5005")
    # 禁用 reloader & debugger；使用 app.run() 更稳定
    app.run(host='0.0.0.0', port=5005, debug=False, use_reloader=False)

2. Pytest 钩子（conftest.py 或插件模块）

import threading
import time
from test_server import run_flask_app

def pytest_collection_finish(session):
    # 启动 Flask 服务线程（非守护线程，确保可被 join）
    flask_thread = threading.Thread(target=run_flask_app, name="FlaskTestServer")
    flask_thread.start()

    # 等待服务真正就绪（简单轮询，生产环境建议用 requests.head 健康检查）
    time.sleep(1.5)

    # 发起触发请求（示例：使用 requests，需安装 pip install requests）
    try:
        import requests
        response = requests.get('http://localhost:5005/trigger', timeout=5)
        if response.status_code == 200:
            logging.info("Trigger sent successfully.")
    except Exception as e:
        logging.error(f"Failed to send trigger: {e}")

    # 关键：join 确保 Flask 线程自然结束（即收到 /trigger 后 shutdown 完成）
    flask_thread.join(timeout=10)  # 最大等待 10 秒
    if flask_thread.is_alive():
        logging.warning("Flask thread did not terminate cleanly.")

⚠️ 注意事项与最佳实践

禁止使用 daemon=True：守护线程无法保证资源清理，必须通过 join() 等待其自然退出。
端口选择策略：Jenkins 多任务并行时建议使用动态端口（如 port=0 让系统分配），或通过环境变量注入唯一端口（PORT=${BUILD_NUMBER}5005）。
超时防护：app.run() 是阻塞调用，/trigger 必须能到达；若网络隔离，改用 curl -s http://localhost:5005/trigger 在 shell 步骤中触发更可靠。
日志与可观测性：启用 logging 并持久化到文件，便于排查 Jenkins 中的静默失败。
替代方案考虑：对复杂场景，推荐使用轻量级异步服务器（如 hypercorn + asyncio）或专用测试工具（pytest-httpserver），避免手动管理 HTTP 生命周期。

通过以上改造，Flask 服务在 Jenkins 中将严格遵循「启动 → 等待触发 → 主动关闭 → 线程退出」流程，彻底规避端口复用错误，保障 CI 稳定性。

文中关于的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《Jenkins多线程启动Flask避坑全攻略》文章吧，也可关注golang学习网公众号了解相关技术文章。