SpringBoot整合RabbitMQ 实现六种工作模式

比眉伴天荒 2024-04-08 08:28 64阅读 0赞

消息生产者或者发送者，使用 P 表示：

![在这里插入图片描述][daa5e73d468d4b1e83d132113242c0b2.jpeg_pic_center]

消息从生产端发送到消费端，一定要通过队列转发，使用 queue\_name 表示：

![在这里插入图片描述][a8ded1e9edeb47669c367b87d6009908.png]

消费的消费者或者接收者，使用 C 表示，如果有多个消费者也可以用 C1、C2 表示：

![68ee1234ee6f45b4a3d792bee795f4a8.png][]

1.  添加 maven 依赖

<dependency>   
      <groupId>org.springframework.boot</groupId>   
      <artifactId>spring-boot-starter-amqp</artifactId>    
      <version>2.2.1.RELEASE</version>
    </dependency>

1.  添加 application.yml 配置

spring:
      rabbitmq:
        host: 127.0.0.1
        port: 5672
        virtual-host: /
        username: guest
        password: guest

1.  消息生产

生产端发送消息，调用 RabbitTemplate 发送消息，比如：

@Autowired
    private RabbitTemplate rabbitTemplate;
    public String send() {
          
    rabbitTemplate.convertAndSend("routingKey","send message");
    }

1.  消费消息

消费消息使用队列监听注解 @RabbitListener, 添加队列名称就能消费发送到队列上的消息了：

