如何检查Celery/Supervisor是否使用Python运行

Question

如何在Python中编写一个脚本,如果芹菜在机器上运行(Ubuntu)？

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

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

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

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...')