如何为Celery group内的单个任务分配自定义task_name以支持单独撤销

解决方案 #

可以实现给Group内每个子任务单独指定task_name，你只需要在构造每个子任务的签名（Signature）时，调用set()方法为单个任务单独配置属性即可，无需在Group的apply_async层面传参。

方法1：单独给每个子任务指定自定义task_name #

for item in response_json:
    # 构造任务签名后通过set方法单独配置task_name
    task_sig = some_task.s(item).set(
        task_name=f"自定义任务名前缀_{item.get('业务标识字段', 'default')}"
    )
    job_list.append(task_sig)

job = group(job_list)
result = job.apply_async(queue="queue1")

方法2：（更适合单独撤销场景）预生成子任务唯一task_id #

如果你的核心需求是后续可以单独撤销组内单个任务，更推荐直接为每个子任务预生成唯一的task_id，后续直接通过id操作即可，不需要依赖自定义task_name：

import uuid
from celery.result import AsyncResult

job_list = []
# 存储业务标识和task_id的映射，方便后续查找对应任务
biz_id_to_task_id = {}

for item in response_json:
    # 假设item里的id是你的业务唯一标识
    biz_id = item["id"]
    custom_task_id = str(uuid.uuid4())
    task_sig = some_task.s(item).set(task_id=custom_task_id)
    job_list.append(task_sig)
    biz_id_to_task_id[biz_id] = custom_task_id

# 执行任务组
job = group(job_list)
result = job.apply_async(queue="queue1")

# 后续需要撤销单个任务时的示例：
target_biz_id = 1001 # 要撤销的任务对应的业务id
target_task_id = biz_id_to_task_id[target_biz_id]
# 撤销任务，terminate参数控制是否终止已经在运行的任务
AsyncResult(target_task_id).revoke(terminate=True)

实现原理 #

Celery的Signature对象的set()方法用于为单个任务签名设置独立的执行参数，这些参数的优先级高于Group层面传入的通用参数，完全可以实现每个子任务独立配置task_name、队列、优先级、超时时间等属性。

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