如何检索队列中尚未处理的任务列表?
当前回答
要从后端检索任务,使用这个
from amqplib import client_0_8 as amqp
conn = amqp.Connection(host="localhost:5672 ", userid="guest",
password="guest", virtual_host="/", insist=False)
chan = conn.channel()
name, jobs, consumers = chan.queue_declare(queue="queue_name", passive=True)
其他回答
这在我的申请中很奏效:
def get_celery_queue_active_jobs(queue_name):
connection = <CELERY_APP_INSTANCE>.connection()
try:
channel = connection.channel()
name, jobs, consumers = channel.queue_declare(queue=queue_name, passive=True)
active_jobs = []
def dump_message(message):
active_jobs.append(message.properties['application_headers']['task'])
channel.basic_consume(queue=queue_name, callback=dump_message)
for job in range(jobs):
connection.drain_events()
return active_jobs
finally:
connection.close()
Active_jobs将是一个字符串列表,对应于队列中的任务。
不要忘记将CELERY_APP_INSTANCE替换为您自己的。
感谢@ashish在这里为我指出了正确的方向:https://stackoverflow.com/a/19465670/9843399
如果你正在使用rabbitMQ,在终端中使用这个:
sudo rabbitmqctl list_queues
它将打印带有挂起任务数量的队列列表。例如:
Listing queues ...
0b27d8c59fba4974893ec22d478a7093 0
0e0a2da9828a48bc86fe993b210d984f 0
10@torob2.celery.pidbox 0
11926b79e30a4f0a9d95df61b6f402f7 0
15c036ad25884b82839495fb29bd6395 1
celerey_mail_worker@torob2.celery.pidbox 0
celery 166
celeryev.795ec5bb-a919-46a8-80c6-5d91d2fcf2aa 0
celeryev.faa4da32-a225-4f6c-be3b-d8814856d1b6 0
右边列的数字是队列中的任务数。在上面,芹菜队列有166个待处理的任务。
据我所知,芹菜没有提供API来检查队列中等待的任务。这是特定于代理的。例如,如果你使用Redis作为代理,那么检查在芹菜(默认)队列中等待的任务就像这样简单:
连接到代理 在芹菜列表中列出项目(以LRANGE命令为例)
请记住,这些任务等待可用的员工来挑选。您的集群可能有一些正在运行的任务——这些任务不会在这个列表中,因为它们已经被选中了。
检索特定队列中的任务的过程是特定于代理的。
如果你正在使用芹菜+Django,最简单的方法是在虚拟环境中直接从终端使用命令检查任务,或者使用芹菜的完整路径:
道格:http://docs.celeryproject.org/en/latest/userguide/workers.html?highlight=revoke inspecting-workers
$ celery inspect reserved
$ celery inspect active
$ celery inspect registered
$ celery inspect scheduled
另外,如果你正在使用芹菜+RabbitMQ,你可以使用下面的命令检查队列列表:
更多信息:https://linux.die.net/man/1/rabbitmqctl
$ sudo rabbitmqctl list_queues
from celery.task.control import inspect
def key_in_list(k, l):
return bool([True for i in l if k in i.values()])
def check_task(task_id):
task_value_dict = inspect().active().values()
for task_list in task_value_dict:
if self.key_in_list(task_id, task_list):
return True
return False
推荐文章
- 证书验证失败:无法获得本地颁发者证书
- 当使用pip3安装包时,“Python中的ssl模块不可用”
- 无法切换Python与pyenv
- Python if not == vs if !=
- 如何从scikit-learn决策树中提取决策规则?
- 为什么在Mac OS X v10.9 (Mavericks)的终端中apt-get功能不起作用?
- 将旋转的xtick标签与各自的xtick对齐
- 为什么元组可以包含可变项?
- 如何合并字典的字典?
- 如何创建类属性?
- 不区分大小写的“in”
- 在Python中获取迭代器中的元素个数
- 解析日期字符串并更改格式
- 使用try和。Python中的if
- 如何在Python中获得所有直接子目录