Celery 任务调用了 3 次，但提供给任务的数据没有改变

Posted 2023-02-24

【中文标题】Celery 任务调用了 3 次，但提供给任务的数据没有改变

【英文标题】：Celery task called three times, but the data supplied to the task does not change

【发布时间】：2020-08-02 15:16:33

【问题描述】：

我有一个 Document 模型，其中有一个 FileField 用于上传图像、pdf 等文件，还有一个 ImageFields 用于从单个上传文件创建的八种不同图像大小。我使用 Pillow 将上传的图像转换为 8 种不同的尺寸。我允许在 DocumentAdmin 中上传多个文件，并在 save_model 中创建一个缩略图和大图像以显示在 DocumentAdmin 页面中，然后使用 celery 任务（使用 redis）在后台创建其他不同的图像大小并将它们保存到文档模型。

我的代码正确地创建了不同的图像，但我在 celery 任务中似乎遇到了一个奇怪的情况。 celery 任务使用与 save_model 中用于创建拇指和大图像的代码相同的代码来创建不同的图像大小。

save_model代码的相关部分：

def save_model(self, request, obj, form, change):
    if form.is_valid():
        if not change:
            # Uploading one or more images
            files = request.FILES.getlist('storage_file_name')
            if files:
                for f in files:
                    with transaction.atomic():
                        original_file_name, extension = os.path.splitext(f.name)
                        new_file_name = original_file_name + "." + settings.DEFAULT_IMAGE_EXTENSION
                        obj2 = Document()
                        # save the original file
                        if extension.lower() == ".pdf":
                        obj2.pdf_file = f
                        else:
                            obj2.storage_file_name = f
                        save_document_images(obj2, new_file_name, image_file=f, rot_angle=0, thumbnail=True, xsmall=False, small=False, medium=False, large=True, xlarge=False, xxlarge=False, xxxlarge=False)
                        obj2.save()
                        logger.debug("Starting transaction for document-id=%s, new_file_name=%s" % (obj2.document_id, new_file_name))
                        transaction.on_commit(lambda: tasks.create_additional_images_task.apply_async(args=[obj2.document_id, new_file_name,], kwargs='rot_angle':0, 'thumbnail': False, 'xsmall':True, 'small':True, 'medium':True, 'large':False, 'xlarge':True, 'xxlarge':True, 'xxxlarge':True)) 

还有 celery 任务的代码：

@app.task(bind=True)
def create_additional_images_task(self, *args, **kwargs):
    from memorabilia.models import Document
    obj2 = Document.objects.get(document_id=args[0])
    image_file = obj2.storage_file_name
    clogger.debug("document_id=%s, obj2=%s, new_file_name=%s, image_file=%s" % (args[0], obj2, args[1], image_file.name))
    for key in kwargs:
        clogger.debug("kwargs: key=%s, value=%s" % (key, kwargs[key]))
    save_document_images(obj2, args[1], image_file, kwargs['rot_angle'], kwargs['thumbnail'], kwargs['xsmall'], kwargs['small'], kwargs['medium'], kwargs['large'], kwargs['xlarge'], kwargs['xxlarge'], kwargs['xxxlarge']) 

例如，我上传了三张不同的图片。当我查看 Django 日志时，我看到 celery 任务被调用了三次，使用了三个不同的 document_ids 和图像名称：

[2020-04-19 17:10:18] DEBUG [memorabilia.admin.save_model:1739] Starting transaction for document-id=110, new_file_name=image-1.jpg
[2020-04-19 17:10:19] DEBUG [memorabilia.admin.save_model:1739] Starting transaction for document-id=111, new_file_name=image-2.jpg
[2020-04-19 17:10:20] DEBUG [memorabilia.admin.save_model:1739] Starting transaction for document-id=112, new_file_name=image-3.jpg

当我查看 celery 日志时，我看到以下内容：

