运行芹菜任务时如何取消确认消息？

Question

现在我有一些同步作业，它们是有状态的，所以如果任务失败，我必须取消确认消息，然后让它们转到RabbitMQ 的前面。但是当我尝试引发错误时，我发现 celery 仍然会确认此消息，并且队列已被清除。

@celery.task(bind=True)
def my_task(self, *args, **kwargs):
    raise ValueError

我发现 celery 任务有一个名为 retry 的方法，但它会将任务添加到队列的后面。这不是我想要的。

@celery.task(bind=True)
def my_task(self, *args, **kwargs):
    try:
        raise ValueError
    except Exception:
        self.retry(countdown=15)

即使我不能用终止信号来做到这一点：

os.kill(os.getpid(), signal.SIGKILL)

我该怎么办？ celery 是否提供了一些错误，以便我可以引发此错误以通知 celery 不确认我的消息？

Answer 1

根据文档 celery 能够RabbitMQs priority queues。 https://docs.celeryproject.org/en/latest/faq.html#does-celery-support-task-priorities

因此，您应该能够通过将重试任务的优先级高于常规任务来将它们推到队列的前面。

@celery.task(bind=True)
def my_task(self, *args, **kwargs):
    try:
        raise ValueError
    except Exception:
        self.retry(countdown=15, priority=9)

或

根据这个github issue，您还可以将重试的任务分配给一个新的专用队列，并分配您的资源来对该队列进行优先级排序。

@celery.task(bind=True)
def my_task(self, *args, **kwargs):
    try:
        raise ValueError
    except Exception:
        self.retry(countdown=15, queue='prioritized_queue_name') 

Answer 2

在文档https://docs.celeryproject.org/en/stable/userguide/configuration.html 上我发现：

task_acks_on_failure_or_timeout 默认为enabled。

---

所以我认为你应该尝试结合

task_acks_late=True + task_acks_on_failure_or_timeout=False
实现NO acknowledgement when a task fails。