如何解决尝试使用Celery + RabbitMQ设置优先级队列时,出现“ PreconditionFailed-队列的等价arg'x-max-priority'”错误
我有RabbitMQ设置,带有两个队列:low
和high
。我希望我的芹菜工人在消耗低优先级队列的任务之前先从高优先级队列中进行消耗。尝试将消息推送到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",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,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",in on_inbound_method
method_sig,payload,content,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'
这是我的芹菜配置
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 -A celery_app worker -l info --config celeryconfig --concurrency=16 -n "%h:celery" -O fair -Q high,low
我正在使用:
- RabbitMQ:3.7.17
- 芹菜:4.3.0
- Python:3.6.7
- 操作系统:Ubuntu 18.04.3 LTS仿生
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。