Rufen Sie eine Liste der Aufgaben in einer Warteschlange in Sellerie ab

Wie kann ich eine Liste von Aufgaben in einer Warteschlange abrufen, die noch verarbeitet werden müssen?

BEARBEITEN: Weitere Antworten zum Abrufen einer Liste der Aufgaben in der Warteschlange finden Sie unter Andere Antworten.

Sie sollten hier nachsehen: Sellerie-Leitfaden - Inspektion von Arbeitern

Grundsätzlich ist dies:

from celery.app.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()

Je nachdem was du willst

Wenn Sie rabbitMQ verwenden, verwenden Sie dies im Terminal:

sudo rabbitmqctl list_queues

Es wird eine Liste der Warteschlangen mit der Anzahl der ausstehenden Aufgaben gedruckt. beispielsweise:

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

Die Nummer in der rechten Spalte gibt die Anzahl der Aufgaben in der Warteschlange an. oben hat die Sellerie-Warteschlange 166 ausstehende Aufgaben.

Wenn Sie keine priorisierten Aufgaben verwenden, ist dies ziemlich einfach, wenn Sie Redis verwenden. So erhalten Sie die Anzahl der Aufgaben:

redis-cli -h HOST -p PORT -n DATABASE_NUMBER llen QUEUE_NAME

Priorisierte Aufgaben verwenden jedoch einen anderen Schlüssel in Redis , sodass das Gesamtbild etwas komplizierter ist. Das vollständige Bild ist, dass Sie Redis für jede Priorität der Aufgabe abfragen müssen. In Python (und aus dem Flower-Projekt) sieht dies folgendermaßen aus:

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's 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])

Wenn Sie eine tatsächliche Aufgabe erhalten möchten, können Sie Folgendes verwenden:

redis-cli -h HOST -p PORT -n DATABASE_NUMBER lrange QUEUE_NAME 0 -1

Von dort aus müssen Sie die zurückgegebene Liste deserialisieren. In meinem Fall konnte ich dies mit etwas erreichen wie:

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

Seien Sie nur gewarnt, dass die Deserialisierung einen Moment dauern kann und Sie die obigen Befehle anpassen müssen, um mit verschiedenen Prioritäten zu arbeiten.

Verwenden Sie diese Option, um Aufgaben aus dem Backend abzurufen

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)

Wenn Sie Sellerie + Django verwenden , können Sie Aufgaben am einfachsten mithilfe von Befehlen direkt von Ihrem Terminal in Ihrer virtuellen Umgebung oder mithilfe eines vollständigen Pfads zum Sellerie überprüfen :

Doc : 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

Auch wenn Sie Sellerie + RabbitMQ verwenden , können Sie die Liste der Warteschlangen mit dem folgenden Befehl überprüfen :

Weitere Informationen : https://linux.die.net/man/1/rabbitmqctl

$ sudo rabbitmqctl list_queues

Eine Copy-Paste-Lösung für Redis mit JSON-Serialisierung:

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

Es funktioniert mit Django. Vergiss nur nicht, dich zu ändern yourproject.celery.

Das Sellerie-Inspektionsmodul scheint die Aufgaben nur aus Sicht der Arbeitnehmer zu kennen. Wenn Sie die Nachrichten anzeigen möchten, die sich in der Warteschlange befinden (noch von den Mitarbeitern abgerufen werden müssen ), empfehle ich die Verwendung von Pyrabbit , das mit der rabbitmq http-API verbunden werden kann, um alle Arten von Informationen aus der Warteschlange abzurufen.

Ein Beispiel finden Sie hier: Abrufen der Warteschlangenlänge mit Sellerie (RabbitMQ, Django)

Ich denke, die einzige Möglichkeit, die wartenden Aufgaben zu erhalten, besteht darin, eine Liste der von Ihnen gestarteten Aufgaben zu führen und die Aufgabe beim Starten von der Liste entfernen zu lassen.

Mit rabbitmqctl und list_queues erhalten Sie einen Überblick darüber, wie viele Aufgaben warten, nicht jedoch die Aufgaben selbst: http://www.rabbitmq.com/man/rabbitmqctl.1.man.html

Wenn die gewünschte Aufgabe enthalten ist, aber noch nicht abgeschlossen ist, können Sie eine Liste Ihrer Aufgaben führen und deren Status überprüfen:

from tasks import add
result = add.delay(4, 4)

result.ready() # True if finished

Oder Sie lassen Sellerie die Ergebnisse mit CELERY_RESULT_BACKEND speichern und überprüfen, welche Ihrer Aufgaben dort nicht enthalten sind.

Dies hat bei mir in meiner Bewerbung funktioniert:

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 wird eine Liste von Zeichenfolgen sein, die Aufgaben in der Warteschlange entsprechen.

Vergessen Sie nicht, CELERY_APP_INSTANCE gegen Ihr eigenes auszutauschen.

Vielen Dank an @ashish, der mich mit seiner Antwort hier in die richtige Richtung gelenkt hat: https://stackoverflow.com/a/19465670/9843399

Soweit ich weiß, gibt Celery keine API zum Untersuchen von Aufgaben an, die in der Warteschlange warten. Dies ist maklerspezifisch. Wenn Sie Redis als Broker für ein Beispiel verwenden, ist das Untersuchen von Aufgaben, die in der celery(Standard-) Warteschlange warten, so einfach wie:

1. Stellen Sie eine Verbindung zur Broker-Datenbank her
2. Listenelemente in der celeryListe (LRANGE-Befehl für ein Beispiel)

Denken Sie daran, dass dies Aufgaben sind, die darauf warten, von verfügbaren Mitarbeitern ausgewählt zu werden. In Ihrem Cluster werden möglicherweise einige Aufgaben ausgeführt. Diese werden nicht in dieser Liste aufgeführt, da sie bereits ausgewählt wurden.

Ich bin zu dem Schluss gekommen, dass der beste Weg, um die Anzahl der Jobs in einer Warteschlange zu ermitteln, die Verwendung ist, rabbitmqctlwie hier mehrmals vorgeschlagen wurde. Damit jeder ausgewählte Benutzer den Befehl mit ausführen kann, habe sudoich die Anweisungen hier befolgt (ich habe die Bearbeitung des Profilteils übersprungen, da es mir nichts ausmacht, vor dem Befehl sudo einzugeben.)

Ich schnappte mir auch Jamesc's grepund cutSnippet und wickelte es in Subprozessaufrufe ein.

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]))

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

Wenn Sie den Code der Aufgaben steuern, können Sie das Problem umgehen, indem Sie eine Aufgabe bei der ersten Ausführung einen trivialen Wiederholungsversuch auslösen lassen und dann überprüfen inspect().reserved(). Der Wiederholungsversuch registriert die Aufgabe mit dem Ergebnis-Backend, und Sellerie kann das sehen. Die Aufgabe muss selfoder contextals erster Parameter akzeptieren, damit wir auf die Anzahl der Wiederholungen zugreifen können.

@task(bind=True)
def mytask(self):
    if self.request.retries == 0:
        raise self.retry(exc=MyTrivialError(), countdown=1)
    ...

Diese Lösung ist maklerunabhängig, dh. Sie müssen sich keine Gedanken darüber machen, ob Sie RabbitMQ oder Redis zum Speichern der Aufgaben verwenden.

EDIT: Nach dem Testen habe ich festgestellt, dass dies nur eine Teillösung ist. Die Größe von reserviert ist auf die Prefetch-Einstellung für den Worker beschränkt.

Mit 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))

Achten Sie darauf, my_projmit zu ändernyour_proj