如何为单个Celery任务实例设置唯一标识（Django环境）

方案1：将文件名作为任务参数传入，通过结果表查询 #

提交任务时把文件名作为参数直接传入任务函数，django_celery_results的TaskResult表会自动记录task_args或task_kwargs字段，后续可通过这些字段精准定位到处理目标文件的任务。

示例代码：

# 定义处理文件的任务
@app.task
def process_file(filename):
    # 文件处理逻辑
    ...

# 提交任务时传入文件名
task = process_file.apply_async(args=("batch-210520.json",))

查询对应文件的任务：

from django_celery_results.models import TaskResult

# 筛选处理batch-210520.json的所有任务
target_tasks = TaskResult.objects.filter(task_args__contains='"batch-210520.json"')

方案2：利用update_state添加自定义元数据 #

如果需要更直观的任务标识，可在任务执行过程中通过Celery的update_state方法，将文件名存入任务的状态元数据中，该数据会保存在TaskResult的meta字段里。

示例代码：

@app.task(bind=True)
def process_file(self, filename):
    # 更新任务状态，附加文件名元数据
    self.update_state(
        state='PROGRESS',
        meta={'filename': filename, 'current': 0, 'total': 100}
    )
    # 文件处理逻辑
    ...

# 提交任务
task = process_file.apply_async(args=("batch-210520.json",))

查询时直接过滤meta字段：

target_tasks = TaskResult.objects.filter(meta__contains='"filename": "batch-210520.json"')

方案3：自定义TaskResult模型（进阶） #

如果需要更结构化、高效的存储和查询，可以继承默认的TaskResult模型，新增filename字段，再配置Celery使用自定义模型。

1. 创建自定义模型：

# 你的app/models.py
from django.db import models
from django_celery_results.models import TaskResult

class FileProcessingTaskResult(TaskResult):
    filename = models.CharField(max_length=255, null=True, blank=True)

    class Meta:
        verbose_name = "文件处理任务结果"

2. 在settings.py中配置使用自定义模型：

CELERY_RESULT_BACKEND = 'django-db'
DJANGO_CELERY_RESULTS_TASK_RESULT_MODEL = 'your_app.models.FileProcessingTaskResult'

3. 在任务中更新自定义字段：

from your_app.models import FileProcessingTaskResult

@app.task(bind=True)
def process_file(self, filename):
    # 文件处理逻辑
    ...
    # 任务完成后更新filename字段
    if self.request.id:
        task_result = FileProcessingTaskResult.objects.get(task_id=self.request.id)
        task_result.filename = filename
        task_result.save()

后续查询可直接通过filename字段筛选，效率和准确性更高。

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