python - 如何使用 Python 检查 Celery/Supervisor 是否正在运行

Question

如果 celery 在机器(Ubuntu)上运行，如何用 Python 编写脚本输出？

我的用例。我有一个包含一些任务的简单 python 文件。我没有使用 Django 或 Flask。我使用主管来运行任务队列。例如，

tasks.py

from celery import Celery, task
app = Celery('tasks')
@app.task()
def add_together(a, b):
    return a + b

主管:

[program:celery_worker]
directory = /var/app/
command=celery -A tasks worker info

这一切都有效，我现在想要有一个页面来检查 celery/supervisor 进程是否正在运行。即这样的事情可能使用 Flask 允许我托管页面提供 200 状态允许我进行负载平衡。

例如...

check_status.py

from flask import Flask

app = Flask(__name__)

@app.route('/')
def status_check():

    #check supervisor is running
    if supervisor:
         return render_template('up.html')
    else:
        return render_template('down.html')

if __name__ == '__main__':
    app.run()

Answer 1

更新 09/2020:Jérôme 在此处更新了 Celery 4.3 的此答案:https://stackoverflow.com/a/57628025/1159735

您可以通过导入celery.bin.celery包通过代码运行celery status命令:

import celery
import celery.bin.base
import celery.bin.celery
import celery.platforms

app = celery.Celery('tasks', broker='redis://')

status = celery.bin.celery.CeleryCommand.commands['status']()
status.app = status.get_app()

def celery_is_up():
    try:
        status.run()
        return True
    except celery.bin.base.Error as e:
        if e.status == celery.platforms.EX_UNAVAILABLE:
            return False
        raise e

if __name__ == '__main__':
    if celery_is_up():
        print('Celery up!')
    else:
        print('Celery not responding...')