如何在 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 稳定性。