RabbitMQ: Двунаправленные федеративные брокеры, как вы делаете все очереди поп-сообщения?

Question

У меня есть два брокера, настроенных [1] с плагином федерации. Оба указывают друг на друга как вверх по течению.

Мой тест:

• опубликовать сообщение о посредничать в
• потребляют брокер B

Результат:

• потребляя на брокера B работает
• < good> очередь на брокера B po п.с. сообщение
• < не хорошо> очередь на брокер А еще имеют сообщение
  • < причины, почему это не хорошо> Проблемы, которую я вижу: если я всегда публикую на одного брокера, а затем всегда потребляю другой -> тогда очередь на издательском брокере будет расти до тех пор, пока она не станет полной и не начнет отбрасывать сообщения.

В результате я хотел бы это:

• обе очереди на брокера А и B поп свои сообщения, когда потребитель потребляет брокер B

Как настроить RabbitMQ вывести сообщение из всех очередей, когда потребитель потребляет сообщение на брокере B? Прямо сейчас я пытаюсь сделать это с помощью плагина федерации RabbitMQ.

---

[1] Два брокеров указывают друг на друга, как и выше по потоку настроить им same way as described in the "simple example" given by the documentation кроме того, что существуют два брокера каждый из которых указывает друг на друга как вверх по течению. code for the publisher looks like this и code for the consumer looks like this.

Answer 1

Я создал эту окружающую среду:

Сервер A, сервер B.

Создано двунаправленной федеративной таким образом:

Федерация Upstream: Server_B = amqp://servera

Федерации Upstream: Server_A = amqp://serverb

Затем создал ту же политику на обоих серверах:

Pattern : ^fed\. 
Apply to: all 
federation-upstream-set:all 

Создана одна очередь для сервера A, вызываемая: fed.test1, затем создана потребитель к серверу B, как:

ConnectionFactory factory = new ConnectionFactory(); 
factory.setHost("localhost"); 
factory.setPort(5673); 
Connection connection = factory.newConnection(); 
Channel channel = connection.createChannel(); 


Consumer consumer = new DefaultConsumer(channel) { 
@Override 
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) 
         throws IOException { 
String message = new String(body, "UTF-8"); 
System.out.println(" [x] Message '" + message); 
       } 
      }; 
channel.basicConsume("fed.test1", true, consumer); 

Затем опубликовал сообщение на сервере A --->fed.test1

Сообщения было потреблено на сервере B и очередь числа сообщений является zero оба очереди (сервер A, сервер B).

Это работает так, как вы ожидали.

Надеюсь, это поможет.