Celery+RabbitMQ如何设置队列消费优先级 优先消费Q1再消费Q2

以下配置完全适配你当前使用的Celery 5.0.5、Python 3.8、RabbitMQ 3.8技术栈，无需升级组件：

方案一：队列级优先级配置（推荐，完全匹配需求） #

该方案可直接实现worker优先消费完Q1所有消息后，再开始消费Q2的消息，配置步骤如下：

• 配置Celery队列参数
  在Celery实例启动文件中新增队列定义，给Q1设置更高的优先级权重，同时指定worker按优先级顺序消费队列：

  from kombu import Queue
  
  app = Celery('app')
  # 定义队列，RabbitMQ队列需声明x-max-priority参数才支持优先级能力
  app.conf.task_queues = (
      Queue('Q1', queue_arguments={'x-max-priority': 10}),
      Queue('Q2', queue_arguments={'x-max-priority': 1}),
  )
  # 配置worker按优先级高低选择要消费的队列
  app.conf.worker_queue_order = 'priority'
  

  如果不想修改代码，也可以提前在RabbitMQ中手动创建Q1、Q2队列，创建时给Q1设置max-priority为10、Q2设置为1，再在Celery配置中加上worker_queue_order = 'priority'即可。
• 调整worker预取配置
  避免worker提前预取大量Q2的消息导致优先级不生效，启动worker时添加--prefetch-multiplier 1参数，也可以直接在Celery配置中添加worker_prefetch_multiplier = 1。
• 启动worker
  最终启动命令修改为：

  celery -A app worker -Q Q1,Q2 --prefetch-multiplier 1
  

方案二：消息级优先级配置（适合同队列内任务分级场景） #

如果后续需要更细粒度的任务优先级区分，可以使用该方案：

• 队列配置和方案一一致，需给队列声明x-max-priority参数
• 发送任务时指定优先级参数：

  # 发往Q1的任务优先级设为最高
  your_task.apply_async(queue='Q1', priority=10)
  # 发往Q2的任务优先级设为最低
  your_task.apply_async(queue='Q2', priority=1)
  

注意：如果RabbitMQ中已经存在旧的Q1、Q2队列，需要先删除旧队列再重启RabbitMQ和Celery服务，否则队列的优先级参数不会生效。

内容的提问来源于stack exchange，提问作者Bun T.