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

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

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

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

添加 maven 依赖

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

添加 application.yml 配置

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

消息生产

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

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

消费消息

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

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

1. 简单（simple）模式

最简单的消息发送

生产者是消费者是一一对应，也叫做 点对点模式 , 生产者发送消息经过队列直接发送给消费者。
生产者和消费者在发送和接收消息时，只需要指定队列名称，而不需要指定 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

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

二工作模式

在多个消费者之间分配任务

工作模式和简单模式差不多，只需要生产端、消费端、队列。

不同在于一个生产者、一个队列对应多个消费者，也就是一对多的关系。

在多个消费者之间分配消息（竞争消费者模式），类似轮询发送消息，每个消息都只发给一个消费者。

生产消息：

@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 负载均衡。

发布订阅模式

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

特点:

发布订阅类似广播消息，每个消息可以同时发送给订阅该消息的消费者，

上图中的 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，直接绑定即可。

生产端：

@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

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

路由模式

根据 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 队列。

生产消息:

// 使用不同的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模式

消息有返回值

特点

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

Direct Exchange（直连）

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

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

Fanout Exchange(扇形)

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

Topic Exchange(主题)

表示可以匹配仅一个。

表示可以匹配零个或多个。

总结

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

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

1. 简单（simple）模式

二工作模式

发布订阅模式

路由模式

主题模式

表示可以匹配多个。

6. RPC模式

交换机类型

Fanout Exchange(扇形)

Topic Exchange(主题)

表示可以匹配零个或多个。

总结

继续阅读

springboot2整合ActiveMQ

SpringBoot——日志文件设置

微人事学习笔记（持续更新）登录页面服务端登录接口制作前后端接口对接Home页面制作前后端分离权限管理思路探讨权限管理：后端接口设计基础信息设置页面

首次阿里云部署学生级web服务

rabbitmq连接出错，记录一下

Spring Boot：静态资源

RabbitMQ：交换机（fanout exchange）

mysql5.7的sql优化

jeecg-boot的shiro认证过程2. jwtFilter

SpringSecurity整合短信验证码认证

基于Springboot+VUE+Mysql的大学生校园疫情防控管理系统

springboot使用 swagger-ui 2.10.5 有关版本更新带来的问题

rabbitmq 命令行操作

Rabbitmq命令行手动创建队列

RabbitMQ命令行的基本操作

spring boot @SpringBootApplication @EnableAutoConfiguration 的工作原理

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

1. 简单（simple）模式

二 工作模式

发布订阅模式

路由模式

主题模式

表示可以匹配多个。

6. RPC模式

交换机类型

Fanout Exchange(扇形)

Topic Exchange(主题)

表示可以匹配零个或多个。

总结

继续阅读

二工作模式