如何在Celery Beat发布的Celery任务消息或头部添加自定义值？

你遇到的问题主要有两个核心原因：代码中的拼写错误，以及before_task_publish信号里修改任务消息的方式不符合Celery的规范。下面一步步帮你修正并解决问题：

1. 先修复基础拼写错误 #

你的发布端代码里有个低级笔误：my_dat应该是my_data，这会直接导致赋值失败，先修正这个问题：

@before_task_publish.connect
def task_before_publish_handler(*args, **kwargs):
    my_data={"foo":"bar"}
    # 修正拼写错误
    kwargs['request'][1]['my_data']=my_data
    return kwargs

不过即使修正了拼写，你大概率还是拿不到数据——因为直接修改kwargs['request']的方式并不是Celery推荐的任务消息修改方式，下面是正确的实现方案：

2. 正确在before_task_publish中添加自定义数据 #

Celery的before_task_publish信号提供了headers（任务头部）和body（任务参数）两个核心参数，你可以根据需求把自定义数据放到这两个位置：

方式一：添加到任务参数（Worker端从任务kwargs中读取） #

from celery.signals import before_task_publish

@before_task_publish.connect
def add_custom_data_to_task(sender=None, body=None, **kwargs):
    # body的结构是 (任务位置参数列表, 任务关键字参数字典)
    task_args, task_kwargs = body
    # 把自定义数据加入任务的关键字参数
    task_kwargs['my_data'] = {"foo": "bar"}
    # 更新body确保修改生效
    kwargs['body'] = (task_args, task_kwargs)

方式二：添加到任务头部（Worker端从请求头部读取） #

from celery.signals import before_task_publish

@before_task_publish.connect
def add_custom_data_to_headers(sender=None, headers=None, **kwargs):
    # 直接在任务头部插入自定义字段
    headers['my_data'] = {"foo": "bar"}

3. Worker端正确获取自定义数据 #

在Worker端的task_received信号中，对应不同的存储位置，用不同的方式读取：

如果是添加到任务参数： #

from celery.signals import task_received

@task_received.connect
def task_receive_handler(request=None, **kwargs):
    # 从任务的关键字参数中提取自定义数据
    my_data = request.kwargs.get('my_data')
    print("从任务参数获取自定义数据：", my_data)

如果是添加到任务头部： #

from celery.signals import task_received

@task_received.connect
def task_receive_handler(request=None, **kwargs):
    # 从请求头部提取自定义数据
    my_data = request.headers.get('my_data')
    print("从任务头部获取自定义数据：", my_data)

为什么原代码无法获取数据？ #

你原代码中直接修改kwargs['request'][1]，但before_task_publish信号里的request并非可直接修改的有效传递对象，而且Celery内部会对任务消息做序列化/反序列化处理，这个位置的修改不会被纳入最终发送的消息中。只有通过修改body或headers，才能确保自定义数据被正确序列化并传递到Worker端。

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