Django 检测 Celery 是否可用正在运行

发布于12月15日

我使用Celery来管理异步任务.但是，有时Celery 进程会关闭，从而导致不执行任何任务.我希望能够判断Celery 的状态，并确保一切工作正常，如果我检测到任何问题，将向用户显示错误消息.从Celery 工人文档看，我似乎可以使用ping或inspect来完成这项任务，但是ping感觉很粗糙，并且不清楚到底应该如何使用Inspect()(如果Inspect().register()为空？).

任何关于这方面的指导都将不胜感激.基本上我要找的方法是这样的:

def celery_is_alive():
    from celery.task.control import inspect
    return bool(inspect().registered()) # is this right??

编辑:它看起来甚至不像是在Celery 2.3.3上提供的注册()(尽管2.1文档中列出了它).也许平是正确的答案.

编辑:ping似乎也没有做我认为它会做的事情，所以仍然不确定这里的答案.

推荐答案

这是我一直在用的代码.celery.task.control.Inspect.stats()将返回一个包含有关当前可用工作进程的大量详细信息的字典，如果没有工作进程正在运行，则为None；如果无法连接到message broker，则返回IOError.我使用的是RabbitMQ-其他消息传递系统的行为可能会略有不同.这在Celery 2.3.x和2.4.x中有效；我不确定它可以追溯到什么时候.

def get_celery_worker_status():
    ERROR_KEY = "ERROR"
    try:
        from celery.task.control import inspect
        insp = inspect()
        d = insp.stats()
        if not d:
            d = { ERROR_KEY: 'No running Celery workers were found.' }
    except IOError as e:
        from errno import errorcode
        msg = "Error connecting to the backend: " + str(e)
        if len(e.args) > 0 and errorcode.get(e.args[0]) == 'ECONNREFUSED':
            msg += ' Check that the RabbitMQ server is running.'
        d = { ERROR_KEY: msg }
    except ImportError as e:
        d = { ERROR_KEY: str(e)}
    return d

Django 检测 Celery 是否可用正在运行

推荐答案

Django相关问答推荐

Django、htmx删除记录并更新

当使用django-tinymce时，我在哪里指定referer？

Django：如何 Select 没有其他项目引用的项目？

测试 - 两个类共享相同的登录代码，但它只能在一个类中工作

Django多对多关系报错：<title> object (None)>需要先设置id字段的值.

Django Rest Framework 认证类覆盖 request.user

使用 Pytest 在 Django 中编写测试用例时如何加载特定目录中的所有固定装置

Django 模型命名约定

Django判断字段是否为空？

Django 1.9：字段与父模型中不存在的字段的字段冲突

在 django 中获取本地时区

如何获取 Django 模型来自的应用程序？

magic有什么问题？

无效的命令 WSGIDaemonProcess 在 CentOS 6.7 上部署 Django 应用程序

如何在 Django 1.9 中删除 DB (sqlite3) 以从头开始？

在 django admin 中链接到外键对象

Django：以 10 为底的 int() 的无效文字

在 Django 中使用 LiveServerTestCase 时使用代码登录

django select_related - 何时使用它

在 Django 中使用 select_related Select 特定字段