如果你想删除所有挂起的任务，以及活动和保留的任务，以完全停止Celery，这对我来说是有效的:

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)

芹菜2。X和3.x:

例如，当使用带-Q参数的worker来定义队列时

celery worker -Q queue1,queue2,queue3

然后芹菜清除将不会工作，因为您不能传递队列参数给它。它只会删除默认队列。 解决方案是用——purge参数启动你的worker，就像这样:

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

然而，这将运行worker。

另一个选择是使用芹菜的amqp子命令

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

从文档中可以看出:

$ celery -A proj purge

or

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

(编辑:更新与当前的方法。)

对于芹菜5.0+版本，使用RabbitMQ作为代理

我们需要先建立一个从程序到经纪人的新连接， 并将连接与要清除的队列绑定。

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

芹菜3+:

CLI:

$ celery -A proj purge

编程:

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

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

芹菜3+

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

CLI

清除命名队列:

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

清除配置的队列

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().

因此，要清除整个队列，必须停止worker。