Django post_save 信号和 celery 任务之间可能的竞争条件

Question

在 django 2.0 应用程序中，我有一个名为 Document 的模型，用于将图像上传并保存到文件系统。那部分有效。我在 celery (v 4.2.1) 任务中使用 https://github.com/ageitgey/face_recognition 对图像执行面部识别。我将图像的 document_id 传递给 celery 任务，这样人脸识别任务就可以找到要处理的图像。如果我在保存图像后从 DocumentAdmin 操作手动调用 face_recognition 任务，这一切都很好。

我尝试从 models.py 中的 (models.signals.post_save, sender=Document) 方法调用 face_recognition 任务，但我在 face_recognition 的 celery 任务中从这一行收到错误：

document = Document.objects.get(document_id=document_id)

错误是：

[2018-11-26 16:54:28,594: ERROR/ForkPoolWorker-1] Task biometric_identification.tasks.find_faces_task[428ca39b-aefb-4174-9906-ff2146fd6f14] raised unexpected: DoesNotExist('Document matching query does not exist.',)
Traceback (most recent call last):
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/celery/app/trace.py", line 382, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/celery/app/trace.py", line 641, in __protected_call__
    return self.run(*args, **kwargs)
  File "/home/mark/python-projects/memorabilia-JSON/biometric_identification/tasks.py", line 42, in find_faces_task
    document = Document.objects.get(document_id=document_id)
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/django/db/models/manager.py", line 82, in manager_method
    return getattr(self.get_queryset(), name)(*args, **kwargs)
  File "/home/mark/.virtualenvs/memorabilia-JSON/lib/python3.6/site-packages/django/db/models/query.py", line 403, in get
    self.model._meta.object_name
memorabilia.models.DoesNotExist: Document matching query does not exist.

另外，这个错误不是一直都会出现，只是偶尔出现。其余时间，该过程有效；即保存图像并识别人脸。

我在 DocumentAdmin class 中覆盖了 save_model，但这只是为另一个模型中的图像保存了一些元数据。最后一行是对 super().save_model(request, obj, form, change) 的调用，所以我假设 post_save 信号在那之后被调用。

在我看来，将模型保存到数据库与 celery 任务在数据库中查询新创建的模型之间存在竞争条件 document_id。我以为 post_save 信号直到模型被保存才被激活？

我是否必须在 celery 任务中添加一些人为延迟 face_recognition 才能解决这种可能的竞争条件，或者我是否遗漏了其他内容？

谢谢！

马克

Answer 1

似乎有时信号超过了您的数据库写入速度！作为有害的解决方法，您可以做的是运行芹菜任务稍晚一些，只需几秒钟。

这是如何完成的：

your_task.apply_async(
            [document_id],
            countdown=5 # this is the delay in seconds - you can adapt it accordingly
        )

让我知道这是否适合您的情况！

Answer 2

检查保存 Document 模型的函数。它包含在 atomic block somewhere or you have ATOMIC_REQUESTS 设置为 True 中。所以当 post_save 被调用时，事务还没有提交。所以你的模型在那一刻并没有真正保存到数据库中。

Answer 3

如@UnholyRaven 的回答所述，问题与任务执行时未提交的事务有关。

为了解决这个问题，我们可以使用 Django 的 transaction.on_commit

在事务提交上安排任务

@receiver(post_save, sender=Document):
find_faces(sender, instance, created, **kwargs):
  transaction.on_commit(lambda: find_faces_task.apply_async([instance.id]))

Django post_save 信号和 celery 任务之间可能的竞争条件

Possible race condition between Django post_save signal and celery task

django

django-models

celery

celery-task