Celery+RabbitMQ任务结果队列报错：Received and deleted unknown message

问题描述 #

使用Celery和RabbitMQ创建简单任务，将任务结果发送至名为"result"的队列。连接该队列处理结果时，系统抛出警告：Received and deleted unknown message. Wrong destination?!?，同时返回参数：'redelivered': False, 'exchange': '', 'routing_key': 'result'。无法正确获取并处理任务结果，确认路由存在问题但未找到解决方法。

日志信息 #

WARNING/MainProcess] Received and deleted unknown message.  Wrong destination?!?

The full contents of the message body was: body: {'task_id': '84913264-6294-4c8d-9eff-927d54f7d256', 'status': 'SUCCESS', 'result': 25, 'traceback': None, 'children': []} (121b)
{content_type:'application/json' content_encoding:'utf-8'
  delivery_info:{'consumer_tag': 'None4', 'delivery_tag': 14, 'redelivered': False, 'exchange': '', 'routing_key': 'result'} headers={}}

问题分析 #

日志中exchange: ''表明消息是通过RabbitMQ默认交换器发送到"result"队列的，但Celery的结果处理器无法识别这种无指定交换器的路由方式，因此判定消息不属于自身处理范围，抛出警告并删除消息。

解决步骤 #

• 配置Celery结果路由参数
  在Celery配置中明确指定结果交换器、类型和路由键，让结果消息通过规范路由发送：

  from celery import Celery
  
  app = Celery('tasks')
  app.conf.update(
      result_backend='rpc://',  # 指定RabbitMQ作为结果后端
      result_exchange='result_exchange',
      result_exchange_type='direct',
      result_routing_key='result'
  )
  

• 绑定队列到结果交换器
  确保"result"队列与配置的交换器绑定，路由键匹配。可通过代码创建绑定：

  from kombu import Exchange, Queue
  
  result_exchange = Exchange('result_exchange', type='direct')
  result_queue = Queue('result', exchange=result_exchange, routing_key='result')
  app.conf.task_queues = (result_queue,)
  

• 使用Celery原生接口获取结果
  优先使用Celery提供的AsyncResult获取结果，无需手动监听队列，这是最稳定的方式：

  from celery.result import AsyncResult
  
  task_result = AsyncResult('84913264-6294-4c8d-9eff-927d54f7d256', app=app)
  if task_result.successful():
      print(task_result.get())  # 输出任务结果25
  

• 对齐手动消费者的路由配置
  若必须手动监听队列，需保证消费者的交换器、路由键与Celery发送结果时的配置完全一致，同时处理逻辑兼容Celery的结果消息格式。

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