指定队列名时celery==4.2.0不消费任务，docker部署如何排查？

问题根因 #

你遇到的问题是三个配置缺失/错误共同导致的：

• 未配置任务路由规则：Celery默认所有任务都会投递到名为celery的默认队列，你仅指定worker监听queue1，但任务没有被路由到该队列，自然无法被消费。不添加-Q参数时worker默认监听celery队列，所以功能正常。
• worker1服务缺失环境变量配置：你在broker服务定义了公共环境变量锚点&envfile，但worker1配置块没有引用该环境变量，导致worker1无法读取broker连接信息，连不上消息中间件自然无法消费任务。
• worker队列配置不符合需求：你当前两个worker都配置了监听queue1，没有对应第二个任务的独立队列，无法实现两个任务路由到不同队列的需求。

修复步骤 #

1. 补充Celery任务路由配置 #

修改celery.py文件，添加task_routes配置，指定不同任务投递到对应队列：

from celery import Celery
import os

app = Celery(
    'worker',
    broker=os.environ['CELERY_BROKER'],
    backend=os.environ['RABBITMQ_BACKEND']
)

# 新增任务路由规则
app.conf.task_routes = {
    "worker.get_message_list": {"queue": "queue1"},
    "worker.get_something_list": {"queue": "queue2"},
}

app.conf.beat_schedule = {
    "get-message": {
        "task": "worker.get_message_list",
        "schedule": 60
    },
    "get-somethingelse": {
        "task": "worker.get_something_list",
        "schedule": 60
    }
}

if __name__ == '__main__':
    app.start()

你也可以选择在beat_schedule的每个任务配置中添加options参数指定队列，效果一致：

"get-message": {
    "task": "worker.get_message_list",
    "schedule": 60,
    "options": {"queue": "queue1"}
},

2. 修正docker-compose.yml配置 #

• 给worker1添加环境变量引用
• 修改worker2的监听队列为queue2
  修正后的配置如下：

version: '3'
services:
  broker:
    image: rabbitmq:3
    environment: &envfile
      RABBITMQ_DEFAULT_USER: user
      RABBITMQ_DEFAULT_PASS: password
      CELERY_BROKER: {something}
      RABBITMQ_BACKEND: {something}
    ports:
      - 5672:5672

  worker1:
    build: .
    image: worker1
    restart: "always"
    environment: *envfile # 新增这行，补充环境变量
    command: ["celery", "worker", "--app=worker.app", "--concurrency=4", "--hostname=worker1@%h", "--loglevel=INFO", "-Q", "queue1"]
    depends_on:
      - broker
  
  worker2:
    build: .
    image: worker2
    restart: "always"
    environment: *envfile
    command: ["celery", "worker", "--app=worker.app", "--concurrency=4", "--hostname=worker2@%h", "--loglevel=INFO", "-Q", "queue2"] # 修改队列名为queue2
    depends_on:
      - broker

  scheduler:
    build: .
    image: scheduler
    restart: "always"
    environment: *envfile
    command: ["celery", "beat", "--app=worker.app", "--loglevel=INFO"]
    depends_on:
      - broker
      - worker1
      - worker2

3. 验证配置 #

重启服务后查看worker启动日志，正常情况下会打印监听的队列信息，类似：

[queues]
. queue1 exchange=celery(direct) key=queue1

确认无报错后即可实现两个任务分别路由到对应队列，由不同worker消费。

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