Python中如何在Celery任务中运行Flask-SocketIO？

在Celery任务中运行Flask-SocketIO的核心挑战是：Celery工作进程是独立的，无法直接访问主Flask应用的SocketIO实例。你需要通过消息队列（如Redis）来桥接它们。

这里的关键是使用socketio.AsyncRedisManager或socketio.KombuManager，让Celery任务能向同一个消息队列发送事件，然后由Flask-SocketIO服务器接收并广播给客户端。

下面是一个完整示例：

首先，安装必要的包：

pip install flask flask-socketio celery redis eventlet

项目结构：

your_project/
├── app.py          # Flask主应用
├── celery_app.py   # Celery配置
└── tasks.py        # Celery任务

1. Celery配置 (celery_app.py):

from celery import Celery

celery_app = Celery(
    'tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/0'
)

2. Flask应用与SocketIO配置 (app.py):

from flask import Flask
from flask_socketio import SocketIO
import eventlet
eventlet.monkey_patch()

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

# 使用Redis作为消息队列
socketio = SocketIO(
    app,
    message_queue='redis://localhost:6379/0',
    cors_allowed_origins="*"
)

@app.route('/')
def index():
    return 'Flask-SocketIO Server'

@socketio.on('connect')
def handle_connect():
    print('Client connected')

if __name__ == '__main__':
    socketio.run(app, debug=True)

3. Celery任务 (tasks.py):

from celery_app import celery_app
from flask_socketio import SocketIO
import eventlet
eventlet.monkey_patch()

# 创建独立的SocketIO实例用于发射事件
socketio = SocketIO(message_queue='redis://localhost:6379/0')

@celery_app.task
def background_task(data):
    """在Celery任务中发射SocketIO事件"""
    # 执行一些耗时操作...
    result = f"Processed: {data}"
    
    # 通过消息队列发射事件到Flask-SocketIO服务器
    socketio.emit('task_update', 
                 {'data': result, 'status': 'completed'},
                 namespace='/')
    
    return result

4. 启动服务：

• 启动Redis：redis-server
• 启动Flask应用：python app.py
• 启动Celery worker：celery -A tasks worker --loglevel=info

5. 客户端示例 (HTML):

<script src="https://cdn.socket.io/4.5.0/socket.io.min.js"></script>
<script>
    const socket = io('http://localhost:5000');
    
    socket.on('task_update', function(data) {
        console.log('Task update:', data);
        document.getElementById('output').innerHTML = 
            `Result: ${data.data}, Status: ${data.status}`;
    });
    
    // 触发Celery任务
    function startTask() {
        fetch('/start-task', { method: 'POST' });
    }
</script>
<button onclick="startTask()">Start Task</button>
<div id="output"></div>

工作原理：

1. Celery任务中的socketio.emit()将事件发送到Redis消息队列
2. Flask-SocketIO服务器从同一个Redis队列接收到事件
3. 服务器将事件广播给所有连接的客户端

关键点：

• 必须使用相同的消息队列URL（这里是Redis）
• 在Celery任务中创建独立的SocketIO实例，只用于发射事件
• 确保安装了eventlet或gevent用于异步支持

一句话总结： 用Redis消息队列桥接Celery任务和Flask-SocketIO服务器。