本文详解在 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 进程正是罪魁祸首。
✅ 正确做法：主动关闭 + 同步等待 + 避免调试模式
原方案存在三处关键风险：
SkillSub Pro - Python 题解与代码注释双功能技能功能概述SkillSub Pro - Python 题解与代码注释双功能技能是一项面向实际任务的技能，主要用于SkillSub Pro 是一个 Python 题解生成与代码注释的 双功能合体技能 ，专为学生、算法学习者和开发者设计；✅ 一个技能，两种用途 ：；核心要点📝 题解模式 ：输入题目/题号，自动生成完整 Python 题解（含详细注释、解题思路、复杂度分析）；💬 注释模式 ：输入 Python 代码，自动添加详细中。它将相关步骤、
- 使用 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 稳定性。
