问题描述

我目前的 celery 设置有大约 10,000 个计划任务.我没有意识到计划任务是什么，并决定提前几个月使用它们发送后续电子邮件.

I have around 10,000 scheduled tasks on my current celery setup. I didn't realize what scheduled tasks were and decided to use them to send follow-up emails months in advance.

回想起来，将任务安排在未来超过 1 小时可能永远不是一个好主意，因为每次重新启动工作器时，它都必须重新接收来自 rabbitMQ 的每个计划任务，然后它们都只是坐在记忆.

Looking back, it's probably never a good idea to schedule a task for more than 1 hour in the future as every time you restart a worker it has to re-receive every scheduled task from rabbitMQ and then they all just sit in the memory.

我的问题是，如果我必须撤销一项任务，它不仅仅是删除它.任务保留在内存中，但撤销队列现在包含任务的 ID.当它准备执行时，celery 会检查它是否被撤销，如果是，它将在此时撤销它.

My problem is that if I have to revoke a task, it doesn't just delete it. The task stays in memory but a revoke queue now contains the ID of the task. When it is up for execution, celery checks to see if it is revoked and if it is, it will revoke it at this point.

但是，直到那时，该任务仍将保留在内存中，如果我在任何时候重新启动我的工作人员，撤销队列将被清除，因为我没有使其持久化.

However, the task will still stay in memory up until then, and if I restart my worker at anytime, the revoke queue will be cleared as I didn't make it persistent.

如何从我的 celery worker 中永久删除任务?我基本上只需要向rabbitMQ发回一个确认消息，这样rabbit就会一劳永逸地删除它，如果我重新启动celery，它就不会回来了.

我查看了文档和源代码，并尝试自己在 shell 中执行此操作，但我无法确定将任务提交给 rabbitMQ 然后永远弹出它的正确位置.

I've looked in the docs and source code and tried to do it myself in the shell but I can't figure out the proper place for acking a task to rabbitMQ and then popping it forever.

推荐答案

如果你只使用一个队列或一个任务，这很容易:

来自 文档:

回答:可以使用 celery purge 命令清除所有已配置的任务队列:

Answer: You can use the celery purge command to purge all configured task queues:

$ celery -A proj purge

或以编程方式:

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

如果您只想从特定队列中清除消息，则必须使用 AMQP API 或 celery amqp 实用程序:

If you only want to purge messages from a specific queue you have to use the AMQP API or the celery amqp utility:

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

数字 1753 是删除的消息数.

The number 1753 is the number of messages deleted.

您也可以使用 --purge 参数启动 worker，以在 worker 启动时清除消息.

You can also start worker with the --purge argument, to purge messages when the worker starts.

更新: 如果您有多个队列或任务

我不知道有什么方法可以在 RabbitMQ 中编辑它们，因为服务器不是为访问/编辑/删除排队任务而构建的，但您始终可以在代码中禁用您的任务:

I don't know of any way to edit them in RabbitMQ since the server is not built to access/edit/delete queued tasks that way, but you could always disable your task in the code:

@task
def my_old_task()
   pass

这样所有任务都会按计划运行，但不会执行任何操作；因为它们既没有被重命名也没有被删除，所以你不会有任何错误.

This way all the tasks will run as scheduled but won't perform anything; since they are neither renamed nor deleted, you won't have any error.

显然，您应该更新代码以停止安排这些任务.一段时间后，将不再安排此类任务，因此您可以删除代码.

Obviously you should update your code to stop scheduling these tasks. After a while, there won't be anymore tasks of this type scheduled so you can delete the code.