Django celery 任务重复 : can't lock DB?

Question

我的 django 应用程序允许用户相互发送消息，我将一些最近的消息汇集在一起​​，并使用 celery 和 redis 将它们发送到电子邮件中。

每次用户发送消息时，我都会向数据库添加一条消息，然后触发一个异步任务来汇集该用户过去 60 秒的消息，并将它们作为电子邮件发送。

tasks.pushMessagePool.apply_async(args = (fromUser,), countdown = 60)

如果用户在接下来的 60 秒内发送了 5 条消息，那么我的假设是应该创建 5 个任务，但只有第一个任务发送电子邮件，其他 4 个任务什么都不做。我实现了一个简单的锁定机制来确保消息只被考虑一次并确保数据库锁定。

@shared_task
def pushMessagePool(fromUser, ignore_result=True):
    lockCode = randint(0,10**9)
    data.models.Messages.objects.filter(fromUser = fromUser, locked=False).update(locked=True, lockCode = lockCode)
    M = data.models.Messages.objects.filter(fromUser = fromUser, lockCode = lockCode)
    sendEmail(M,lockCode)

使用此设置，我仍然会偶尔 (~10%) 重复。副本将在 10 毫秒内相互触发，并且它们具有不同的锁定代码。

为什么这个锁定机制不起作用？ celery 是否引用旧的数据库快照？那没有任何意义。

Answer 1

Djangojack，这里有类似的问题吗？但是对于 SQS。我不确定它是否也适用于 Redis？

When creating your SQS queue you need to set the Default Visibility timeout to some time that's greater than the max time you expect a task to run. This is the time SQS will make a message invisible to all other consumers after delivering to one consumer. I believe the default is 30 seconds. So, if a task takes more than 30 seconds, SQS will deliver the same message to another consumer because it assumes the first consumer died and did not complete the task.

摘自@gustavo-ambrozio 在 this answer 上的评论.