django - celery 和 transaction.atomic

Question

在某些 Django View 中，我使用这样的模式来保存对模型的更改，然后根据新的模型数据进行一些异步更新(例如生成图像，进一步更改模型)。 mytask 是一个 celery 任务:

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

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

编辑:实际上以这种方式注册 on_commit celery 任务更好(没有 lambda):

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