Celery 和 transaction.atomic

Question

在一些 Django 视图中，我使用这样的模式来保存对模型的更改，然后根据新的模型数据进行一些异步更新（例如生成图像，进一步更改模型）。 mytask 是一个芹菜任务：

with transaction.atomic():
    mymodel.save()
    mytask.delay(mymodel.id).get()

问题是任务永远不会返回。查看 celery 的日志，任务排队（我在日志中看到“已接收任务”），但它从未完成。如果我将mytask.delay...get 调用移出事务，它会成功完成。

transaction.atomic 和 celery 之间是否存在一些不兼容？在 Django 1.6 或 1.7 中，我是否可以在一个事务下同时进行常规模型更新和来自单独任务流程的更新？

我的数据库是 postgresql 9.1。我正在使用 celery==3.1.16 / django-celery 3.1.16，amqp==1.4.6，Django==1.6.7，kombu==3.0.23。 broker后端是amqp，队列是rabitmq。

Answer 1

作为@dotz mentioned，生成异步任务并立即阻塞并继续等待直到完成几乎没有用处。

此外，如果您以这种方式附加到它（最后的.get()），您可以确定您的工作人员不会看到刚刚进行的mymodel 实例更改，因为它们不会被提交然而 - 记住你还在 atomic 块内。

您可以做的（从 Django 1.9 开始）是使用 django.db.transaction.on_commit 钩子将任务延迟到提交当前活动事务之后：

from django.db import transaction

with transaction.atomic():
    mymodel.save()
    transaction.on_commit(lambda:
        mytask.delay(mymodel.id))

我经常在我的post_save 信号处理程序中使用这种模式，这些处理程序会触发对新模型实例的一些处理。例如：

from django.db import transaction
from django.db.models.signals import post_save
from django.dispatch import receiver
from . import models   # Your models defining some Order model
from . import tasks   # Your tasks defining a routine to process new instances

@receiver(post_save, sender=models.Order)
def new_order_callback(sender, instance, created, **kwargs):
    """ Automatically triggers processing of a new Order. """
    if created:
        transaction.on_commit(lambda:
            tasks.process_new_order.delay(instance.pk))

但是，如果数据库事务失败，您的任务将不会被执行。这通常是期望的行为，但请记住这一点。

编辑：以这种方式注册 on_commit celery 任务实际上更好（没有 lambda）：

transaction.on_commit(tasks.process_new_order.s(instance.pk).delay)

Answer 2

“单独的任务”= 由工人运行的东西。

"Celery worker" = 另一个进程。

我不知道有任何方法可以让您在 2 个或多个进程之间共享单个数据库事务。您想要的是在该事务中以同步方式运行任务，然后等待结果......但是，如果这是您想要的，为什么还需要任务队列？

Answer 3

Celery 任务中有一些竞争条件。 我认为here 是对您问题的解释。

查看这些文档。还有一些像 django-celery-transactions 这样的包可以帮助你解决你的问题。