Flask后台线程搭配Gunicorn部署：后台任务未启动问题排查

本地运行python3 test.py时，脚本作为主模块执行，if __name__ == '__main__':分支内的后台线程启动代码会被触发；但用Gunicorn部署时，它是直接导入你的app对象，此时模块的__name__不等于__main__，该分支代码完全不会运行，导致后台任务没启动。另外你配置了GeventWebSocketWorker但启动命令未指定，默认的sync worker不兼容SocketIO的长连接特性。

1. 重构后台任务启动逻辑 #

把后台任务的启动逻辑从__main__块移到Worker初始化函数中，确保每个Gunicorn Worker启动时都能触发任务：

修改后的完整代码：

import multiprocessing
import threading
import logging
from flask import Flask
from flask_socketio import SocketIO

app = Flask(__name__)
# 根据实际需求配置跨域
socketio = SocketIO(app, cors_allowed_origins="*")

logging.basicConfig(level=logging.DEBUG)

@app.route('/')
def index():
    return "OK"

def background_task():
    logging.debug("Background task started in thread: %s", threading.current_thread().ident)
    while True:
        logging.debug("Background task running")
        socketio.emit('background_task_response', {'data': 'Background Task Result'})
        socketio.sleep(5)

def worker_init(worker):
    logging.debug("Worker initialized with pid: %d", worker.pid)
    # 用SocketIO原生方法启动后台任务，避免线程兼容性问题
    socketio.start_background_task(target=background_task)

if __name__ == '__main__':
    bind_host, bind_port = "0.0.0.0", 8000
    # 本地运行时启动后台任务
    socketio.start_background_task(target=background_task)
    socketio.run(app, host=bind_host, port=bind_port)

2. 执行正确的Gunicorn启动命令 #

必须指定兼容WebSocket的Worker类和初始化函数：

gunicorn test:app --worker-class=geventwebsocket.gunicorn.workers.GeventWebSocketWorker --worker-init=test.worker_init --bind=0.0.0.0:8000 --workers=2

• --worker-class：指定支持WebSocket的Gevent Worker
• --worker-init：指定Worker启动时调用的初始化函数
• --workers：根据服务器CPU核心数调整，若后台任务不需要多实例，可设为1

关键注意事项 #

• 若启用多Worker，每个Worker都会启动独立的后台线程，若任务不需要重复执行，建议将--workers设为1，或通过共享存储/锁机制避免重复逻辑
• 确保安装所有依赖：pip install flask flask-socketio gevent gevent-websocket gunicorn

内容的提问来源于stack exchange，提问作者Noki