如何将Celery任务的终端输出展示至前端？

方案一：任务完成后返回输出（适合短任务） #

直接修改Celery任务，把命令执行的stdout和stderr作为返回值，前端通过任务ID查询最终结果。

修改任务代码：

@shared_task(bind=True)
def trigger_model(self, main_config, runmode_config, modelreport):
    with cd("/another_project"):
        spark_submit_str = "python3 run_trigger.py " + "/this_project" + main_config

        process = subprocess.Popen(spark_submit_str, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                   universal_newlines=True, shell=True)

        stdout, stderr = process.communicate()
        # 整理输出内容，失败时返回错误信息
        output_content = stdout if process.returncode == 0 else stderr
        return {
            "returncode": process.returncode,
            "output": output_content
        }

前端流程：

1. 触发任务时，保存Celery返回的task_id
2. 定时调用后端接口，通过task_id调用AsyncResult(task_id).get()获取任务结果
3. 将拿到的输出内容渲染到前端页面

方案二：实时推送输出（适合长任务） #

如果任务执行耗时久，需要实时展示输出，可结合Redis+WebSocket实现实时推送：

1. 修改Celery任务，实时推送输出 #

改用逐行读取子进程输出，将每一行内容推送到Redis频道：

import redis
from celery import shared_task

# 初始化Redis连接（根据你的实际配置调整）
redis_client = redis.Redis(host='localhost', port=6379, db=0)

@shared_task(bind=True)
def trigger_model(self, main_config, runmode_config, modelreport):
    task_id = self.request.id
    with cd("/another_project"):
        spark_submit_str = "python3 run_trigger.py " + "/this_project" + main_config

        # 合并stdout和stderr，统一读取
        process = subprocess.Popen(spark_submit_str, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                                   universal_newlines=True, shell=True)
        
        # 逐行读取输出并推送
        for line in iter(process.stdout.readline, ''):
            if line:
                redis_client.publish(f"task_output_{task_id}", line.strip())
        
        process.wait()
        # 推送任务结束标记
        redis_client.publish(f"task_output_{task_id}", f"任务执行完成，返回码：{process.returncode}")

2. 后端WebSocket服务 #

以Flask-SocketIO为例，实现WebSocket服务订阅Redis频道，将输出推送给前端：

from flask import Flask
from flask_socketio import SocketIO, join_room, emit
import redis
from threading import Thread

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def redis_message_listener():
    pubsub = redis_client.pubsub()
    pubsub.psubscribe('task_output_*')
    for message in pubsub.listen():
        if message['type'] == 'pmessage':
            channel = message['channel'].decode('utf-8')
            task_id = channel.split('_')[-1]
            output_line = message['data'].decode('utf-8')
            socketio.emit('real_time_output', {'line': output_line}, room=task_id)

# 启动Redis监听线程
Thread(target=redis_message_listener, daemon=True).start()

@socketio.on('subscribe_task')
def handle_subscribe(data):
    task_id = data['task_id']
    join_room(task_id)
    emit('tip', f"已开始接收任务 {task_id} 的实时输出")

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

3. 前端实现 #

用SocketIO客户端连接后端，订阅指定任务的实时输出：

import { io } from "socket.io-client";

const socket = io("http://你的后端服务地址");

// 订阅任务输出
function subscribeToTask(taskId) {
  socket.emit('subscribe_task', { task_id: taskId });
}

// 监听实时输出并渲染
socket.on('real_time_output', (data) => {
  const outputContainer = document.getElementById('task-output');
  outputContainer.innerHTML += `<div>${data.line}</div>`;
});

socket.on('tip', (msg) => {
  console.log(msg);
});

方案三：通过Celery日志查询输出 #

将任务输出写入Celery日志，后端提供接口根据任务ID过滤日志内容返回给前端：

修改任务代码，使用Celery内置logger记录输出：

from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(bind=True)
def trigger_model(self, main_config, runmode_config, modelreport):
    with cd("/another_project"):
        spark_submit_str = "python3 run_trigger.py " + "/this_project" + main_config

        process = subprocess.Popen(spark_submit_str, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                   universal_newlines=True, shell=True)

        stdout, stderr = process.communicate()
        if process.returncode != 0:
            logger.error(f"任务[{self.request.id}]执行失败：{stderr}")
        else:
            logger.info(f"任务[{self.request.id}]输出：{stdout}")
        return {"task_id": self.request.id}

后端实现日志查询接口：读取Celery的日志文件（比如celery.log），根据task_id匹配对应的日志条目，整理后返回给前端展示。

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