python - 在 celery 中流式传输结果

Question

我正在尝试使用 celery 在一组服务器上安排和运行任务。每个任务运行时间都比较长(几个小时)，并且涉及使用子进程调用具有给定输入的特定程序。该程序在 stdout 和 stderr 中产生大量输出。

有没有办法近乎实时地向客户端显示程序产生的输出？流式传输输出，以便客户端无需登录服务器即可观看服务器上运行的任务产生的输出？

Answer 1

您没有指定很多要求和约束。我假设您已经在某处拥有一个 Redis 实例。

你可以做的是逐行读取其他进程的输出并通过redis发布:

这是一个示例，您可以在其中将数据echo 到文件/tmp/foo 中以进行测试:

import redis
redis_instance = redis.Redis()
p = subprocess.Popen(shlex.split("tail -f /tmp/foo"), stdout=subprocess.PIPE)

while True:
    line = p.stdout.readline()
    if line:
        redis_instance.publish('process log', line)
    else:
        break

在一个单独的进程中:

import redis

redis_instance = redis.Redis()
pubsub = redis_instance.pubsub()
pubsub.subscribe('process log')

while True:
    for message in pubsub.listen():
        print message  #  or use websockets to comunicate with a browser

如果您希望流程结束，您可以例如 celery 任务完成后发送“退出”。

您可以使用不同的 channel (subscribe 中的字符串)来分隔不同进程的输出。

如果你愿意，你也可以将你的日志输出存储在redis中，

redis_instance.rpush('process log', message)

然后完整地检索它。