Celery 4.4.7仅作为消费者对接外部Java生产者无法接收Redis消息问题

核心问题原因 #

1. 通信模式不匹配：原生Redis生产者使用的是Pub/Sub发布订阅模式推送消息，而Celery搭配Kombu使用Redis作为Broker时，默认采用List队列轮询拉取的模式存储、消费任务，两种模式的消息存储完全隔离，因此无法互通。
2. 消息格式不匹配：Celery消费的任务消息有固定的字段规范，不符合规范的消息会被直接丢弃。

解决方案 #

方案一：调整Java生产者逻辑（推荐，改动最小）

该方案无需修改现有Celery配置，仅需Java生产者调整两处逻辑即可：

1. 放弃使用PUBLISH命令发消息，改为调用LPUSH命令，写入的Redis键名就是你定义的Celery队列名redis.test.topic
2. 消息体按照Celery要求的格式构造，必填字段参考如下：

{
  "task": "task_process_message",
  "id": "全局唯一的任务ID字符串",
  "kwargs": {"message": "业务消息内容"},
  "args": []
}

你可以直接参考现有kombu_producer.py中的消息结构，Java侧构造相同结构的JSON字符串写入对应List即可。
你可以先修改redis_producer.py做本地验证：

import json
import redis

r = redis.Redis(host='localhost', port=6379)
for i in range(10):
    # 构造Celery兼容的消息结构
    celery_msg = {
        "task": "task_process_message",
        "id": str(i),
        "kwargs": {"message": f"Hello {i}"},
        "args": []
    }
    # 写入队列对应的Redis List
    r.lpush("redis.test.topic", json.dumps(celery_msg))

运行修改后的生产者，即可看到Celery Worker正常消费消息。

方案二：新增适配层（适用于Java生产者逻辑无法修改的场景）

如果Java侧必须保留Pub/Sub模式推送消息的逻辑，你可以单独部署一个轻量适配服务，接收Pub/Sub消息后转成Celery任务提交执行：

# pubsub_adapter.py 适配层示例
import json
import redis
# 导入你已经定义的Celery任务
from celery_consumer import task_process_message

redis_conn = redis.Redis(charset='utf-8', decode_responses=True)
pubsub = redis_conn.pubsub()
pubsub.subscribe('redis.test.topic')

for message in pubsub.listen():
    if message.get('type') == 'message':
        data = json.loads(message.get('data'))
        # 转成Celery任务提交
        task_process_message.delay(message=data['message'])

该方案无需修改现有Java生产者和Celery消费逻辑，仅需额外部署适配服务即可。

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