@RabbitListener(queuesToDeclare = @Queue("queue_name"))
    public void consume(String message) {

### 1. 简单（simple）模式 ###

![在这里插入图片描述][1b965044a40b4d03b020e8a444bfcba7.png]

> 最简单的消息发送

*  生产者是消费者是一一对应，也叫做`点对点模式`, 生产者发送消息经过队列直接发送给消费者。
 *  生产者和消费者在发送和接收消息时，只需要指定队列名称，而不需要指定 Exchange 交换机。

生产消息:

@GetMapping("/simple-send")
    public String simpleSend() {
         
     rabbitTemplate.convertAndSend("simple","this is news");  return "ok";
     }

消费消息

@RabbitListener(queuesToDeclare = @Queue("simple"))
    public void consume(String message) {
        
      System.out.println(message);
      }

输出：

this is news

> `无需创建交换机和绑定队列，只需要匹配发送端和消费端的队列名称就能成功发送消息。`

### 二 工作模式 ###

![在这里插入图片描述][dbde2e4a117a423ba4644e170143bef4.png]

> 在多个消费者之间分配任务
> 
>  *  工作模式和简单模式差不多，只需要生产端、消费端、队列。
>  *  不同在于一个生产者、一个队列对应多个消费者，也就是一对多的关系。
>  *  在多个消费者之间分配消息（竞争消费者模式），类似轮询发送消息，每个消息都只发给一个消费者。

*  生产消息：

@GetMapping("/work-send")
    public String simpleSend() {
         
     rabbitTemplate.convertAndSend("work","this is news");  return "ok";
     }

*  消费消息：

@RabbitListener(queuesToDeclare = @Queue("work"))
    public void consume(String message) {
         
     System.out.println("first:" + message);
     }
    
    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void consumeSecond(String message) {
          
    System.out.println("second:" + message);
    }

创建一个生产者，两个消费者，发送两条消息，两个消费者分别接收到消息，输出：

first:this is newssecond:this is news

两个消费者，轮流消费消息。类似 nginx 负载均衡。

### 发布订阅模式 ###

![在这里插入图片描述][5b07a5e4ba1c46459eea9736c40dba80.png]

> 一次向多个消费者发送消息

**特点:**

> 发布订阅类似广播消息，每个消息可以同时发送给订阅该消息的消费者，  
> 上图中的 X 表示交换机，使用的扇形交换机 (fanout), 它将发送的消息发送到所有绑定交换机的队列。

*  创建队列、交换机以及绑定：

@Bean
    public FanoutExchange fanoutExchange() {
         
     return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");
     }
    
    @Beanpublic Queue psFirstQueue() {
          return new Queue("psFirstQueue");
    }
    
    @Beanpublic Queue psSecondQueue() {
          
    return new Queue("psSecondQueue");
    }
    
    @Bean
    public Queue psThirdQueue() {
          
    return new Queue("psThirdQueue");
    }
    
    @Bean
    public Binding routingFirstBinding() {
         
     return BindingBuilder.bind(psFirstQueue()).to(fanoutExchange());
     }
    
    @Bean
    public Binding routingSecondBinding() {
          
    return BindingBuilder.bind(psSecondQueue()).to(fanoutExchange());}
    
    @Bean
    public Binding routingThirdBinding() {
         
     return BindingBuilder.bind(psThirdQueue()).to(fanoutExchange());
     }

*  上面定义一个交换机 fanoutExchange。
 *  分别绑定三个队列 psFirstQueue、psSecondQueue、psThirdQueue。
 *  队列绑定交换机不需要 routingKey，直接绑定即可。

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-2P0G5V4K-1660459653174)(https://p3-sign.toutiaoimg.com/tos-cn-i-qvj2lq49k0/131e322bf878460eabf77f7e052a9edd~noop.image?\_iz=58558&from=article.pc\_detail&x-expires=1661064383&x-signature=HipEXVrUaBAr2xRuSBCmcbgLhyU%3D)\]][img-2P0G5V4K-1660459653174_https_p3-sign.toutiaoimg.com_tos-cn-i-qvj2lq49k0_131e322bf878460eabf77f7e052a9edd_noop.image_iz_58558_from_article.pc_detail_x-expires_1661064383_x-signature_HipEXVrUaBAr2xRuSBCmcbgLhyU_3D]

*  生产端：

@GetMapping("/publish-sub-send")
    public String publishSubSend() {
         
     rabbitTemplate.convertAndSend("PUBLISH_SUBSCRIBE_EXCHANGE", null, "publish/subscribe hello");  return "ok";
     }

无需指定 routingKey, 设置为 null。

*  消费端：

@RabbitListener(queues = "psFirstQueue")
    public void pubsubQueueFirst(String message) {
         
     System.out.println("【first】:" + message);
     }
    
    @RabbitListener(queues = "psSecondQueue")
    public void pubsubQueueSecond(String message) {
          
    System.out.println("【second】:" + message);
    }
    
    @RabbitListener(queues = "psThirdQueue")
    public void pubsubQueueThird(String message) {
         
     System.out.println("【third】:" + message);
     }

*  **输出：**

【first】: publish/subscribe hello
    【second】: publish/subscribe hello
    【third】: publish/subscribe hello

> **发送一条消息，绑定的队列都能接收到消息。**

### 路由模式 ###

![在这里插入图片描述][0a53ac0fcae34910a9a1faf6365be661.png]

> 根据 routingKey 有选择性的接收消息

特点

*  每个队列根据不同 routingKey 绑定交换机
 *  消息发送到交换机后通过 routingKey 发送给特定的队列，然后传到消费者消费。
 *  交换由扇形交换机 (fanout) 改成直连交换机 (direct)。

创建队列、交换机以及绑定：

@Bean
    public Queue routingFirstQueue() {
            
    return new Queue("routingFirstQueue");
    }
    
    @Bean
    public Queue routingSecondQueue() {
          
      return new Queue("routingSecondQueue");
      }
    @Bean
    public Queue routingThirdQueue() {
           
     return new Queue("routingThirdQueue");
    }
    
    @Bean
    public DirectExchange routingExchange() {
            
    return new DirectExchange("routingExchange");
    }
    
    @Bean
    public Binding routingFirstBind() {
           
     return BindingBuilder.bind(routingFirstQueue()).to(routingExchange()).with("firstRouting");
     }
    
    @Bean
    public Binding routingSecondBind() {
           
     return BindingBuilder.bind(routingSecondQueue()).to(routingExchange()).with("secondRouting");
     }
    @Bean
    public Binding routingThirdBind() {
            
    return BindingBuilder.bind(routingThirdQueue()).to(routingExchange()).with("thirdRouting");}

*  创建一个交换机，根据不同的路由规则匹配不同的队列 routingExchange, 根据不同的 routingKey 绑定不同的队列：
 *  firstRouting 路由键绑定 routingFirstQueue 队列。
 *  secondRouting 路由键绑定 routingSecondQueue 队列。
 *  thirdRouting 路由键绑定 routingThirdQueue 队列。  
    ![在这里插入图片描述][d045bcf0d6f644b6b1841993e9fe9d56.png]
 *  生产消息:

// 使用不同的routingKey 转发到不同的队列    
    @GetMapping("/routing-first")
    public String routingFirst() {
            rabbitTemplate.convertAndSend("routingExchange","firstRouting"," first routing message");    rabbitTemplate.convertAndSend("routingExchange","secondRouting"," second routing message");    rabbitTemplate.convertAndSend("routingExchange","thirdRouting"," third routing message");    return "ok";}

*  消费消息：

@RabbitListener(queues = "routingFirstQueue")
    public void routingFirstListener(String message) 
    {
            System.out.println("【routing first】" + message);}
    
    @RabbitListener(queues = "routingSecondQueue")
    public void routingSecondListener(String message) {
          
      System.out.println("【routing second】" + message);}
    
    @RabbitListener(queues = "routingThirdQueue")
    public void routingThirdListener(String message) {
            
    System.out.println("【routing third】" + message);}

**输出：**

【routing first】first routing message
    【routing second】second routing message
    【routing third】third routing message

**分析:**

rabbitTemplate.convertAndSend("routingExchange","firstRouting"," first routing message");

> 消息从生产者指定 firstRouting 路由键，找到对应的绑定队列 routingFirstQueue, 就被 routingFirstQueue 队列消费了。

### 主题模式 ###

> 基于某个主题接收消息

*特点:*  
路由模式发送的消息，是需要指定固定的 routingKey，如果想要针对一类路由。 比如：

*  只接收以.com 结尾的消息。
 *  www. 开头的消息。

主题模式就派上场了，路由模式和主题模式类似，路由模式是设置特定的 routingKey 绑定唯一的队列，而主题模式的是使用通配符匹配一个或者多个队列。

*  创建交换机和队列：

@Beanpublic Queue topicFirstQueue() {
          
      return new Queue("topicFirstQueue");}
    @Bean
    public Queue topicSecondQueue() {
            
    return new Queue("topicSecondQueue");}
    
    @Bean
    public Queue topicThirdQueue() {
           
     return new Queue("topicThirdQueue");}
    @Bean
    public TopicExchange topicExchange() {
            
    return new TopicExchange("topicExchange");}

*  使用通配符绑定交换机和交换机：

// .com 为结尾 
    @Bean
    public Binding topicFirstBind() {
               return BindingBuilder.bind(topicFirstQueue()).to(topicExchange()).with("*.com");}
    
     // www.为开头 
    @Bean
    public Binding topicSecondBind() {
              
    return BindingBuilder.bind(topicSecondQueue()).to(topicExchange()).with("www.#");}

通配符有两种，\* 和\#,

*   *  表示可以匹配一个。
 *  ## 表示可以匹配多个。 ##

**比如：**

*  \#.com 表示接收多个以.com 结尾的字段。 例如： taobao.com、www.taobao.com、www.jd.com。
 *  \*.com 表示接收一个以.com 结尾的字段。 例如： taobao.com、jd.com。多个字段是无法匹配的，比如 www.taobao.com、cn.taobao.com。
 *  www.\#可以匹配多个以 www 开头的字段。 例如 www.taobao、www.jd。
 *  www.\* 可以匹配一个以 www 开头的字段。 例如：www.taobao、www.jd。多个字段是无法匹配的, 比如 www.taobao.com、www.jd.com。
 *  
 *  生产消息:

@GetMapping("/topic-first-send")
    public String topicFirstSend() {
            
    rabbitTemplate.convertAndSend("topicExchange","www.taobao.com","www.taobao.com");    
    
    rabbitTemplate.convertAndSend("topicExchange","taobao.com","taobao.com");    
    
    rabbitTemplate.convertAndSend("topicExchange","www.jd","www.jd");   
     return "topic ok";
     }

*  消费消息：

@RabbitListener(queues = "topicFirstQueue")
    public void topicFirstListener(String message) {
            
    System.out.println("【topic first】" + message);}
    
    @RabbitListener(queues = "topicSecondQueue")
    public void topicSecondListener(String message) {
          
      System.out.println("【topic second】" + message);}

*  **输出：**

【topic second】www.taobao.com
    【topic first】taobao.com
    【topic second】www.jd

> www.\#可以匹配多个以 www. 开头的路由键，例如 www.taobao.com、www.jd。而 \*.com 只能匹配一个以.com 结尾的路由键，例如 taobao.com，而无法匹配 www.taobao.com。

### 6. RPC模式 ###

![在这里插入图片描述][f5bd6c5557db40fdb0a45f9fe11142cd.png]

> 消息有返回值

特点

*  PRC 模式和上面的几种模式唯一不同的点在于，该模式可以收到消费端的返回值。
 *  生成端接收消费端的返回值。
 *  消费端添加返回值：

@RabbitListener(queuesToDeclare =@Queue("rpcQueue"))
    public String rpcListener(String message) {
         
     System.out.println("【rpc接收消息】" + message);  
     return "rpc 返回" + message;
     }

*  生产端发送消息：

@GetMapping("/rpc-send") 
    public void rpcSend() {
          
    Object receive = rabbitTemplate.convertSendAndReceive("rpcQueue","rpc send message");  
    System.out.println("【发送消息消息】" + receive);
     }

*  输出：

【rpc接收消息】rpc send message
    【发送端接收消息】rpc 返回rpc send message

### 交换机类型 ###

上面的 订阅发布模式、路由模式以及主题模式使用到了不同的交换机, 分别是：

>  *  直连交换机 Direct
>  *  扇形交换机 Fanout
>  *  主题交换器 Topic

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-m35w5FvH-1660459653187)(https://p3-sign.toutiaoimg.com/tos-cn-i-qvj2lq49k0/7e402a6903cb45a49d2101709ad2574c~noop.image?\_iz=58558&from=article.pc\_detail&x-expires=1661064383&x-signature=tG4eyvgh5nT8Rw35qzZNKfoSRj4%3D)\]][img-m35w5FvH-1660459653187_https_p3-sign.toutiaoimg.com_tos-cn-i-qvj2lq49k0_7e402a6903cb45a49d2101709ad2574c_noop.image_iz_58558_from_article.pc_detail_x-expires_1661064383_x-signature_tG4eyvgh5nT8Rw35qzZNKfoSRj4_3D]

**Direct Exchange（直连）**

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Js3RZ8du-1660459653190)(https://p3-sign.toutiaoimg.com/tos-cn-i-qvj2lq49k0/eadf35186255415884407250ef80a49b~noop.image?\_iz=58558&from=article.pc\_detail&x-expires=1661064383&x-signature=irvxix%2FQH3RysVY91qYEXstf2dg%3D)\]][img-Js3RZ8du-1660459653190_https_p3-sign.toutiaoimg.com_tos-cn-i-qvj2lq49k0_eadf35186255415884407250ef80a49b_noop.image_iz_58558_from_article.pc_detail_x-expires_1661064383_x-signature_irvxix_2FQH3RysVY91qYEXstf2dg_3D]

直连交换机被应用在路由模式下，该交换机需要通过特定的 routingKey 来绑定队列，交换机只有接收到了匹配的 routingKey 才会将消息转发到对应的队列中，否则就不会转发消息。

路由模式使用直连交换机, 该模式下根据 routingKey 绑定特定的队列。

### Fanout Exchange(扇形) ###

![在这里插入图片描述][c7c80b9f192646999bf57a158ecec8ad.png]

扇形交换机没有路由键的概念，只需将队列绑定在交换机上，发送到交换机上的消息会转发到交换机所以绑定的队列里面，类似广播，只要打开收音机都能接收到广播消息。扇形交换机应用于发布订阅模式。

### Topic Exchange(主题) ###

![在这里插入图片描述][60acea349f2942c4927975d6b052ec2f.png]

主题模式是将路由键根据一个主题进行分类，和直连模式不同的是, 直连模式绑定特定的路由键，而主题模式使用通配符绑定路由键, 绑定键有两种：

*   *  表示可以匹配仅一个。
 *  ## 表示可以匹配零个或多个。 ##

### 总结 ###

整合 SpringBoot 实现 RabbitMQ 六种工作模式，并详细讲解 RabbitMQ 六种工作模式：

*  简单模式 无需创建交换机，匹配生产端和消费的 routingKey 即可。
 *  工作模式 多个消费端公平竞争同一个消息。
 *  发布订阅模式 一次向多个消费者发送消息。
 *  路由模式 根据特定的路由键转发消息。
 *  主题模式 根据通配符，匹配路由键转发消息。
 *  RPC 模式 生产端接收消费端发送的返回值。

[daa5e73d468d4b1e83d132113242c0b2.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/7e81665e3c204d37bf91bda53d331a89.jpeg
[a8ded1e9edeb47669c367b87d6009908.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e0a449fd2b424c439af342acc5f7a81e.png
[68ee1234ee6f45b4a3d792bee795f4a8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/1077626fa3074f2e870a632cbc20a643.png
[1b965044a40b4d03b020e8a444bfcba7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/6c0dedb4d2d043938802bfffbe770d89.png
[dbde2e4a117a423ba4644e170143bef4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/4f64384154b845899b94610935a98040.png
[5b07a5e4ba1c46459eea9736c40dba80.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/396729066f20404d926c6adc11f3ac85.png
[img-2P0G5V4K-1660459653174_https_p3-sign.toutiaoimg.com_tos-cn-i-qvj2lq49k0_131e322bf878460eabf77f7e052a9edd_noop.image_iz_58558_from_article.pc_detail_x-expires_1661064383_x-signature_HipEXVrUaBAr2xRuSBCmcbgLhyU_3D]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/dc8cafbfb0234cbb8b7d2635511b58f0.png
[0a53ac0fcae34910a9a1faf6365be661.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/1700ddc039ff4b758f32ed5acd89a123.png
[d045bcf0d6f644b6b1841993e9fe9d56.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/2a861a69212044148dcb451e27aea679.png
[f5bd6c5557db40fdb0a45f9fe11142cd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/bcce98e47970423fa2d10000dc1f9432.png
[img-m35w5FvH-1660459653187_https_p3-sign.toutiaoimg.com_tos-cn-i-qvj2lq49k0_7e402a6903cb45a49d2101709ad2574c_noop.image_iz_58558_from_article.pc_detail_x-expires_1661064383_x-signature_tG4eyvgh5nT8Rw35qzZNKfoSRj4_3D]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/e89e5ad123b142da9616d45d5567634f.png
[img-Js3RZ8du-1660459653190_https_p3-sign.toutiaoimg.com_tos-cn-i-qvj2lq49k0_eadf35186255415884407250ef80a49b_noop.image_iz_58558_from_article.pc_detail_x-expires_1661064383_x-signature_irvxix_2FQH3RysVY91qYEXstf2dg_3D]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/b93a9d0a42ae4d63a2d3c94631dee15c.png
[c7c80b9f192646999bf57a158ecec8ad.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/5a40b58c5f0a4c73a94a524b1c25d498.png
[60acea349f2942c4927975d6b052ec2f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/08/ebd9952187c04e05a0aa1a1d6b147b7f.png