如何在python3中使用amqp镜像队列？答案

【问题标题】：how to use amqp mirror queue in python3?如何在python3中使用amqp镜像队列？
【发布时间】：2012-12-20 06:19:16
【问题描述】：

我使用rabbitmq作为amqp服务器（集群）和kombu+py-amqp作为amqp客户端。我可以发送/接收普通消息队列的消息，但我不知道如何发送/接收镜像队列的消息，我谷歌找不到答案。如何发送/接收镜像队列消息？

我的代码：

 with Connection(hostname='192.168.1.10',userid='test',password='test',virtual_host='test') as conn:
        simple_queue = conn.SimpleQueue('test')

并获得异常

  File "./test_amqp.py", line 38, in send
    simple_queue = conn.SimpleQueue('test')
  File "/usr/local/lib/python3.2/site-packages/kombu/connection.py", line 671, in SimpleQueue
    exchange_opts, **kwargs)
  File "/usr/local/lib/python3.2/site-packages/kombu/simple.py", line 122, in __init__
    consumer = messaging.Consumer(channel, queue)
  File "/usr/local/lib/python3.2/site-packages/kombu/messaging.py", line 338, in __init__
    self.revive(self.channel)
  File "/usr/local/lib/python3.2/site-packages/kombu/messaging.py", line 350, in revive
    self.declare()
  File "/usr/local/lib/python3.2/site-packages/kombu/messaging.py", line 360, in declare
    queue.declare()
  File "/usr/local/lib/python3.2/site-packages/kombu/entity.py", line 471, in declare
    self.queue_declare(nowait, passive=False)
  File "/usr/local/lib/python3.2/site-packages/kombu/entity.py", line 497, in queue_declare
    nowait=nowait)
  File "/usr/local/lib/python3.2/site-packages/amqp/channel.py", line 1240, in queue_declare
    (50, 11),  # Channel.queue_declare_ok
  File "/usr/local/lib/python3.2/site-packages/amqp/abstract_channel.py", line 70, in wait
    return self.dispatch_method(method_sig, args, content)
  File "/usr/local/lib/python3.2/site-packages/amqp/abstract_channel.py", line 88, in dispatch_method
    return amqp_method(self, args)
  File "/usr/local/lib/python3.2/site-packages/amqp/channel.py", line 222, in _close
    (class_id, method_id), ChannelError)
amqp.exceptions.PreconditionFailed: Queue.declare: (406) PRECONDITION_FAILED - inequivalent arg 'x-ha-policy'for queue 'smarton' in vhost 'smarton': received none but current is the value 'all' of type 'longstr'

【问题讨论】：

标签： queue rabbitmq amqp mirror python-3.2

【解决方案1】：

当您声明一个队列时，所有选项都必须与服务器上已经存在的选项相同。在这种情况下，服务器的额外选项是'x-ha-policy': 'all'。

试试simple_queue = conn.SimpleQueue('test', queue_opts={"x-ha-policy": "all"})

我没有对此进行测试，但我认为它会起作用。

【讨论】：

我知道这一点，但我找不到将 {'x-ha-policy':'all'} 放在哪里

【解决方案2】：

也许kombu有一些问题，参数：{“x-ha-policy”：“all”}应该发布到entity.Queue.queue_arguments，但是entity.Queue没有设置值的方法entity.Queue.queue_arguments，我更改了 kombu.simple.SimpleQueue 并得到了正确的结果：

112         if not isinstance(queue, entity.Queue):
113             exchange = entity.Exchange(name, 'direct', **exchange_opts)
114             queue = entity.Queue(name, exchange, name, **queue_opts)
115+            queue.queue_arguments={'x-ha-policy':'all'}
116         else:
117             name = queue.name
118             exchange = queue.exchange

【讨论】：