尝试使用Celery+RabbitMQ设置优先级队列时,出现“PreconditionFailed - inequivalent arg 'x-max-priority' for queue”错误

jaql4c8m  于 2022-11-08  发布在  RabbitMQ
关注(0)|答案(1)|浏览(522)

我设置了RabbitMQ,其中有两个队列,名为:lowhigh。我想让我的celery工作线程在使用低优先级队列的任务之前先使用高优先级队列的任务。当我尝试将消息推送到RabbitMQ时,我收到以下错误

>>> import tasks
>>> tasks.high.apply_async()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/vagrant/.local/lib/python3.6/site-packages/celery/app/task.py", line 570, in apply_async
  **options
  File "/home/vagrant/.local/lib/python3.6/site-packages/celery/app/base.py", line 756, in send_task
    amqp.send_task_message(P, name, message,**options)
  File "/home/vagrant/.local/lib/python3.6/site-packages/celery/app/amqp.py", line 552, in send_task_message
  **properties
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/messaging.py", line 181, in publish
    exchange_name, declare,
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/connection.py", line 510, in _ensured
    return fun(*args,**kwargs)
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/messaging.py", line 194, in _publish
    [maybe_declare(entity) for entity in declare]
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/messaging.py", line 194, in <listcomp>
    [maybe_declare(entity) for entity in declare]
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/messaging.py", line 102, in maybe_declare
    return maybe_declare(entity, self.channel, retry,**retry_policy)
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/common.py", line 121, in maybe_declare
    return _maybe_declare(entity, channel)
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/common.py", line 145, in _maybe_declare
    entity.declare(channel=channel)
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/entity.py", line 609, in declare
    self._create_queue(nowait=nowait, channel=channel)
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/entity.py", line 618, in _create_queue
    self.queue_declare(nowait=nowait, passive=False, channel=channel)
  File "/home/vagrant/.local/lib/python3.6/site-packages/kombu/entity.py", line 653, in queue_declare
    nowait=nowait,
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/channel.py", line 1154, in queue_declare
    spec.Queue.DeclareOk, returns_tuple=True,
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/abstract_channel.py", line 80, in wait
    self.connection.drain_events(timeout=timeout)
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/connection.py", line 500, in drain_events
    while not self.blocking_read(timeout):
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/connection.py", line 506, in blocking_read
    return self.on_inbound_frame(frame)
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/method_framing.py", line 55, in on_frame
    callback(channel, method_sig, buf, None)
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/connection.py", line 510, in on_inbound_method
    method_sig, payload, content,
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/abstract_channel.py", line 126, in dispatch_method
    listener(*args)
  File "/home/vagrant/.local/lib/python3.6/site-packages/amqp/channel.py", line 282, in _on_close
    reply_code, reply_text, (class_id, method_id), ChannelError,
amqp.exceptions.PreconditionFailed: Queue.declare: (406) PRECONDITION_FAILED - inequivalent arg 'x-max-priority' for queue 'high' in vhost '/': received none but current is the value '10' of type 'signedint'

这是我的celery 配置

import ssl
broker_url="amqps://"
result_backend="amqp://"
include=["tasks"]
task_acks_late=True
task_default_rate_limit="150/m"
task_time_limit=300
worker_prefetch_multiplier=1
worker_max_tasks_per_child=2
timezone="UTC"
broker_use_ssl = {'keyfile': '/usr/local/share/private/my_key.key', 'certfile': '/usr/local/share/ca-certificates/my_cert.crt', 'ca_certs': '/usr/local/share/ca-certificates/rootca.crt', 'cert_reqs': ssl.CERT_REQUIRED, 'ssl_version': ssl.PROTOCOL_TLSv1_2}
from kombu import Exchange, Queue
task_default_priority=5
task_queue_max_priority = 10
task_queues = [Queue('high', Exchange('high'), routing_key='high', queue_arguments={'x-max-priority': 10}),]
task_routes = {'tasks.high': {'queue': 'high'}}

我有一个tasks.py脚本,其中定义了以下任务

from __future__ import absolute_import, unicode_literals
from celery_app import celery_app

@celery_app.task
def low(queue='low'):
    print("Low Priority")

@celery_app.task(queue='high')
def high():
    print("HIGH PRIORITY")

还有我的celery_app.py脚本:

from __future__ import absolute_import, unicode_literals
from celery import Celery
from celery_once import QueueOnce
import celeryconfig

celery_app = Celery("test")
if __name__ == '__main__':
    celery_app.start()

我用这个命令启动celery 工人

celery -A celery_app worker -l info --config celeryconfig --concurrency=16 -n "%h:celery" -O fair -Q high,low

我正在使用:

  • 兔子MQ:3.7.17
  • celery :4.3.0
  • Python语言:3.6.7
  • 操作系统:Ubuntu 18.04.3
23c0lvtd

23c0lvtd1#

最近我坚持同一个问题,发现了这个问题。我决定张贴可能的解决方案,为别人谁会发现它在未来。
当前的错误消息意味着队列已经声明为优先级10,但是现在它的签名包含优先级none。下面的示例是x-expires的类似问题,并给出了很好的解释:
Celery坚持要求每个客户端都事先知道队列是如何创建的。
要解决此问题,您可以更改以下内容:

  • 更改task_queue_max_priority(它定义了队列x-max-priority的默认值)或删除它。
  • 使用queue_arguments={'x-max-priority': 10}声明队列low,就像您对队列high所做的那样。

对我来说,当所有队列声明与先前创建的队列匹配时,问题就解决了。

相关问题