Я использую RabbitMQ 3.5.1 с плагинами rabbit_presence_exchange (бинарный дистрибутив) и rabbitmq_event_exchange (для отладки этой проблемы) и клиентом Python Pika.
Плагин присутствия работает, давая вам новый тип обмена: x -sence .Привязка очереди к этому с помощью ключа маршрутизации генерирует уведомления о присутствии, когда очередь связана и не связана (например, где ключом маршрутизации является имя пользователя).Привязка очереди без ключа маршрутизации подписывает вас на получение уведомлений о присутствии.
Это нормально, я могу успешно генерировать и получать уведомления о присутствии, подобные этой.Однако теперь я хотел бы направить сообщения о присутствии через обмен.Первоначально я пытался использовать обмен заголовками, но я не видел ни одного приходящего сообщения, поэтому я переключился на обмен разветвления (на случай, если я настроил сопоставление заголовка неправильно), но я до сих пор не вижу ничего, что приходит через.
Это мой сценарий для генерации и получения сообщений о присутствии без дополнительного обмена (т. Е. Тот, который работает):
#!/usr/bin/env python3
import pika
import names
MY_NAME = names.get_first_name()
PRESENCE_EXCHANGE = 'presence'
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.exchange_declare(exchange=PRESENCE_EXCHANGE,
exchange_type='x-presence')
result = channel.queue_declare('', exclusive=True)
queue_name = result.method.queue
print('My name is %s and my queue is %r' % (MY_NAME, queue_name))
channel.queue_bind(exchange=PRESENCE_EXCHANGE,
queue=queue_name,
routing_key=MY_NAME)
channel.queue_bind(exchange=PRESENCE_EXCHANGE,
queue=queue_name,
routing_key='')
def on_message(ch, method, properties, body):
print(method, '\n', properties, '\n', body)
exchange = method.exchange
if exchange == PRESENCE_EXCHANGE:
action = properties.headers['action']
who = properties.headers['key']
if action == 'bind':
print(' [+] %s has come online.' % (who,))
elif action == 'unbind':
print(' [-] %s has gone offline.' % (who,))
channel.basic_consume(queue=queue_name,
on_message_callback=on_message,
auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
try:
channel.start_consuming()
except KeyboardInterrupt:
pass
finally:
connection.close()
Я изменил вышеприведенное, чтобы маршрутизировать сообщения о присутствии ввстроенный обмен вентиляторами и привязка моей очереди к этому:
...
print('My name is %s and my queue is %r' % (MY_NAME, queue_name))
channel.queue_bind(exchange=PRESENCE_EXCHANGE,
queue=queue_name,
routing_key=MY_NAME)
channel.exchange_bind(source=PRESENCE_EXCHANGE,
destination='amq.fanout',
routing_key='')
channel.queue_bind(exchange='amq.fanout',
queue=queue_name)
def on_message(ch, method, properties, body):
...
Я озадачен тем, почему обмен не получает сообщения.Erlang не является одним из моих языков, поэтому у меня возникают проблемы при попытке прочитать исходный код плагина присутствия, чтобы определить, поддерживается ли он (хотя я не могу понять, почему это не будет).
Если кто-нибудьесть какие-либо идеи (или лучший способ обработки присутствия с RabbitMQ), я хотел бы услышать это.
РЕДАКТИРОВАТЬ:
С этим кодом и двумя работающими клиентами, мои обмены ипривязки выглядят так:
Listing exchanges ...
direct
amq.direct direct
amq.fanout fanout
amq.headers headers
amq.match headers
amq.rabbitmq.event topic
amq.rabbitmq.log topic
amq.rabbitmq.trace topic
amq.topic topic
presence x-presence
Listing bindings ...
exchange amq.gen-6aU7qS-ikR4cLmxcT6VKDQ queue amq.gen-6aU7qS-ikR4cLmxcT6VKDQ []
exchange amq.gen-MiyEpW9VIxD49PE9SqATFA queue amq.gen-MiyEpW9VIxD49PE9SqATFA []
amq.fanout exchange amq.gen-6aU7qS-ikR4cLmxcT6VKDQ queue amq.gen-6aU7qS-ikR4cLmxcT6VKDQ []
amq.fanout exchange amq.gen-MiyEpW9VIxD49PE9SqATFA queue amq.gen-MiyEpW9VIxD49PE9SqATFA []
presence exchange amq.fanout exchange []
presence exchange amq.gen-6aU7qS-ikR4cLmxcT6VKDQ queue Sheila []
presence exchange amq.gen-MiyEpW9VIxD49PE9SqATFA queue Joaquin []