[2020-04-19 17:10:20,665: INFO/MainProcess] Received task: memorabilia.tasks.create_additional_images_task[ec7699d8-7b73-4004-bb78-9371bcf0e1d6]  
[2020-04-19 17:10:20,666: DEBUG/MainProcess] TaskPool: Apply <function _fast_trace_task at 0x7f85ce09ae18> (args:('memorabilia.tasks.create_additional_images_task', 'ec7699d8-7b73-4004-bb78-9371bcf0e1d6', 'lang': 'py', 'task': 'memorabilia.tasks.create_additional_images_task', 'id': 'ec7699d8-7b73-4004-bb78-9371bcf0e1d6', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'retries': 0, 'timelimit': [None, None], 'root_id': 'ec7699d8-7b73-4004-bb78-9371bcf0e1d6', 'parent_id': None, 'argsrepr': "[112, 'image-3.jpg']", 'kwargsrepr': "'rot_angle': 0, 'thumbnail': False, 'xsmall': True, 'small': True, 'medium': True, 'large': False, 'xlarge': True, 'xxlarge': True, 'xxxlarge': True", 'origin': 'gen18286@tsunami', 'reply_to': '9dc88644-e203-30ff-a643-a14314f2572a', 'correlation_id': 'ec7699d8-7b73-4004-bb78-9371bcf0e1d6', 'delivery_info': 'exchange': '', 'routing_key': 'celery', 'priority': 0, 'redelivered': None, b'[[112, "image-3.jpg"], "rot_angle": 0, "thumbnail": false, "xsmall": true, "small": true, "medium": true, "large": false, "xlarge": true, "xxlarge": true, "xxxlarge": true, "callbacks": null,... kwargs:)

[2020-04-19 17:10:20,667: INFO/MainProcess] Received task: memorabilia.tasks.create_additional_images_task[eeb2ac99-ae1f-4530-9e41-fe923cfba912]  
[2020-04-19 17:10:20,667: DEBUG/MainProcess] TaskPool: Apply <function _fast_trace_task at 0x7f85ce09ae18> (args:('memorabilia.tasks.create_additional_images_task', 'eeb2ac99-ae1f-4530-9e41-fe923cfba912', 'lang': 'py', 'task': 'memorabilia.tasks.create_additional_images_task', 'id': 'eeb2ac99-ae1f-4530-9e41-fe923cfba912', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'retries': 0, 'timelimit': [None, None], 'root_id': 'eeb2ac99-ae1f-4530-9e41-fe923cfba912', 'parent_id': None, 'argsrepr': "[112, 'image-3.jpg']", 'kwargsrepr': "'rot_angle': 0, 'thumbnail': False, 'xsmall': True, 'small': True, 'medium': True, 'large': False, 'xlarge': True, 'xxlarge': True, 'xxxlarge': True", 'origin': 'gen18286@tsunami', 'reply_to': '9dc88644-e203-30ff-a643-a14314f2572a', 'correlation_id': 'eeb2ac99-ae1f-4530-9e41-fe923cfba912', 'delivery_info': 'exchange': '', 'routing_key': 'celery', 'priority': 0, 'redelivered': None, b'[[112, "image-3.jpg"], "rot_angle": 0, "thumbnail": false, "xsmall": true, "small": true, "medium": true, "large": false, "xlarge": true, "xxlarge": true, "xxxlarge": true, "callbacks": null,... kwargs:)

[2020-04-19 17:10:20,668: INFO/MainProcess] Received task: memorabilia.tasks.create_additional_images_task[04cb7f54-285d-4e0f-86e0-aaa2995d1cb4]  
[2020-04-19 17:10:20,669: DEBUG/MainProcess] TaskPool: Apply <function _fast_trace_task at 0x7f85ce09ae18> (args:('memorabilia.tasks.create_additional_images_task', '04cb7f54-285d-4e0f-86e0-aaa2995d1cb4', 'lang': 'py', 'task': 'memorabilia.tasks.create_additional_images_task', 'id': '04cb7f54-285d-4e0f-86e0-aaa2995d1cb4', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '04cb7f54-285d-4e0f-86e0-aaa2995d1cb4', 'parent_id': None, 'argsrepr': "[112, 'image-3.jpg']", 'kwargsrepr': "'rot_angle': 0, 'thumbnail': False, 'xsmall': True, 'small': True, 'medium': True, 'large': False, 'xlarge': True, 'xxlarge': True, 'xxxlarge': True", 'origin': 'gen18286@tsunami', 'reply_to': '9dc88644-e203-30ff-a643-a14314f2572a', 'correlation_id': '04cb7f54-285d-4e0f-86e0-aaa2995d1cb4', 'delivery_info': 'exchange': '', 'routing_key': 'celery', 'priority': 0, 'redelivered': None, b'[[112, "image-3.jpg"], "rot_angle": 0, "thumbnail": false, "xsmall": true, "small": true, "medium": true, "large": false, "xlarge": true, "xxlarge": true, "xxxlarge": true, "callbacks": null,... kwargs:)

创建了三个任务（1d6、912、cb4 - 使用任务 ID 的最后三位）。但是，所有三个任务都将其作为参数，[112, "image-3.jpg"] (document_id=112, file_name=image-3jpg) 当我查看文件系统时，我看到 celery 任务应该具有的图像为 document_ids 110 和 111 制作的图像尚未制作，但为 document_id 112 制作了图像。所有三个 document_id 的缩略图和大图像都已创建，因此制作图像的代码似乎可以正常工作。

我在代码中做错了什么？我必须承认，这是我第一次尝试使用芹菜，所以我有点失落。

谢谢！

标记

【参考方案1】：

1) 我会说你不正确地使用on_commit。根据文档，它应该是这样的

for f in files:
    with transaction.atomic():
        transaction.on_commit(...)

否则 Django 应该如何知道on_commit 处理的是哪个事务？所以你添加了三张图片，保存它们，这就是三个监听器的一个事务。

2) 请不要在for 语句中使用相同的import。只需在for之前将其移高

【讨论】：

感谢您的建议。我按照您的建议（在我的应用程序上方和中）更新了 save_model 的代码，但我得到的行为与以前完全相同。我在 Django 日志中看到，transaction.on_commit 被调用了三次，每个图像的值都是正确的，但是 celery 日志显示这三个工作人员只处理选择的最后一个图像 image-3.jpg。