如何为Celery批量生成的任务设置关联业务的动态名称？

要实现每个任务实例的名称包含处理的数字（如celery_test.print_num.1），不需要重复定义任务函数，直接在生成任务签名时动态设置任务名称即可，具体修改如下：

修改代码实现 #

import logging
from time import sleep
from celery import Celery, group

logger = logging.getLogger(__name__)

CELERY_BROKER_URL = 'redis://default:redis@redis:6379/0'
celery_app = Celery('celery_test', broker=CELERY_BROKER_URL, result_backend=CELERY_BROKER_URL)   

@celery_app.task(bind=True)
def print_num(self, num):
    # 可在任务内部获取自定义任务名称
    logger.info(f"Task name: {self.request.name}")
    logger.info(f"Processing number: {num}")
    sleep(20)
    logger.info(f"Number processed: {num}")

# 生成签名时通过set()方法动态设置每个任务的名称
category_tasks = group(
    print_num.s(num).set(name=f"celery_test.print_num.{num}")
    for num in range(1, 11)  # 生成1-10的带编号任务名称
)
result = category_tasks.async_apply()

关键说明 #

• 使用print_num.s(num)创建任务签名后，通过.set(name=xxx)方法覆盖默认任务名称，每个任务实例会拥有唯一的、带参数标识的名称，在Flower/监控工具中可直接区分。
• 绑定的任务函数（bind=True）中，可通过self.request.name获取当前任务的自定义名称，方便日志输出和调试。
• 该方式不会重复注册任务，避免了循环定义任务函数带来的冗余问题。

其他可选方案 #

如果不需要修改任务名称，也可以通过以下方式增强辨识度：

• 添加任务元数据：生成签名时通过.set(headers={'endpoint': num})添加自定义元数据，Flower监控中可查看任务headers信息，任务内部也能通过self.request.headers['endpoint']获取该值。
• 自定义任务ID：通过.set(task_id=f"print_num_{num}")设置自定义任务ID，替代默认UUID，同样能在日志和监控中快速区分任务。

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