Django应用中Celery定时任务重复执行（两次）问题排查求助

近几周我们的Django应用中，Celery任务出现重复执行的问题（实际均为两次）。我在本地、预发布及生产环境做了大量测试，但始终无法复现和定位问题，目前毫无头绪。

• 近期我们从Celery v4升级至v5，同时还完成了其他多项系统升级
• 问题仅出现在耗时约15分钟的长时周期性任务（由Celery Beat调度）中
• 我们在托管Kubernetes环境中运行Celery，Beat和Worker采用独立部署，当前各有一个Pod

重复执行的任务示例 #

@idempotent_notification_task(notification_tag='CREDITS_DIRECTDEBIT_PAYMENT')
def create_directdebit_payments_for_credits():
    plus_admin_ids = Order.objects.filter(
        subscription__type=Subscription.PLUS, credits_auto_top_up_amount__gt=0).values_list('administration_id',
                                                                                            flat=True)
    admins = Administration.objects.filter(id__in=plus_admin_ids)
    for admin in admins:
        remaining_credits = admin.remaining_credits
        if remaining_credits < 0:
            po_exists = PurchaseOrder.objects.filter(administration=admin,
                                                     created__date=today(),
                                                     service=PurchaseOrder.SERVICE.CREDITS).exists()
            if po_exists:
                print(f"PurchaseOrder for client {admin.title} already exists!")
                return

            purchase_order = PurchaseOrder.objects.create(
                administration=admin, service=PurchaseOrder.SERVICE.CREDITS,
                credits_amount=admin.order.first().credits_auto_top_up_amount)
            payment = create_credits_mollie_payment_directdebit(purchase_order)
            Notification.create_credits_directdebit_payment(purchase_order)

自定义幂等装饰器（未解决问题） #

为修复问题，我编写了一个包装函数使任务具备幂等性，但即使如此问题仍未解决。该包装函数在本地测试中表现正常，但任务中的对象仍会被创建两次。

def idempotent_notification_task(notification_tag):
    """
    Decorator function that makes the celery task related to a notification template idempotent
    """
    def decorator(function):
        def wrapper(*args, **kwargs):
            nt = NotificationTemplate.objects.select_for_update().get(tag=notification_tag)
            with atomic():
                if nt.last_sent is not None and nt.last_sent.date() == today():
                    print(f"Notification {notification_tag} already sent today")
                    return
                else:
                    function()
                    nt.last_sent = now()
                    nt.save()
        return wrapper
    return decorator

异常现象 #

还有一个奇怪的现象：有时在Celery日志中能看到多次任务接收记录，有时看不到任务日志但任务确实已执行。

内容的提问来源于stack exchange，提问作者Gijsriet