在 gunicorn 部署 flask 应用时，`if __name__ == '__main__'` 块不会执行，导致后台线程无法启动；需将线程初始化逻辑移至模块顶层，并确保仅在主 worker 进程中运行，避免多进程重复创建。
在 Flask 应用中启用后台线程（如定时任务、长周期数据采集或 WebSocket 心跳推送）非常常见，但本地开发（python app.py）与生产部署（gunicorn app:app）的启动机制存在本质差异：
- 本地运行时，脚本作为主模块执行，__name__ == '__main__' 为 True，线程可正常启动；
- Gunicorn 加载的是模块对象（如 test:app），不执行 if __name__ == '__main__' 分支，因此线程初始化代码被完全跳过。
更关键的是：Gunicorn 默认启用多 worker 进程（如 --workers 4），若直接在模块顶层无条件启动线程，每个 worker 都会创建一份副本，不仅浪费资源，还可能导致竞态、重复事件推送（如多次发送相同 Socket.IO 消息）等严重问题。
✅ 正确做法是：
- 将后台线程启动逻辑移出 if __name__ == '__main__'；
- 利用 Gunicorn 的 on_starting 或 post_fork 钩子，或更稳妥地——通过检查 os.environ.get('SERVER_SOFTWARE') / os.getenv('GUNICORN_CMD_ARGS') 等标识，但最通用可靠的方式是判断当前是否为主 worker 进程**；
- 推荐使用 gunicorn 的 post_fork 钩子函数，并在其中对首个 fork 后的 worker（通常 PID 最小者）启动线程，或使用 threading.Lock + 进程级标志实现单例保障。
不过，对于多数场景，一个简洁、安全且兼容性强的方案是：在模块顶层启动线程，但仅当 Gunicorn 尚未 fork 子进程时（即主进程阶段）执行。由于 Gunicorn 主进程本身不处理请求，我们实际需要的是「在第一个工作进程启动后、首次处理请求前」初始化后台服务——此时应借助 worker_init_fn（如原代码所示），但注意该函数在每个 worker 初始化时都会调用，因此必须加锁或进程判别：
import threading
import logging
import os
from flask import Flask
from flask_socketio import SocketIO
app = Flask(__name__)
socketio = SocketIO(app, async_mode='gevent', cors_allowed_origins='*')
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
# 全局线程引用 & 启动锁
_background_thread = None
_thread_start_lock = threading.Lock()
def background_task():
logger.debug("Background task started in thread: %s", threading.current_thread().ident)
while True:
logger.debug("Background task running")
socketio.emit('background_task_response', {'data': 'Background Task Result'})
socketio.sleep(5)
def start_background_thread():
global _background_thread
with _thread_start_lock:
if _background_thread is None or not _background_thread.is_alive():
_background_thread = threading.Thread(
target=background_task,
name="Flask-Background-Worker",
daemon=True # 确保主进程退出时自动终止
)
_background_thread.start()
logger.info("Background thread started successfully.")
# ✅ 关键：在 Gunicorn worker 初始化时启动（每个 worker 调用一次，但线程全局唯一）
def worker_init(worker):
logger.debug(f"Worker {worker.pid} initializing...")
# 注意：此处仍可能并发，故依赖上面的锁机制
start_background_thread()
# ⚠️ 错误示范：不要在此处无条件启动！
# if __name__ != '__main__':
# start_background_thread() # ❌ 多 worker 会重复触发
# ✅ 正确暴露应用实例（供 Gunicorn 加载）
gunicorn_app = app然后使用以下命令启动（需安装 gevent 和 gevent-websocket）：
pip install gevent gevent-websocket flask-socketio gunicorn \ --bind "0.0.0.0:8000" \ --workers 2 \ --worker-class "geventwebsocket.gunicorn.workers.GeventWebSocketWorker" \ --preload \ # ? 关键！确保模块在 fork 前加载，使 worker_init 生效 --worker-init-fn "test:worker_init" \ test:gunicorn_app
? 重要说明与最佳实践：
- --preload 参数必不可少：它让 Gunicorn 在 fork 子进程前先导入并执行模块，确保 worker_init 钩子能被识别和调用；
- daemon=True 是必须的，防止后台线程阻塞 worker 进程退出；
- 若使用 sync worker（非 gevent），请改用 socketio.sleep() → time.sleep()，并确保 async_mode='threading'；
- 对于需严格单例的后台服务（如数据库连接池维护、全局缓存刷新），建议改用外部任务队列（Celery + Redis）或系统级守护进程，而非应用内线程；
- 日志中若看到多条 "Background thread started successfully."，说明锁未生效，请检查 worker_init 是否被正确注册及 --preload 是否启用。
综上，Gunicorn 下的后台线程不是“不能跑”，而是必须适配其多进程模型——通过钩子 + 线程安全控制，即可稳健支撑实时通知、状态轮询等典型需求。
