如何自定义Celery推送至RabbitMQ的消息存储格式

Celery默认推送的消息遵循其内置任务协议，会固定封装任务标识、参数、回调配置、任务ID等元数据，要完全替换为你需要的{"operation": "add", "listOfNumbers": [5,5]}结构，需要绕过默认的任务发布封装，同时自定义消费端的消息解析逻辑，具体操作如下：

1. 生产端：替换默认的delay调用，直接发布自定义格式消息 #

不要使用任务实例自带的delay()/apply_async()方法发消息，这两个方法会强制走Celery默认的消息封装逻辑。直接通过Celery底层的连接通道发布自定义结构的消息即可，示例代码：

from myproject.celery import app

def send_task(operation, numbers):
    msg = {
        "operation": operation,
        "listOfNumbers": numbers
    }
    # 直连RabbitMQ队列发布消息，不经过Celery任务协议封装
    with app.connection() as conn:
        target_queue = conn.SimpleQueue("your_task_queue") # 替换为你实际使用的队列名
        target_queue.put(msg)

需要触发加法任务时，直接调用send_task("add", [5,5])即可，此时RabbitMQ队列中存储的就是你定义的纯字典结构，没有额外的Celery封装字段。

2. 消费端：自定义消费者逻辑，适配自定义消息格式 #

默认Celery Worker只能识别符合官方任务协议的消息，需要重写消费者的消息处理逻辑，识别自定义格式的消息并路由到对应业务函数，示例：

from celery.worker.consumer import Consumer
from myproject.tasks import add

# 自定义消费者类
class CustomMsgConsumer(Consumer):
    def get_task_handler(self, *args, **kwargs):
        original_handler = super().get_task_handler(*args, **kwargs)

        def handler(message, *handler_args, **handler_kwargs):
            msg_body = message.payload
            # 匹配自定义格式消息
            if isinstance(msg_body, dict) and {"operation", "listOfNumbers"}.issubset(msg_body.keys()):
                # 建立操作名和业务函数的映射
                op_mapping = {"add": add}
                task = op_mapping.get(msg_body["operation"])
                if task:
                    # 解析参数执行业务逻辑
                    x, y = msg_body["listOfNumbers"]
                    task(x, y)
                    # 手动确认消息消费完成
                    message.ack()
                    return
            # 不符合自定义格式的消息，走Celery默认处理流程
            return original_handler(message, *handler_args, **handler_kwargs)
        return handler

之后在Celery实例的配置中指定使用这个自定义消费者：

# 在celery.py的app配置中添加
app.conf.update(
    worker_consumer=CustomMsgConsumer,
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json"
)

注意事项 #

• 完全替换消息结构后，你将无法使用Celery原生的任务ID追踪、失败重试、任务链/和弦、结果存储、异常自动上报等内置能力，如果需要这些能力，建议不要修改核心消息体，通过自定义请求头传递自定义字段即可。
• 如果你不需要使用Celery的任务调度、Worker管理等能力，只是想复用RabbitMQ连接，直接使用Celery依赖的底层消息库kombu实现独立的生产者消费者逻辑会更简单，不需要兼容Celery的任务协议。

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