252

How can I delete all pending tasks without knowing the task_id for each task?

12 Answers 12

409

From the docs:

$ celery -A proj purge

or

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

(EDIT: Updated with current method.)

7
  • 64
    Or, from Django, for celery 3.0+: manage.py celery purge (celeryctl is now deprecated and will be gone in 3.1). Commented Apr 19, 2013 at 14:26
  • 9
    I found this answer looking for how to do this with a redis backend. Best method I found was redis-cli KEYS "celery*" | xargs redis-cli DEL which worked for me. This will wipe out all tasks stored on the redis backend you're using.
    – Melignus
    Commented Aug 28, 2013 at 17:22
  • 1
    How can i do this in celery 3.0 ?
    – luistm
    Commented Nov 8, 2013 at 15:25
  • 5
    For me, it was simply celery purge (inside the relevant virtual env). Ooops - there's an answer with the same below..... stackoverflow.com/a/20404976/1213425
    – Erve1879
    Commented Jun 30, 2014 at 5:54
  • For Celery 4.0+ in combination with Django it's again this command, where the argument to -A is the Django app where the celery.py is located.
    – gitaarik
    Commented Mar 23, 2017 at 12:46
154

For celery 3.0+:

$ celery purge

To purge a specific queue:

$ celery -Q queue_name purge
2
  • 12
    If you get connection errors, make sure you specify the app, e.g. celery -A proj purge. Commented Apr 3, 2016 at 17:22
  • 4
    I believe the -Q flag has been deprecated (didn't work for me, "no such option"), to delete a specific queue on Celery 5.0.5 you'd run celery -A appname purge --queues queuename
    – Thorvald
    Commented Jan 2, 2022 at 12:16
35

For Celery 2.x and 3.x:

When using worker with -Q parameter to define queues, for example

celery worker -Q queue1,queue2,queue3

then celery purge will not work, because you cannot pass the queue params to it. It will only delete the default queue. The solution is to start your workers with --purge parameter like this:

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

This will however run the worker.

Other option is to use the amqp subcommand of celery

celery amqp queue.delete queue1
celery amqp queue.delete queue2
celery amqp queue.delete queue3
1
  • Yes, this is for older (2.x and maybe 3.x) versions of celery. I cannot edit the answer
    – smido
    Commented Jul 11, 2017 at 9:53
19

In Celery 3+:

CLI:

$ celery -A proj purge

Programatically:

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

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

10

I found that celery purge doesn't work for my more complex celery config. I use multiple named queues for different purposes:

$ 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

The first column is the queue name, the second is the number of messages waiting in the queue, and the third is the number of listeners for that queue. The queues are:

  • celery - Queue for standard, idempotent celery tasks
  • apns - Queue for Apple Push Notification Service tasks, not quite as idempotent
  • analytics - Queue for long running nightly analytics
  • *.pidbox - Queue for worker commands, such as shutdown and reset, one per worker (2 celery workers, one apns worker, one analytics worker)
  • bcast.* - Broadcast queues, for sending messages to all workers listening to a queue (rather than just the first to grab it)
  • celeryev.* - Celery event queues, for reporting task analytics

The analytics task is a brute force tasks that worked great on small data sets, but now takes more than 24 hours to process. Occasionally, something will go wrong and it will get stuck waiting on the database. It needs to be re-written, but until then, when it gets stuck I kill the task, empty the queue, and try again. I detect "stuckness" by looking at the message count for the analytics queue, which should be 0 (finished analytics) or 1 (waiting for last night's analytics to finish). 2 or higher is bad, and I get an email.

celery purge offers to erase tasks from one of the broadcast queues, and I don't see an option to pick a different named queue.

Here's my process:

$ 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
4
  • Not an answer though, is it? Very informative however! Commented Jan 28, 2015 at 17:17
  • 5
    celeryctl purge didn't work with named queues. python manage.py celery amqp queue.purge <queue_name> did. I think the context is useful for those with complex setups, so they can figure out what they need to do if celeryctl purge fails for them.
    – jwhitlock
    Commented Jan 29, 2015 at 4:08
  • I cannot find manage.py in my Celery 3.1.17, has the file been removed or just spanking new? I found what looks like the corresponding interface (queue.purge) in */bin/amqp.py, however. But after trying to correlate the contents of the file with the documentation, I must regrettably admit that Celery is woefully undocumented and also a very convoluted piece of work, at least judging it by its source code. Commented Jan 29, 2015 at 20:49
  • manage.py is the Django management script, and manage.py celery runs celery after loading configuration from Django settings. I haven't used celery outside of Django, but the included celery command may be what you are looking for: celery.readthedocs.org/en/latest/userguide/monitoring.html
    – jwhitlock
    Commented Jan 29, 2015 at 21:24
10

If you want to remove all pending tasks and also the active and reserved ones to completely stop Celery, this is what worked for me:

from proj.celery import app
from celery.task.control import inspect, revoke

# remove pending tasks
app.control.purge()

# remove active tasks
i = inspect()
jobs = i.active()
for hostname in jobs:
    tasks = jobs[hostname]
    for task in tasks:
        revoke(task['id'], terminate=True)

# remove reserved tasks
jobs = i.reserved()
for hostname in jobs:
    tasks = jobs[hostname]
    for task in tasks:
        revoke(task['id'], terminate=True)
1
  • If you also want to revoke scheduled tasks, ie, those waiting due to an eta or countdown, you also need to revoke tasks in the i.scheduled() queue. For these, the id is inside a request key (at least for me on Redis), ie, you need to run revoke(task['request']['id']). Also, for me on Celery 5.2.7 in Django at least, I needed to run app.control.inspect, and app.control.revoke - I couldn't import them independently (got an unbound error). My final code is here.
    – Chris
    Commented Jul 27, 2022 at 9:43
8

In Celery 3+

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

CLI

Purge named queue:

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

Purge configured queue

celery -A proj purge

I’ve purged messages, but there are still messages left in the queue? Answer: Tasks are acknowledged (removed from the queue) as soon as they are actually executed. After the worker has received a task, it will take some time until it is actually executed, especially if there are a lot of tasks already waiting for execution. Messages that are not acknowledged are held on to by the worker until it closes the connection to the broker (AMQP server). When that connection is closed (e.g. because the worker was stopped) the tasks will be re-sent by the broker to the next available worker (or the same worker when it has been restarted), so to properly purge the queue of waiting tasks you have to stop all the workers, and then purge the tasks using celery.control.purge().

So to purge the entire queue workers must be stopped.

5

For Celery 5.0+, to do it from the CLI and to target a specific queue:

celery -A APP_NAME purge --queues QUEUE_NAME

Add -f option to the end to skip the confirmation step if you're trying to do it in on step like I was.

4

celery 4+ celery purge command to purge all configured task queues

celery -A *APPNAME* purge

programmatically:

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

all pending task will be purged. Reference: celerydoc

3

For Celery Version 5.0+ with RabbitMQ as broker

We need establish a new connection from program to broker first, and bind the connection with the queues to purge.

# proj/celery.py
from celery import Celery
app = Celery('proj')
from proj.celery import app
queues = ['queue_A', 'queue_B', 'queue_C']
with app.connection_for_write() as conn:
    conn.connect()
    for queue in queues:
        count = app.amqp.queues[queue].bind(conn).purge()
        print(f'Purge {queue} with {count} message(s)')
2

1. To properly purge the queue of waiting tasks you have to stop all the workers (http://celery.readthedocs.io/en/latest/faq.html#i-ve-purged-messages-but-there-are-still-messages-left-in-the-queue):

$ sudo rabbitmqctl stop

or (in case RabbitMQ/message broker is managed by Supervisor):

$ sudo supervisorctl stop all

2. ...and then purge the tasks from a specific queue:

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

3. Start RabbitMQ:

$ sudo rabbitmqctl start

or (in case RabbitMQ is managed by Supervisor):

$ sudo supervisorctl start all
0

I think that may have solved it. the old tasks were still in django-celery-beat>periodic tasks in my django admin, so go to django admin page and delete all of them, then if you are in docker container, restart the it all the problem solved

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.