使用Celery连接RabbitMQ仲裁队列时如何禁用全局QoS

RabbitMQ 仲裁队列（quorum queue）原生不支持全局QoS机制，Celery默认启用全局QoS，因此触发540 NOT_IMPLEMENTED错误。

你只需要添加worker_default_qos_global = False配置即可禁用全局QoS，有两种配置方式可选：

方式1：在celeryconfig.py配置文件中添加 #

直接修改现有配置文件，新增一行配置即可：

from kombu import Queue

task_queues = [Queue(name="Replica_que", queue_arguments={"x-queue-type": "quorum"})]

task_routes = {
    'tasks.add': 'Replica_que',
}
# 新增以下配置禁用全局QoS
worker_default_qos_global = False

方式2：在业务代码中直接设置app.conf #

如果你不想修改配置文件，也可以在初始化Celery实例的代码中添加配置：

from celery import Celery
from time import sleep
import kombu


broker_uri=['amqp://xxxx:5672/', 'amqp://xxxx:5672/','amqp://xxx:5672/']
backend_uri="mongodb+srv://xxxxx"

app = Celery('TestApp', broker=broker_uri,backend=backend_uri)
# 新增禁用全局QoS配置
app.conf.worker_default_qos_global = False

app.config_from_object('celeryconfig')
app.conf.task_default_exchange='Replica_que'
app.conf.task_default_routing_key='Replica'

@app.task
def reverse(text):
    sleep(10)
    return text[:-1]

• 如果你使用的是Celery 4.x版本，配置项需调整为全大写格式：CELERY_WORKER_DEFAULT_QOS_GLOBAL = False
• 配置修改完成后重启Celery worker即可生效，不再抛出对应报错。

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