RabbitMQ Exchange几种模式简单使用

旧城等待， 2021-12-16 07:47 359阅读 0赞

AMQP协议中的核心思想就是生产者和消费者隔离，生产者从不直接将消息发送给队列。生产者通常不知道是否一个消息会被发送到队列中，只是将消息发送到一个交换机。先由Exchange来接收，然后Exchange按照特定的策略转发到Queue进行存储。同理，消费者也是如此。Exchange 就类似于一个交换机，转发各个消息分发到相应的队列中。

RabbitMQ提供了四种Exchange模式：fanout,direct,topic,header 。 header模式在实际使用中较少，本文只对前三种模式进行比较。

[完整源码地址][Link 1]  
1.Direct模式  
所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue。

Direct模式,可以使用rabbitMQ自带的Exchange：default Exchange 。所以不需要将Exchange进行任何绑定(binding)操作 。消息传递时，RouteKey必须完全匹配，才会被队列接收，否则该消息会被抛弃。  
举例：

RabbitMQConfig.java

@Configuration
    public class RabbitMQConfig {
        public static final String QUEUE = "yearns";
            /**
         * Direct模式
         *
         * @return
         */
        @Bean
        public Queue directQueue() {
            // 第一个参数是队列名字， 第二个参数是指是否持久化
            return new Queue(QUEUE, true);
        }
    }

RabbitMQReceiver.java

@Component
    public class RabbitMQReceiver {
        @RabbitListener(queues = RabbitMQConfig.QUEUE)
        public void receiverDirectQueue(User user) {
            System.out.println("Receiver : " + user.toString());
        }
    }

RabbitMQSender.java

@Component
    public class RabbitMQSender {
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void sendDirectQueue() {
            User user = new User();
            user.setUserId("123456");
            user.setName("yearns");
            this.rabbitTemplate.convertAndSend(RabbitMQConfig.QUEUE,user);
        }
        }

效果如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70]  
注意：发送者与接收者的Queue名字一定要相同，否则接收收不到消息  
2.Topic模式  
所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue。

Direct模式,可以使用rabbitMQ自带的Exchange：default Exchange 。所以不需要将Exchange进行任何绑定(binding)操作 。消息传递时，RouteKey必须完全匹配，才会被队列接收，否则该消息会被抛弃。  
举例：  
RabbitMQConfig.java

@Configuration
    public class RabbitMQConfig {
          public static final String TOPIC_QUEUE1 = "topic.yearns1";
      	  public static final String TOPIC_QUEUE2 = "topic.yearns2";
          public static final String TOPIC_EXCHANGE = "topic.exchange";
        /**
         * Topic模式
         *
         * @return
         */
        @Bean
        public Queue topicQueue1() {
            return new Queue(TOPIC_QUEUE1);
        }
    
        @Bean
        public Queue topicQueue2() {
            return new Queue(TOPIC_QUEUE2);
        }
    
        @Bean
        public TopicExchange topicExchange() {
            return new TopicExchange(TOPIC_EXCHANGE);
        }
    
        @Bean
        public Binding topicBinding1() {
            return BindingBuilder.bind(topicQueue1()).to(topicExchange()).with("lzc.message");
        }
    
        @Bean
        public Binding topicBinding2() {
            return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("lzc.#");
        }
    
    }

RabbitMQReceiver.java

@Component
    public class RabbitMQReceiver {
        /**
         * queues是指要监听的队列的名字
         */
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE1)
        public void receiveTopic1(User user) {
            System.out.println("【receiveTopic1监听到消息】" + user.toString());
        }
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE2)
        public void receiveTopic2(User user) {
            System.out.println("【receiveTopic2监听到消息】" + user.toString());
        }
    
    }

RabbitMQSender.java

@Component
    public class RabbitMQSender {
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void sendTopic() {
            User user1 = new User();
            user1.setUserId("123456");
            user1.setName("lizhencheng");
    
            User user2 = new User();
            user2.setUserId("456789");
            user2.setName("张三");
            this.rabbitTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, "lzc.message", user1 );
            this.rabbitTemplate.convertAndSend(RabbitMQConfig.TOPIC_EXCHANGE, "lzc.lzc", user2);
    
        }
        }

效果如下：  
![在这里插入图片描述][20190705153142246.png]  
![在这里插入图片描述][20190705153202222.png]  
由输出记录可以看出lzc.message可以被receiveTopic1和receiveTopic2所接收，而lzc.lzc只能被receiveTopic2接收。

3.Fanout模式  
所有发送到Fanout Exchange的消息都会被转发到与该Exchange 绑定(Binding)的所有Queue上。

Fanout Exchange 不需要处理RouteKey 。只需要简单的将队列绑定到exchange 上。这样发送到exchange的消息都会被转发到与该交换机绑定的所有队列上。类似子网广播，每台子网内的主机都获得了一份复制的消息。

所以，Fanout Exchange 转发消息是最快的。  
举例：  
RabbitMQConfig.java

@Configuration
    public class RabbitMQConfig {
              public static final String TOPIC_QUEUE3 = "topic.yearns3";
        public static final String TOPIC_QUEUE4 = "topic.yearns4";
    
        public static final String FANOUT_EXCHANGE = "fanout.exchange";
    
        @Bean
        public Queue topicQueue3() {
            return new Queue(TOPIC_QUEUE3);
        }
        @Bean
        public Queue topicQueue4() {
            return new Queue(TOPIC_QUEUE4);
        }
    
    
        /**
         * Fanout模式
         * Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。
         * @return
         */
        @Bean
        public FanoutExchange fanoutExchange() {
            return new FanoutExchange(FANOUT_EXCHANGE);
        }
        @Bean
        public Binding fanoutBinding1() {
            return BindingBuilder.bind(topicQueue3()).to(fanoutExchange());
        }
        @Bean
        public Binding fanoutBinding2() {
            return BindingBuilder.bind(topicQueue4()).to(fanoutExchange());
        }
    }

RabbitMQReceiver.java

@Component
    public class RabbitMQReceiver {
        /**
         * queues是指要监听的队列的名字
         */
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE3)
        public void receiveTopic3(User user) {
            System.out.println("【receiveTopic3监听到消息】" + user.toString());
        }
        @RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE4)
        public void receiveTopic4(User user) {
            System.out.println("【receiveTopic4监听到消息】" + user.toString());
        }
    
    }

RabbitMQSender.java

@Component
    public class RabbitMQSender {
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void sendFanout() {
            User user = new User();
            user.setUserId("123456");
            user.setName("lizhencheng");
    
            /**
             * 注意， 这里的第2个参数为空。
             * 因为fanout 交换器不处理路由键，只是简单的将队列绑定到交换器上，
             * 每个发送到交换器的消息都会被转发到与该交换器绑定的所有队列上
             */
    
            this.rabbitTemplate.convertAndSend(RabbitMQConfig.FANOUT_EXCHANGE, "", user );
        }
        }

效果如下：  
![在这里插入图片描述][20190705153901530.png]  
由输出可以看出，两个接收者都接收到了消息，因为交换机FANOUT\_EXCHANGE绑定了两个队列。

[Link 1]: https://gitee.com/yearns/yearns-kafka
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI0MzQ3NTQx_size_16_color_FFFFFF_t_70]: /images/20211214/8e7776503e9841a280b4dea52ebec7f9.png
[20190705153142246.png]: /images/20211214/dd5ff9a1ae1242b58d506312a2e94c5c.png
[20190705153202222.png]: /images/20211214/200509357dd040a585f850118c0374b3.png
[20190705153901530.png]: /images/20211214/9107ec4c8e9348ae938cdcd10bf10081.png