Flask SocketIO+Celery+RabbitMQ环境下Celery任务阻塞问题求助

问题现象 #

使用Flask-SocketIO构建应用，通过Celery运行后台任务向Socket连接推送消息。Celery在成功执行2-3个任务后，无法从RabbitMQ队列获取新任务，Worker失联导致队列阻塞。

核心配置与代码 #

SocketIO、Flask及Celery初始化（config/app_config.py） #

from gevent import monkey
monkey.patch_all()
from celery import Celery
from celery.signals import setup_logging
from threading import local
import logging.config
import os
from logstash.formatter import LogstashFormatterBase
from logging.config import dictConfig
from flask import Flask
from flask_socketio import SocketIO
from flask_cors import CORS

_locals = local()

app = Flask(__name__)

CORS(app)

BASE_DIR = os.getcwd()
LOGS_DIR = os.path.join(BASE_DIR, "logs")

RMQ_BROKER_URL = os.environ.get("RMQ_URL", "")
RMQ_SSL = True if "amqps" in RMQ_BROKER_URL else False
MACHINE_HOST = os.environ.get("HOSTNAME", "")

celery = Celery(
    Flask(__name__).name,
    broker=RMQ_BROKER_URL
)
celery.conf.worker_prefetch_multiplier = 1
celery.conf.task_acks_late = False
celery.conf.update({
    'worker_hijack_root_logger': False,
    'worker_redirect_stdouts': False,
})

socket_io_rmq_list = RMQ_BROKER_URL.split("/")
socket_io_rmq_list[-1] = socket_io_rmq_list[-1] + "_socketio"
socket_io_rmq_url = "/".join(socket_io_rmq_list)

socketio = SocketIO(
    cors_allowed_origins="*",
    async_mode='gevent',
    transports=['websocket'],
    message_queue=socket_io_rmq_url
)

Gunicorn启动命令 #

gunicorn --preload --workers 10 --worker-class gevent --timeout 600 --log-level debug --access-logfile logs/gunicorn_access.log --error-logfile logs/gunicorn_error.log main:app --bind 0.0.0.0:5000 &

Celery Worker启动命令（Docker环境） #

celery -A main.celery worker --without-heartbeat --without-gossip --without-mingle --loglevel=debug --logfile=logs/celery_worker.log --concurrency=1 -n "${HOSTNAME}_METRIC_REFRESH_NEXTGEN" --max-tasks-per-child=1

Flask入口文件（main.py） #

from api.views import app, socketio
from config.app_config import celery

if __name__ == '__main__':
    socketio.run(app, host="0.0.0.0", debug=False, threaded=False)

代码结构 #

│   Dockerfile
│   docker_entrypoint.sh
│   main.py
│   requirements.txt
│   __init__.py
│
├───api
│   │   auth.py
│   │   connectors.py
│   │   constants.py
│   │   tasks.py
│   │   views.py
│   │   __init__.py
│   
│
├───config
│   │   app_config.py
│   │   exceptions.py
│   │   utils.py
│   │   __init__.py
│   
│
├───entities
│       metric_refresh.py
│       metric_validations.py
│       sample_template.py
│       __init__.py

结构说明 #

• SocketIO、Flask应用及Celery实例均在config/app_config.py中初始化，该文件顶部执行了gevent猴子补丁
• 实例被导入到api/views.py（处理Flask路由、Socket连接/断开逻辑）和api/tasks.py（后台任务向Socket推送消息）中

排查线索 #

• 未集成SocketIO的服务中，Celery与RabbitMQ交互正常，任务启动时会发送确认（acks_late=False）
• 集成SocketIO后，Celery执行1-2次任务后无法向RabbitMQ发送确认，Worker失联导致队列阻塞

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