Suppression de toutes les tâches en attente dans celery / rabbitmq

Question

Comment puis-je supprimer toutes les tâches en suspens sans connaître le nom de l'entreprise ? task_id pour chaque tâche ?

Answer 1

De la docs :

$ celery -A proj purge

ou

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

(EDIT : Mis à jour avec la méthode actuelle).

Answer 2

Pour celery 3.0+ :

$ celery purge

Pour purger une file d'attente spécifique :

$ celery -Q queue_name purge

Answer 3

Pour Celery 2.x et 3.x :

Lorsque vous utilisez worker avec le paramètre -Q pour définir des files d'attente, par exemple

celery worker -Q queue1,queue2,queue3

puis celery purge ne fonctionnera pas, car vous ne pouvez pas lui passer les paramètres de la file d'attente. Elle supprimera uniquement la file d'attente par défaut. La solution est de démarrer vos travailleurs avec --purge comme ceci :

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

Cependant, cela fera tourner le travailleur.

Une autre option consiste à utiliser la sous-commande amqp de celery

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

Answer 4

Dans Celery 3+ :

CLI :

$ celery -A proj purge

De manière programmée :

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

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

Answer 5

J'ai trouvé que celery purge ne fonctionne pas pour ma configuration celery plus complexe. J'utilise plusieurs files d'attente nommées à des fins différentes :

$ sudo rabbitmqctl list_queues -p celery name messages consumers
Listing queues ...  # Output sorted, whitespaced for readability
celery                                          0   2
celery@web01.celery.pidbox                      0   1
celery@web02.celery.pidbox                      0   1
apns                                            0   1
apns@web01.celery.pidbox                        0   1
analytics                                       1   1
analytics@web01.celery.pidbox                   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

La première colonne est le nom de la file d'attente, la deuxième est le nombre de messages en attente dans la file, et la troisième est le nombre d'auditeurs pour cette file. Les files d'attente sont :

• celery - File d'attente pour les tâches standard et idempotentes de celery
• apns - File d'attente pour les tâches du service de notification Push d'Apple, pas tout à fait aussi idempotent
• analytics - File d'attente pour les analyses nocturnes de longue durée
• *.pidbox - File d'attente pour les commandes des travailleurs, telles que shutdown et reset, une par travailleur (2 travailleurs celery, un travailleur apns, un travailleur analytics)
• bcast.* - File d'attente de diffusion, pour envoyer des messages à tous les travailleurs qui écoutent une file d'attente (plutôt qu'au premier à l'attraper).
• celeryev.* - Files d'attente d'événements Celery, pour l'analyse des tâches de reporting

La tâche analytique est une tâche de force brute qui fonctionnait très bien sur de petits ensembles de données, mais qui prend maintenant plus de 24 heures à traiter. De temps en temps, quelque chose se passe mal et la tâche reste bloquée en attendant la base de données. Elle doit être réécrite, mais en attendant, quand elle est bloquée, je tue la tâche, vide la file d'attente et réessaie. Je détecte le "blocage" en regardant le nombre de messages de la file d'attente des analyses, qui devrait être de 0 (analyses terminées) ou de 1 (attente de la fin des analyses de la nuit dernière). 2 ou plus est mauvais, et je reçois un email.

celery purge propose d'effacer les tâches d'une des files d'attente de diffusion, et je ne vois pas d'option pour choisir une autre file d'attente nommée.

Voici mon processus :

$ 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