Как я могу получить список задач в очереди, которые еще не обработаны?
Получить список задач в очереди в Сельдерей
Ответ 1
EDIT: см. другие ответы для получения списка задач в очереди.
Вы должны посмотреть здесь: Руководство по сельдерею - проверка работников
В основном это:
>>> from celery.task.control import inspect
# Inspect all nodes.
>>> i = inspect()
# Show the items that have an ETA or are scheduled for later processing
>>> i.scheduled()
# Show tasks that are currently active.
>>> i.active()
# Show tasks that have been claimed by workers
>>> i.reserved()
В зависимости от того, что вы хотите
Ответ 2
если вы используете rabbitMQ, используйте это в терминале:
sudo rabbitmqctl list_queues
он распечатает список очередей с количеством ожидающих задач. например:
Listing queues ...
0b27d8c59fba4974893ec22d478a7093 0
0e0a2da9828a48bc86fe993b210d984f 0
[email protected] 0
11926b79e30a4f0a9d95df61b6f402f7 0
15c036ad25884b82839495fb29bd6395 1
[email protected] 0
celery 166
celeryev.795ec5bb-a919-46a8-80c6-5d91d2fcf2aa 0
celeryev.faa4da32-a225-4f6c-be3b-d8814856d1b6 0
число в правом столбце - это количество задач в очереди. В приведенном выше примере очередь сельдерея имеет 166 ожидающих задач.
Ответ 3
Если вы не используете приоритетные задачи, это на самом деле довольно просто, если вы используете Redis. Чтобы получить значение задачи:
redis-cli -h HOST -p PORT -n DATABASE_NUMBER llen QUEUE_NAME
Но в приоритетных задачах используется другой ключ в redis, поэтому полная картина немного сложнее. Полная картина такова, что вам нужно запросить redis для каждого приоритета задачи. В python (и из проекта Flower) это выглядит так:
PRIORITY_SEP = '\x06\x16'
DEFAULT_PRIORITY_STEPS = [0, 3, 6, 9]
def make_queue_name_for_pri(queue, pri):
"""Make a queue name for redis
Celery uses PRIORITY_SEP to separate different priorities of tasks into
different queues in Redis. Each queue-priority combination becomes a key in
redis with names like:
- batch1\x06\x163 <-- P3 queue named batch1
There more information about this in Github, but it doesn't look like it
will change any time soon:
- https://github.com/celery/kombu/issues/422
In that ticket the code below, from the Flower project, is referenced:
- https://github.com/mher/flower/blob/master/flower/utils/broker.py#L135
:param queue: The name of the queue to make a name for.
:param pri: The priority to make a name with.
:return: A name for the queue-priority pair.
"""
if pri not in DEFAULT_PRIORITY_STEPS:
raise ValueError('Priority not in priority steps')
return '{0}{1}{2}'.format(*((queue, PRIORITY_SEP, pri) if pri else
(queue, '', '')))
def get_queue_length(queue_name='celery'):
"""Get the number of tasks in a celery queue.
:param queue_name: The name of the queue you want to inspect.
:return: the number of items in the queue.
"""
priority_names = [make_queue_name_for_pri(queue_name, pri) for pri in
DEFAULT_PRIORITY_STEPS]
r = redis.StrictRedis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
db=settings.REDIS_DATABASES['CELERY'],
)
return sum([r.llen(x) for x in priority_names])
Если вы хотите получить реальную задачу, вы можете использовать что-то вроде:
redis-cli -h HOST -p PORT -n DATABASE_NUMBER lrange QUEUE_NAME 0 -1
Оттуда вам придется десериализовать возвращенный список. В моем случае я смог сделать это с помощью чего-то вроде:
r = redis.StrictRedis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
db=settings.REDIS_DATABASES['CELERY'],
)
l = r.lrange('celery', 0, -1)
pickle.loads(base64.decodestring(json.loads(l[0])['body']))
Просто имейте в виду, что десериализация может занять некоторое время, и вам нужно настроить приведенные выше команды для работы с различными приоритетами.
Ответ 4
Чтобы получить задания из бэкэнд, используйте этот
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)
Ответ 5
Решение для копирования и вставки Redis с сериализацией json:
def get_celery_queue_items(queue_name):
import base64
import json
# Get a configured instance of a celery app:
from yourproject.celery import app as celery_app
with celery_app.pool.acquire(block=True) as conn:
tasks = conn.default_channel.client.lrange(queue_name, 0, -1)
decoded_tasks = []
for task in tasks:
j = json.loads(task)
body = json.loads(base64.b64decode(j['body']))
decoded_tasks.append(body)
return decoded_tasks
Это работает с Джанго. Только не забудьте изменить yourproject.celery.
Ответ 6
Если вы используете Celery + Django самый простой способ проверять задачи, используя команды непосредственно из вашего терминала в виртуальной среде или используя полный путь к сельдерею:
Док: http://docs.celeryproject.org/en/latest/userguide/workers.html?highlight=revoke#inspecting-workers
$ celery inspect reserved
$ celery inspect inspect
$ celery inspect registered
$ celery inspect scheduled
Также, если вы используете Celery + RabbitMQ, вы можете проверить список очередей, используя следующую команду:
Дополнительная информация: https://linux.die.net/man/1/rabbitmqctl
$ sudo rabbitmqctl list_queues
Ответ 7
Модуль проверки сельдерея, по-видимому, знает только о задачах с точки зрения рабочих. Если вы хотите просмотреть сообщения, которые находятся в очереди (пока их не натягивают рабочие), я предлагаю использовать pyrabbit, что может взаимодействовать с http-абитуриентом rabbitmq для извлечения всех видов информации из очереди.
Пример можно найти здесь: Получить длину очереди с помощью Celery (RabbitMQ, Django)
Ответ 8
Я думаю, что единственный способ получить ожидаемые задачи - сохранить список запущенных задач и позволить удаленной работе из списка при запуске.
С rabbitmqctl и list_queues вы можете получить общее представление о том, сколько задач ждут, но не сами задачи: http://www.rabbitmq.com/man/rabbitmqctl.1.man.html
Если то, что вы хотите, включает обрабатываемую задачу, но еще не закончено, вы можете сохранить список своих задач и проверить их состояния:
from tasks import add
result = add.delay(4, 4)
result.ready() # True if finished
Или вы позволяете Celery сохранять результаты с помощью CELERY_RESULT_BACKEND и проверять, какие из ваших задач там нет.
Ответ 9
Насколько я знаю, Celery не предоставляет API для проверки задач, ожидающих в очереди. Это зависит от брокера. Если вы используете Redis в качестве посредника для примера, то изучение задач, ожидающих в очереди celery (по умолчанию), так же просто, как:
- подключиться к базе данных брокера
- список элементов в списке
celery(например, команда LRANGE)
Имейте в виду, что это задачи, ожидающие выбора доступных рабочих. В вашем кластере могут быть запущены некоторые задачи - их не будет в этом списке, поскольку они уже выбраны.
Ответ 10
Я пришел к выводу, что лучший способ получить количество заданий в очереди - использовать rabbitmqctl, как было предложено несколько раз здесь. Чтобы разрешить любому выбранному пользователю выполнить команду с помощью sudo, я выполнил инструкции здесь (я пропустил редактирование части профиля, t ум ввод в sudo перед командой.)
Я также захватил jamesc grep и cut фрагмент и завернул его в вызовы подпроцесса.
from subprocess import Popen, PIPE
p1 = Popen(["sudo", "rabbitmqctl", "list_queues", "-p", "[name of your virtula host"], stdout=PIPE)
p2 = Popen(["grep", "-e", "^celery\s"], stdin=p1.stdout, stdout=PIPE)
p3 = Popen(["cut", "-f2"], stdin=p2.stdout, stdout=PIPE)
p1.stdout.close()
p2.stdout.close()
print("number of jobs on queue: %i" % int(p3.communicate()[0]))
Ответ 11
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
Ответ 12
Если вы управляете кодом задач, то вы можете обойти проблему, позволяя задаче тривиально повторить попытку при первом запуске, а затем проверяя inspect().reserved(). Повторная попытка регистрирует задачу с помощью обработчика результатов, и сельдерей может это увидеть. Задача должна принять self или context качестве первого параметра, чтобы мы могли получить доступ к количеству повторов.
@task(bind=True)
def mytask(self):
if self.request.retries == 0:
raise self.retry(exc=MyTrivialError(), countdown=1)
...
Это решение не зависит от брокера, т.е. вам не нужно беспокоиться о том, используете ли вы RabbitMQ или Redis для хранения задач.
РЕДАКТИРОВАТЬ: после тестирования я обнаружил, что это только частичное решение. Размер зарезервированного ограничен настройкой предварительной выборки для работника.
Ответ 13
запуск цветка - просмотрщик селери заданий
celery -A app.celery flower
а затем открыть в браузере
localhost:5555
Ответ 14
С subprocess.run:
import subprocess
import re
active_process_txt = subprocess.run(['celery', '-A', 'my_proj', 'inspect', 'active'],
stdout=subprocess.PIPE).stdout.decode('utf-8')
return len(re.findall(r'worker_pid', active_process_txt))
Будьте осторожны, чтобы изменить my_proj на your_proj
Ответ 15
Это работает для меня в моем приложении:
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 за то, что он указал мне правильное направление своим ответом здесь: fooobar.com/info/67479/...