最近在Django编写的一个系统中希望能加入对当前celery积压任务的展示,并且能提供功能来清理队列。我们使用了RabbitMQ作为了celery的broker,除了直接用kombu来编码实现需求外,还可以直接使用以下的方法来实现需要的功能,减少一些broker配置的代码。
可使用以下代码来获取指定队列的积压任务数,请将代码中的”your_app_name”和”your_queue_name”替换为实际环境中的参数:
1 | from celery import Celery |
注意,如果指定队列不存在,则会报队列不存在的错误。
如需清空队列,则只需将上面的代码进行下修改,将queue_declare修改为queue_purge,修改后的代码如下:
1 | app.connection().channel().queue_purge(your_queue_name) |