如何从 celery 任务获取实时标准输出？

Question

我有一个 Django 项目，其中我使用 celery 运行一些后台任务，我的目标之一是显示正在运行的任务的实时标准输出，以便用户可以看到进度。问题是我只有在完成执行时才获得进程的标准输出（所有内容都立即转储）。

视图.py

def run_script(request, file_type, file_id):
    run_features.delay(arguments)
    return

任务.py

@shared_task
def run_features(arguments):
    process = subprocess.Popen(arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)

        channel_layer = get_channel_layer()

        while True:
            output = process.stdout.readline()
            if output == '' and process.poll() is not None:
                break
            if output:
                print(output)
                async_to_sync(channel_layer.group_send)(
                    'output_group',
                    {
                        'type': 'send_output',
                        'output': output.strip()
                    }
                )

        process.wait()
        return process.returncode

芹菜信息

 -------------- celery@ritik v5.3.1 (emerald-rush)
--- ***** ----- 
-- ******* ---- Linux-6.5.0-25-generic-x86_64-with-glibc2.35 2024-03-21 09:30:54
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         Automation:0x7e2bb4073280
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery

期望 - 在任务运行时，我逐行获取标准输出。 实际结果 - 在子进程完成后，我立即得到标准输出。

另外，我不确定这是 django 问题还是 celery 问题。

感谢您的帮助！这是我的第一个 stackOverflow 问题。

Answer 1

找到解决方案了！只需使用 Pexpect.spawn() 而不是 subprocess.Popen()

# Start the process
child = pexpect.spawn(command, args, encoding='utf-8', timeout=180)

# Log the output
while True:
    try:
        line = child.readline()
        if not line:
            break
        print(line.strip())
    except pexpect.EOF:
        break

child.wait()
child.close()

return None