Celery worker指定-Q参数后仍消费所有绑定队列问题排查

• Celery命令行的-Q（指定消费队列）、-X（排除消费队列）参数仅作用于Celery原生注册的任务队列，对你手动通过bootsteps.ConsumerStep自定义注册的Kombu消费者完全不生效。你代码里直接把两个消费者都加到了worker的启动步骤里，所以每个启动的worker都会同时运行两个消费者，分别绑定两个队列消费。
• 自定义Kombu消费者没有遵循Celery的消息协议格式，worker主进程收到这类非Celery任务格式的消息时，会判定为未知消息，随机触发Received and deleted unknown message报错，多worker场景下还会出现channel抢占、消息路由异常的问题。

提供两种可选方案，第二种更贴合Celery的设计逻辑，稳定性更高：

方案1：改造自定义消费者，增加队列判断逻辑 #

在注册自定义ConsumerStep之前，先解析当前worker的启动参数，仅注册对应队列的消费者：

from celery.platforms import argv

# 解析启动命令中的队列配置
def get_allowed_queues():
    allowed = set()
    excluded = set()
    args = argv[1:]
    for i, arg in enumerate(args):
        if arg == '-Q' and i+1 < len(args):
            allowed.update(q.strip() for q in args[i+1].split(','))
        if arg == '-X' and i+1 < len(args):
            excluded.update(q.strip() for q in args[i+1].split(','))
    # 未指定-Q时默认取Celery默认队列
    if not allowed:
        from celery.app.defaults import DEFAULT_QUEUE
        allowed.add(DEFAULT_QUEUE)
    return allowed - excluded

allowed_queues = get_allowed_queues()

# 仅注册符合队列要求的消费者
if "myqueue" in allowed_queues:
    app.steps["consumer"].add(MyConsumer1)
if "myotherqueue" in allowed_queues:
    app.steps["consumer"].add(MyConsumer2)

同时为避免未知消息报错，给自定义Consumer增加auto_declare=False配置，启动worker时追加--without-mingle参数避免进程同步冲突。

方案2：改用Celery原生任务绑定队列（推荐） #

完全放弃自定义ConsumerStep，用Celery原生能力绑定队列，自动适配命令行队列过滤规则，无协议兼容问题：

# Celery配置中注册队列
app.conf.task_queues = [
    kombu.Queue('myqueue', exchange=kombu.Exchange('myexchange', type='direct', durable=True), routing_key='mykey', durable=True),
    kombu.Queue('myotherqueue', exchange=kombu.Exchange('myexchange', type='direct', durable=True), routing_key='mykey', durable=True),
]

# 两个任务分别绑定对应队列
@app.task(queue='myqueue')
def handle_queue1(body):
    print(f"\n### 1 ###\nBODY: {body}\n#########\n")

@app.task(queue='myotherqueue')
def handle_queue2(body):
    print(f"\n### 2 ###\nBODY: {body}\n#########\n")

该方案下保持原有消息发布逻辑不变，消息会被正常分发到两个队列，启动celery -A settings worker -Q myqueue --hostname 1@%h时就只会消费myqueue队列的消息，完全符合需求。

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