По следам мануала: RabbitMQ + federation

Ещё один мануал, по которому настраивал недавно - это мануал по обмену RabbitMQ через federation plugin. Задача та же - обмен сообщениями между хостом на Windows 10 и Ubuntu.


Я делал в первую очередь federation для обменов (exchange) и целью было прокинуть обмен между двумя нодами, привязав его к локальной очереди через связывание (binding), которое позволяет получать из обмена сообщения на приём.

И столкнулся с проблемами. Вот стандартный мануал в разделе Define an upstream описывает то, как настроить федерацию. И возникает простой вопрос: где эти опции настраивать - на ноде, которая будет принимать сообщения или посылать? Первая мысль - на отправляющей... И это не так. Их нужно настраивать на downstream. Поэтому мануал становится немного непонятным, и настраивать трудновато.

Для примера, вот тут мануал гораздо проще и доступнее описывает те же действия, показывая то, что нужно делать для каждой ноды. Единственное что - не надо настраивать федерацию для обменов с префиксом amp. Любых, даже своих - это внутренние обмены и для них, у меня, даже для своих с таким префиксом, настройка не удалась. Ну и vhost непонятно зачем создаётся, скорее всего для безопасности.

А  в остальном - после такого остаётся только проверить доступность портов.

Тестировал в итоге простыми скриптами, на что-то более сложное не хватило времени и желания (test.direct - имя обмена):

#!/usr/bin/env python
import pika
import logging
logging.basicConfig()
c = pika.PlainCredentials('user', 'password')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='upstream host', credentials=c))
channel = connection.channel()


i = 1
while (i < 100):
    channel.basic_publish(exchange='test.direct', routing_key='', body='Hello World!')
    print(" [x] Sent 'Hello World!'")
    i = i + 1
connection.close()

Что же касается приёма, то тут достаточно было использовать очередь, привязанную к обмену:

#!/usr/bin/env python
import pika
import logging
logging.basicConfig()
c = pika.PlainCredentials('user', 'password')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()


def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)


channel.basic_consume(
    queue='test', on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()