我正在使用Celery来管理异步任务。然而,偶尔芹菜过程会失效,导致任何任务都无法执行。我希望能够检查芹菜的状态并确保一切正常,如果我发现任何问题,则向用户显示错误消息。从Celery Worker文档来看,我可能会使用ping或inspect,但ping感觉很乱,并且不清楚究竟如何使用inspect(如果inspect()。registered()是空的?)。
任何有关这方面的指导将不胜感激。基本上我正在寻找的方法是这样的:
def celery_is_alive():
from celery.task.control import inspect
return bool(inspect().registered()) # is this right??
编辑:它看起来甚至不像celery 2.3.3上的registered()(即使2.1文档列出它)。也许ping是正确的答案。
编辑:Ping似乎也没有做我认为会做的事情,所以仍然不确定这里的答案。
这是我一直在使用的代码。 celery.task.control.Inspect.stats()
返回一个包含有关当前可用工作人员的大量详细信息的dict,如果没有工作人员正在运行,则返回None;如果无法连接到消息代理,则会引发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
来自the documentation of celery 4.2:
from your_celery_app import app
def get_celery_worker_status():
i = app.control.inspect()
stats = i.stats()
registered_tasks = i.registered()
active_tasks = i.active()
scheduled_tasks = i.scheduled()
result = {
'stats': stats,
'registered_tasks': registered_tasks,
'active_tasks': active_tasks,
'scheduled_tasks': scheduled_tasks
}
return result
当然你可以/应该通过错误处理改进代码......
以下对我有用:
import socket
from kombu import Connection
celery_broker_url = "amqp://localhost"
try:
conn = Connection(celery_broker_url)
conn.ensure_connection(max_retries=3)
except socket.error:
raise RuntimeError("Failed to connect to RabbitMQ instance at {}".format(celery_broker_url))
要在celery作为守护进程运行的情况下使用命令行检查相同的内容,
celery -A [app_name] status
测试是否有任何工作人员正在响应的一种方法是发送'ping'广播并在第一个响应上返回成功的结果。
from .celery import app # the celery 'app' created in your project
def is_celery_working():
result = app.control.broadcast('ping', reply=True, limit=1)
return bool(result) # True if at least one result
这会播放“ping”并等待一秒钟的响应。一旦第一个响应进入,它将返回一个结果。如果你想更快地获得False
结果,你可以添加一个timeout
参数来减少它在放弃之前等待的时间。