如何在Django模板中返回Celery任务结果以触发前端模态框

嘿，我完全懂这种窘境——同步模式下前端触发模态框顺顺当当，一换成Celery异步执行，任务结果乖乖躺进数据库，但前端就像聋了一样收不到信号，对吧？试过tornado-celery这类工具却发现依赖都没人维护了，确实头疼。下面给你几个靠谱的、不需要依赖冷门库的解决方案：

方案1：前端轮询（最简单，无额外依赖） #

这是最容易落地的方案，不需要改太多后端逻辑，前端定时发送请求查询任务状态，一旦检测到flag:True就触发模态框。

前端代码示例 #

// 页面加载时从后端模板获取任务ID（比如Django/Flask渲染页面时传入）
const taskId = "{{ task_id }}";
// 设置轮询间隔（比如每2秒查一次）
const pollTimer = setInterval(() => {
  fetch(`/api/check-task/${taskId}`)
    .then(res => res.json())
    .then(data => {
      if (data.flag === true) {
        // 任务完成，清除定时器并触发模态框
        clearInterval(pollTimer);
        document.getElementById('success-modal').style.display = 'block';
      } else if (data.status === 'failed') {
        // 处理任务失败场景
        clearInterval(pollTimer);
        alert('任务执行失败，请稍后重试');
      }
    })
    .catch(err => console.error('轮询任务状态出错:', err));
}, 2000);

后端代码示例（Flask为例） #

from celery.result import AsyncResult
from flask import jsonify, request

@app.route('/api/check-task/<task_id>')
def check_task_status(task_id):
    # 通过Celery的AsyncResult获取任务状态
    task_result = AsyncResult(task_id)
    if task_result.ready():
        if task_result.successful():
            # 假设任务返回的结果包含flag字段
            return jsonify({
                'flag': task_result.result.get('flag', False),
                'status': 'success'
            })
        else:
            return jsonify({
                'flag': False,
                'status': 'failed'
            })
    else:
        # 任务还在执行中
        return jsonify({
            'flag': False,
            'status': 'pending'
        })

方案2：Server-Sent Events（SSE，高效单向推送） #

SSE是浏览器原生支持的服务器推送技术，比轮询更高效，不需要频繁发送请求，服务器在任务完成后主动推送消息给前端。

后端代码示例（Flask为例） #

from flask import Response, stream_with_context
from celery.result import AsyncResult
import time

@app.route('/api/task-sse/<task_id>')
def task_sse(task_id):
    def event_stream():
        task_result = AsyncResult(task_id)
        # 等待任务完成
        while not task_result.ready():
            time.sleep(1)
            # 发送任务执行中的状态
            yield f"data: {{\"status\": \"pending\"}}\n\n"
        # 任务完成后推送结果
        if task_result.successful():
            yield f"data: {{\"flag\": true, \"status\": \"success\"}}\n\n"
        else:
            yield f"data: {{\"flag\": false, \"status\": \"failed\"}}\n\n"
    
    return Response(stream_with_context(event_stream()), mimetype='text/event-stream')

前端代码示例 #

const taskId = "{{ task_id }}";
// 创建SSE连接
const eventSource = new EventSource(`/api/task-sse/${taskId}`);

eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    if (data.flag === true) {
        // 关闭连接并触发模态框
        eventSource.close();
        document.getElementById('success-modal').style.display = 'block';
    } else if (data.status === 'failed') {
        eventSource.close();
        alert('任务执行失败');
    }
};

// 处理连接错误
eventSource.onerror = function(error) {
    console.error('SSE连接异常:', error);
    eventSource.close();
};

方案3：WebSocket（双向实时通信） #

如果你的项目后续有更多实时交互需求（比如任务进度更新），可以用WebSocket实现双向通信。这里以Flask-SocketIO为例：

后端代码示例 #

from flask import Flask
from flask_socketio import SocketIO, emit, join_room
from celery import Celery
from celery.signals import task_success

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")
celery = Celery(app.name, broker='redis://localhost:6379/0')

# 监听Celery任务成功信号，推送消息给前端
@task_success.connect
def on_task_success(sender=None, result=None, **kwargs):
    if result.get('flag', False):
        # 向对应任务ID的房间推送消息
        emit('task_completed', {'flag': True}, room=sender.id)

# 前端连接时加入对应任务ID的房间
@socketio.on('join_task_room')
def handle_join_room(data):
    task_id = data['task_id']
    join_room(task_id)

前端代码示例 #

const taskId = "{{ task_id }}";
const socket = io();

// 连接成功后加入任务房间
socket.on('connect', () => {
    socket.emit('join_task_room', { task_id: taskId });
});

// 监听任务完成事件
socket.on('task_completed', (data) => {
    if (data.flag === true) {
        socket.disconnect();
        document.getElementById('success-modal').style.display = 'block';
    }
});

关键注意事项 #

• Celery结果后端配置：确保你已经正确配置了Celery的结果后端（比如Redis或RabbitMQ），这样才能通过AsyncResult获取任务状态。
• 数据库替代方案：如果不想依赖Celery的结果后端，也可以在任务完成后将状态写入数据库，后端接口直接查询数据库状态即可。
• 跨域处理：如果前端和后端不在同一域名下，需要在后端配置跨域支持（比如Flask的CORS扩展）。

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