Celery 和 transaction.atomic

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

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

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

之间是否存在某些不兼容之处transaction.atomic还有芹菜？在 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作为队列。

As @dotz 提到 https://stackoverflow.com/a/31030040/2263517，生成异步任务并立即阻塞并继续等待直到完成几乎没有用处。

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

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

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

然而，这样你的任务不会被执行如果数据库事务失败。这通常是所需的行为，但请记住这一点。

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

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