15.federation

federation and shovel

federation-exchange

The origin of the problem ：

City A Yes rabbitmqA, City B Yes rabbitmqB, When the city B Your app will send a message to exchangeA When , Because of the Internet , It will delay the sending time .

federation-exchange The role of ：

federation Provides a capability , Let the city B Of mq To receive exchangeA The news of , Then forward the message to the city A Of exchangeA

Case presentation

• Prepare two rabbitmq service , Ensure that each node operates independently

  • [email protected]
  • [email protected]

• Turn on... On each machine federation Related to the plug-in

  rabbitmq-plugins enable rabbitmq_federation --offline
  rabbitmq-plugins enable rabbitmq_federation_managemen --offline
  

• After opening, find the new tab on the management console page

  

• function ConsumerFeb Code for

  • adopt ConsumerFeb The code creates fed-queue and fed-exchange

  /** * Feb  Message consumer  */
  public class ConsumerFeb {
        
  
      private static final String  QUEUE_NAME="fed-queue";
      private static final String  EXCHANGE_NAME="fed-exchange";
  
      public static void main(String[] args) throws Exception {
        
          ConnectionFactory connectionFactory = new ConnectionFactory();
          connectionFactory.setHost("172.16.140.131");
          connectionFactory.setUsername("admin");
          connectionFactory.setPassword("123456");
  
          Connection connection = connectionFactory.newConnection();
          Channel channel = connection.createChannel();
  
          channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT);
  
          channel.queueDeclare(QUEUE_NAME,false,false,false,null);
  
          channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"aaa");
          
          System.out.println(" Waiting to receive message ");
  
          // The interface callback of how to consume the pushed message 
          DeliverCallback deliverCallback = (consumerTag, message) -> {
        
            String result = new String(message.getBody());
              System.out.println(" Consumer receives message , The message is ："+result);
          };
  
          // A callback interface for canceling consumption 
          CancelCallback cancelCallback = consumerTag -> {
        
              System.out.println(" Message consumption is interrupted ");
          };
  
          channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
      }
  }
  

• Feb add to upstream

• add to policy

  

• see federation status

  

• add to ConsumerJan Consumer code

  /** * Jan  Message consumer  */
  public class ConsumerJan {
        
  
      private static final String  QUEUE_NAME="federation-queue";
  
      public static void main(String[] args) throws Exception {
        
          ConnectionFactory connectionFactory = new ConnectionFactory();
          connectionFactory.setHost("172.16.140.130");
          connectionFactory.setUsername("admin");
          connectionFactory.setPassword("123456");
  
          Connection connection = connectionFactory.newConnection();
          Channel channel = connection.createChannel();
          channel.queueDeclare(QUEUE_NAME,false,false,false,null);
  
  
          System.out.println(" Waiting to receive message ");
  
          // The interface callback of how to consume the pushed message 
          DeliverCallback deliverCallback = (consumerTag, message) -> {
        
            String result = new String(message.getBody());
              System.out.println(" Consumer receives message , The message is ："+result);
          };
  
          // A callback interface for canceling consumption 
          CancelCallback cancelCallback = consumerTag -> {
        
              System.out.println(" Message consumption is interrupted ");
          };
  
          channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
      }
  }
  

• Add producer code

  /** * federation-exchange- Message producer  */
  public class Producer {
        
  
      private static final String  EXCHANGE_NAME="fed-exchange";
  
      public static void main(String[] args) throws Exception {
        
          // Create a connection factory 
          ConnectionFactory connectionFactory = new ConnectionFactory();
          connectionFactory.setHost("172.16.140.130");
          connectionFactory.setUsername("admin");
          connectionFactory.setPassword("123456");
          // Get the connection 
          Connection connection = connectionFactory.newConnection();
          Channel channel = connection.createChannel();
          String message = "hello world";
          channel.basicPublish(EXCHANGE_NAME, "aaa", null, message.getBytes(StandardCharsets.UTF_8));
  
          System.out.println(" Message sent ");
  
      }
  }
  

• Start consumer Feb, consumer Jan, Restart producer

  • Discovery producer sends message to Jan, adopt federation Forward the message to Feb, In consumer Feb You get the output from

federation-queue

shovel

• Open plug-in

  rabbitmq-plugins enable rabbitmq_shovel --offline
  rabbitmq-plugins enable rabbitmq_shovel_management --offline