带有 Celery 的 Django - 未找到现有对象

Posted 2023-02-24

【中文标题】带有 Celery 的 Django - 未找到现有对象

【英文标题】：Django with Celery - existing object not found

【发布时间】：2015-11-20 06:39:51

【问题描述】：

我在从另一个 celery 任务执行 celery 任务时遇到问题。

这是有问题的sn-p（数据对象已经存在于数据库中，它的属性只是在finalize_data函数中更新）：

def finalize_data(data):
    data = update_statistics(data)
    data.save()
    from apps.datas.tasks import optimize_data
    optimize_data.delay(data.pk)

@shared_task
def optimize_data(data_pk):
    data = Data.objects.get(pk=data_pk)
    #Do something with data

在 optimize_data 函数中获取调用失败，并显示“数据匹配查询不存在”。

如果我在 finalize_data 函数中调用通过 pk 函数检索，它工作正常。如果我将 celery 任务调用延迟一段时间，它也可以正常工作。

这一行：

optimize_data.apply_async((data.pk,), countdown=10)

而不是

optimize_data.delay(data.pk)

工作正常。但我不想在我的代码中使用黑客。 .save() 调用是否可能异步阻止对该行/对象的访问？

【问题讨论】：

如果数据对象已经存在，那么你不应该得到那个错误 - 如果有一个块，那么如果有一些错误配置，你可能会看到超时。在没有countdown 的情况下使用apply_async() 有效吗？

我猜您的调用者在 celery 开始处理任务之前尚未提交的事务中。因此 celery 找不到记录。这就是为什么添加倒计时使它起作用的原因。 1秒倒计时有用吗？我在整个代码中使用了 1 秒倒计时来处理这个问题。其他解决方案是停止使用事务。

你使用的是哪个版本的 Django？

@Lee 好电话，调用者在尚未提交的事务中（finalize_data 是用事务原子调用的），我重构了代码，一切正常。留下答案，以便我可以将其标记为已解决。谢谢大家的帮助。

【参考方案1】：

我知道这是一篇旧帖子，但我今天偶然发现了这个问题。 Lee 的回答为我指明了正确的方向，但我认为今天存在更好的解决方案。

使用 Django 提供的 on_commit 处理程序可以解决这个问题，而无需在代码中使用骇人听闻的倒计时方式，这对于用户来说可能无法直观地了解其存在的原因。

我不确定发布问题时是否存在这种情况，但我只是发布答案，以便将来来到这里的人知道替代方案。

【讨论】：

我觉得这样比较合适。但是，我对这种方式感到担忧。基本上，代码顺序是：首先是“object.save()”，然后是第二个“transaction.on_commit(func)”，对吧？是否存在在执行第二个代码之前触发提交信号的情况？在这种情况下，不会调用第二个代码中的“func”。

有效问题。我的回答假设两行代码都包含在事务中。这保证了on_commit 函数将在事务成功提交时执行。

啊哈，刚刚发现其实没问题，“func”仍然会像文档中那样被调用：“如果你在没有活动事务的情况下调用 on_commit()，回调将立即处决。” docs.djangoproject.com/en/2.2/topics/db/transactions/…

【参考方案2】：

我猜您的调用者在 celery 开始处理任务之前尚未提交的事务中。因此 celery 找不到记录。这就是为什么添加倒计时使它起作用的原因。

在您的示例中，1 秒倒计时可能与 10 秒倒计时一样有效。我在整个代码中使用了 1 秒倒计时来处理这个问题。

另一种解决方案是停止使用事务。

【参考方案3】：

您可以使用on_commit 挂钩来确保在事务提交之前不会触发 celery 任务？

DjangoDocs#performing-actions-after-commit

这是 Django 1.9 中添加的功能。

from django.db import transaction

def do_something():
    pass  # send a mail, invalidate a cache, fire off a Celery task, etc.

transaction.on_commit(do_something)

您还可以将函数包装在 lambda 中：

transaction.on_commit(lambda: some_celery_task.delay('arg1'))

您传入的函数将在调用 on_commit() 的假设数据库写入成功提交后立即被调用。

如果在没有活动事务的情况下调用 on_commit()，回调将立即执行。

如果假设的数据库写入被回滚（通常是在 atomic() 块中引发未处理的异常时），您的函数将被丢弃并且永远不会调用。