Usuwanie wszystkich oczekujących zadań w selery / rabbitmq

Dla selera 3.0+:

$ celery purge

Aby wyczyścić określoną kolejkę:

$ celery -Q queue_name purge

Dla Selera 2.x i 3.x:

Podczas używania workera z parametrem-Q do definiowania kolejek, na przykład

celery worker -Q queue1,queue2,queue3

Wtedy celery purge nie zadziała, ponieważ nie można przekazać do niej param kolejki. Usunie tylko domyślną kolejkę. Rozwiązaniem jest uruchomienie pracowników z --purge parametrem takim jak:

celery worker -Q queue1,queue2,queue3 --purge

To jednak uruchomi pracownika.

Inną opcją jest użycie subkomandu amqp selera

celery amqp queue.delete queue1
celery amqp queue.delete queue2
celery amqp queue.delete queue3

W Selerze 3+:

CLI:

$ celery -A proj purge

Programowo:

>>> from proj.celery import app
>>> app.control.purge()

Http://docs.celeryproject.org/en/latest/faq.html#how-do-i-purge-all-waiting-tasks

Odkryłem, że celery purge nie działa dla mojej bardziej złożonej konfiguracji selera. Używam wielu nazwanych kolejek do różnych celów:

$ sudo rabbitmqctl list_queues -p celery name messages consumers
Listing queues ...  # Output sorted, whitespaced for readability
celery                                          0   2
[email protected]                      0   1
[email protected]                      0   1
apns                                            0   1
[email protected]                        0   1
analytics                                       1   1
[email protected]                   0   1
bcast.361093f1-de68-46c5-adff-d49ea8f164c0      0   1
bcast.a53632b0-c8b8-46d9-bd59-364afe9998c1      0   1
celeryev.c27b070d-b07e-4e37-9dca-dbb45d03fd54   0   1
celeryev.c66a9bed-84bd-40b0-8fe7-4e4d0c002866   0   1
celeryev.b490f71a-be1a-4cd8-ae17-06a713cc2a99   0   1
celeryev.9d023165-ab4a-42cb-86f8-90294b80bd1e   0   1

Pierwsza kolumna to nazwa kolejki, druga to liczba wiadomości oczekujących w kolejce, a trzecia to liczba słuchaczy dla tej kolejki. Kolejki:

• seler-Kolejka standardowych, idempotentnych zadań selera
[[9]}apns-Kolejka Zadań usługi Apple Push Notification Service, nie do końca jako idempotent
• analityka - Kolejka do długich analiz nocnych
• *.pidbox-kolejkowanie poleceń workera, takich jak shutdown i reset, po jednym na pracownika (2 pracowników selerowych, jeden pracownik apns, jeden pracownik analityczny)
• bcast.* - Broadcast queues, do wysyłania wiadomości do wszystkich pracowników słuchających kolejki (a nie tylko pierwszych, którzy ją przechwycili)
• seler.* - Seler event queues, for reporting task analytics

Zadanie analytics to zadanie brute force, które działało świetnie na małych zestawach danych, ale teraz trwa więcej niż 24 godziny, aby przetworzyć. Czasami coś pójdzie nie tak i utknie w bazie danych. Musi być napisane od nowa, ale do tego czasu, kiedy się zacina, zabijam zadanie, opróżniam kolejkę i próbuję ponownie. Wykrywam "stuckness"patrząc na liczbę wiadomości dla kolejki analytics, która powinna wynosić 0 (zakończona analityka) lub 1 (oczekiwanie na wczorajszą analizę, aby zakończyć). 2 lub wyżej jest źle, i dostaję e-mail.

celery purge oferty kasowania zadań z jedna z kolejek transmisji i nie widzę opcji wyboru innej kolejki nazwanej.

Oto mój proces:

$ sudo /etc/init.d/celeryd stop  # Wait for analytics task to be last one, Ctrl-C
$ ps -ef | grep analytics  # Get the PID of the worker, not the root PID reported by celery
$ sudo kill <PID>
$ sudo /etc/init.d/celeryd stop  # Confim dead
$ python manage.py celery amqp queue.purge analytics
$ sudo rabbitmqctl list_queues -p celery name messages consumers  # Confirm messages is 0
$ sudo /etc/init.d/celeryd start

W Selerze 3+

Http://docs.celeryproject.org/en/3.1/faq.html#how-do-i-purge-all-waiting-tasks

CLI

Nazwa kolejki:

 celery -A proj amqp queue.purge <queue name>

Wyczyść skonfigurowaną kolejkę

celery -A proj purge

Wyczyściłem wiadomości, ale nadal są wiadomości w kolejce? Odpowiedź: zadania są potwierdzane (usuwane z kolejki) zaraz po ich wykonaniu. Po tym, jak pracownik otrzyma zadanie, zajmie trochę czasu, zanim zostanie faktycznie wykonywane, zwłaszcza jeśli Na wykonanie czeka już wiele zadań. Komunikaty, które nie są potwierdzone, są przechowywane przez worker, dopóki nie zamknie połączenia z brokerem (serwerem AMQP). Gdy to połączenie zostanie zamknięte (np. ponieważ worker został zatrzymany), zadania zostaną ponownie wysłane przez brokera do następnego dostępnego workera (lub tego samego workera, gdy został ponownie uruchomiony), więc aby poprawnie wyczyścić kolejkę oczekujących zadań, musisz zatrzymać wszystkich workerów, a następnie wyczyścić zadania przy użyciu selera.Kontrola.purge ().

Aby wyczyścić całą kolejkę należy zatrzymać pracowników.

1. Aby prawidłowo wyczyścić kolejkę oczekujących zadań, musisz zatrzymać wszystkich pracowników (http://celery.readthedocs.io/en/latest/faq.html#i-ve-purged-messages-but-there-are-still-messages-left-in-the-queue):

$ sudo rabbitmqctl stop

Lub (w przypadku gdy RabbitMQ / message broker jest zarządzany przez Supervisora):

$ sudo supervisorctl stop all

2. ...a następnie wyczyść zadania z określonej kolejki:

$ cd <source_dir>
$ celery amqp queue.purge <queue name>

3. Start RabbitMQ:

$ sudo rabbitmqctl start

Lub (w przypadku, gdy RabbitMQ jest zarządzany przez przełożonego):

$ sudo supervisorctl start all