如何将Celery任务的stdout流式传输至Web应用并实时展示

核心逻辑拆解 #

要实现Web端的实时日志监控，需要打通三个环节，也是之前调研方案缺失的关键部分：

• 精准捕获子进程（xyz.py）的逐行stdout输出，避免等任务跑完一次性读取
• 给每个任务实例的日志打唯一标记，做带过期的实时存储，避免不同任务的日志混叠
• 给Web端提供增量拉取日志的轻量接口，实现低延迟的监控效果

为什么之前的调研方案达不到实时要求 #

• Celery内置日志链路：Celery的日志系统只捕获主worker进程、Celery任务函数内的print/日志输出，你用subprocess.Popen起的xyz.py子进程默认继承worker的终端stdout，根本不经过Celery的日志管道，自然没法通过日志配置抓到对应的输出；而且Celery的日志默认带块缓冲，就算能抓到，也会有数十秒的延迟
• Flower监控能力：Flower是通过订阅Celery的事件总线，只收集任务的元数据（开始/结束状态、运行时长、返回值），不会钩子捕获子进程的stdout，只能看到任务跑没跑完，看不到中间的逐行输出

分步实现代码 #

1. 改造Celery任务实现实时日志捕获 #

首先要调整Popen的参数，关闭子进程的输出缓冲，逐行读取stdout，把日志实时写入带任务唯一标识的存储位置（这里选Redis，因为结构轻、性能够，适合存短周期的追加式日志）：

import subprocess
import redis
from celery import Celery

# 初始化Redis客户端，用于存储实时任务日志
redis_client = redis.Redis(host='你的Redis地址', port=6379, db=0)

@celery.task(name='my_long_task', bind=True)
def my_long_task(self, arg_1):
    # 获取当前任务的唯一ID，作为日志的存储key
    task_id = self.request.id
    log_key = f"celery:stream:log:{task_id}"
    # 清空旧key，设置12小时过期时间，覆盖最长数小时的任务运行周期，避免内存泄漏
    redis_client.delete(log_key)
    redis_client.expire(log_key, 43200)

    # 改造Popen参数，关闭缓冲、捕获输出到管道
    proc = subprocess.Popen(
        # 给python命令加-u参数，完全关闭子进程的stdout块缓冲，保证输出实时刷出
        "python -u xyz.py",
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        bufsize=1,  # 行缓冲模式，读一行处理一行
        text=True  # 直接输出字符串，避免手动解码
    )

    # 逐行读取子进程输出，边读边写Redis
    for line in proc.stdout:
        line = line.strip()
        if line:
            redis_client.rpush(log_key, line)

    # 等待子进程执行完成
    exit_code = proc.wait()
    # 写入任务结束标记，供前端判断是否停止拉取
    redis_client.rpush(log_key, f"__TASK_FINISHED__ 退出码: {exit_code}")
    return exit_code

2. 改造Flask提供增量日志拉取接口 #

接口支持按上次拉取的偏移量返回新日志，避免每次请求全量读取所有历史内容，降低性能开销：

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.route('/task/<task_id>/stream-log')
def fetch_task_log(task_id):
    log_key = f"celery:stream:log:{task_id}"
    # 从请求参数获取上次已经拉取到的日志偏移量，默认从第0条开始
    offset = int(request.args.get('offset', 0))
    # 从Redis List里拉取偏移量之后的所有新日志
    new_logs = redis_client.lrange(log_key, offset, -1)
    log_list = [log.decode('utf-8') for log in new_logs]
    
    return jsonify({
        "logs": log_list,
        "next_offset": offset + len(log_list),
        # 判断日志里有没有结束标记，标记前端是否可以停止轮询
        "task_running": "__TASK_FINISHED__" not in log_list
    })

注意：在start_background_task函数里调用my_long_task.delay(arg_1)时，返回的AsyncResult对象自带当前任务的唯一ID，把这个ID直接返回给前端，前端就可以用这个ID请求日志拉取接口，完成监控链路打通。

如果要降低监控的等待延迟，不需要前端频繁轮询，可以把这个接口替换成Server-Sent Events（SSE）的流式响应，直接在服务端把实时日志推给前端，性能比WebSocket更轻量，完全适配日志单向推送的场景。

关键避坑点 #

• 一定要关闭子进程的输出缓冲：如果不给python加-u参数，哪怕逐行读管道，子进程默认会攒够4K的输出块才刷一次stdout，会出现日志隔几分钟一次性跳出几十行的情况，完全失去实时性
• 日志必须和任务ID强绑定：不要把不同任务的日志存在同一个存储key里，否则多个任务同时运行时，输出会混在一起没法区分
• 必须设置日志的过期时间：如果不清理旧的任务日志，存储里的内容会越攒越多，慢慢占满内存
• 避免存超大单行日志：如果xyz.py会输出超长单行内容，写入存储时要做长度截断，避免单条记录占用过大内存

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