SpringBoot-RabbitMQ04-交换器【topic】介绍

╰+攻爆jí腚メ 2022-01-29 16:27 220阅读 0赞

# 交换器介绍 #

RabbitMQ中有三种主要的交互器分别如下

<table> 
 <thead> 
  <tr> 
   <th>交换器</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>direct</td> 
   <td align="left">发布与订阅 完全匹配</td> 
  </tr> 
  <tr> 
   <td>topic</td> 
   <td align="left">主体，规则匹配</td> 
  </tr> 
  <tr> 
   <td>fanout</td> 
   <td align="left">广播</td> 
  </tr> 
 </tbody> 
</table>

# topic介绍 #

TopicExchange 是比较复杂也比较灵活的 种路由策略,在TopicExchange 中,Queue 通过routingkey 绑定到 TopicExchange 上,当消息到达 TopicExchange 后,TopicExchange 根据消息的routingkey 消息路由到一个或者多 Queue上，相比direct模式topic会更加的灵活些。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]  
  本案例通过两个项目来实现，一个consumer项目和一个provider项目。

## 1.创建消费者 ##

**项目结构**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]

**配置文件**

spring.application.name=springcloud-mq
    spring.rabbitmq.host=192.168.88.150
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=dpb
    spring.rabbitmq.password=123
    
    #设置交换器的名称
    mq.config.exchange=log.topic
    
    #info 队列名称
    mq.config.queue.info=log.info
    
    #error 队列名称
    mq.config.queue.error=log.error
    
    # log 队列名称
    mq.config.queue.logs=log.all

**三个消费者**

@Component
    @RabbitListener(
            bindings=@QueueBinding(
                    value=@Queue(value="${mq.config.queue.info}",autoDelete="true"),
                            exchange=@Exchange(value="${mq.config.exchange}",type= ExchangeTypes.TOPIC),
                            key="*.log.info"
                    )
            )
    public class InfoReceiver { 
    
        /** * 接收消息的方法。采用消息队列监听机制 * @param msg */
        @RabbitHandler
        public void process(String msg){ 
            System.out.println("Info........receiver: "+msg);
        }
    }

@Component
    @RabbitListener(
            bindings=@QueueBinding(
                    value=@Queue(value="${mq.config.queue.error}",autoDelete="true"),
                            exchange=@Exchange(value="${mq.config.exchange}",type= ExchangeTypes.TOPIC),
                            key="*.log.error"
                    )
            )
    public class ErrorReceiver { 
    
        /** * 接收消息的方法。采用消息队列监听机制 * @param msg */
        @RabbitHandler
        public void process(String msg){ 
            System.out.println("Error........receiver: "+msg);
        }
    }

@Component
    @RabbitListener(
            bindings=@QueueBinding(
                    value=@Queue(value="${mq.config.queue.logs}",autoDelete="true"),
                            exchange=@Exchange(value="${mq.config.exchange}",type= ExchangeTypes.TOPIC),
                            key="*.log.*"
                    )
            )
    public class LogsReceiver { 
    
        /** * 接收消息的方法。采用消息队列监听机制 * @param msg */
        @RabbitHandler
        public void process(String msg){ 
            System.out.println("All........receiver: "+msg);
        }
    }

然后启动项目等待消息即可~

## 2.创建服务提供者 ##

**目录结构**

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 2]

**配置文件**

**三个服务提供者**

@Component
    public class UserSender { 
    
        @Autowired
        private AmqpTemplate rabbitAmqpTemplate;
        //exchange 交换器名称
        @Value("${mq.config.exchange}")
        private String exchange;
        /* * 发送消息的方法 */
        public void send(String msg){ 
            //向消息队列发送消息
            //参数一：交换器名称。
            //参数二：路由键
            //参数三：消息
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.debug", "user.log.debug....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.info", "user.log.info....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.warn","user.log.warn....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"user.log.error", "user.log.error....."+msg);
        }
    }

@Component
    public class ProductSender { 
    
        @Autowired
        private AmqpTemplate rabbitAmqpTemplate;
        //exchange 交换器名称
        @Value("${mq.config.exchange}")
        private String exchange;
        /* * 发送消息的方法 */
        public void send(String msg){ 
            //向消息队列发送消息
            //参数一：交换器名称。
            //参数二：路由键
            //参数三：消息
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"product.log.debug", "product.log.debug....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"product.log.info", "product.log.info....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"product.log.warn","product.log.warn....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"product.log.error", "product.log.error....."+msg);
        }
    }

@Component
    public class OrderSender { 
    
        @Autowired
        private AmqpTemplate rabbitAmqpTemplate;
        //exchange 交换器名称
        @Value("${mq.config.exchange}")
        private String exchange;
        /* * 发送消息的方法 */
        public void send(String msg){ 
            //向消息队列发送消息
            //参数一：交换器名称。
            //参数二：路由键
            //参数三：消息
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.debug", "order.log.debug....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.info", "order.log.info....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.warn","order.log.warn....."+msg);
            this.rabbitAmqpTemplate.convertAndSend(this.exchange,"order.log.error", "order.log.error....."+msg);
        }
    }

**单元测试**

@RunWith(SpringRunner.class)
    @SpringBootTest(classes = RabbitmqDirectProviderApplication.class)
    public class RabbitmqDirectProviderApplicationTests { 
    
        @Autowired
        private UserSender usersender;
        @Autowired
        private ProductSender productsender;
        @Autowired
        private OrderSender ordersender;
    
        @Test
        public void contextLoads() throws Exception{ 
            this.usersender.send("UserSender.....");
            this.productsender.send("ProductSender....");
            this.ordersender.send("OrderSender......");
        }
    }

启动服务观察消费者控制台的输出

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 3]

也可以观察控制台

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 4]

搞定~

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20220129/b71186bf397f4166b5477430418a7bcd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]: /images/20220129/4324d4a7856a40729492f7c3817e5c53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 2]: /images/20220129/f148855fb66a45279a5e5acde68f3b27.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 3]: /images/20220129/210afb0494e44dae9ee8d47747092802.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9kcGItYm9ib2thb3lhLXNtLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 4]: /images/20220129/a218a4c060fa452f8f0f43484ce96db5.png