关于java:SpringBoot整合RabbitMQ实现六种工作模式

RabbitMQ次要有六种种工作模式，本文整合SpringBoot别离介绍工作模式的实现。

前提概念

生产者

音讯生产者或者发送者，应用P示意：

队列

音讯从生产端发送到生产端，肯定要通过队列转发，应用queue_name示意：

消费者

生产的消费者或者接收者，应用C示意，如果有多个消费者也能够用C1、C2示意：

SpringBoot整合RabbitMQ根本配置

1. 增加maven依赖

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

2. 增加application.yml 配置

spring:  rabbitmq:    host: 192.168.3.19    port: 5672    username: admin    password: 123456

3. 音讯生产

生产端发送音讯，调用RabbitTemplate发送音讯，比方：

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

4. 生产音讯

生产音讯应用队列监听注解@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

无需创立交换机和绑定队列，只须要匹配发送端和生产端的队列名称就能胜利发送音讯。

2. 工作模式

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

特点

• 工作模式和简略模式差不多，只须要生产端、生产端、队列。
• 不同在于一个生产者、一个队列对应多个消费者，也就是一对多的关系。
• 在多个消费者之间调配音讯（竞争消费者模式），相似轮询发送音讯，每个音讯都只发给一个消费者。

代码示例

• 生产音讯：

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

3. 公布订阅模式

一次向多个消费者发送音讯

特点

• 公布订阅相似播送音讯，每个音讯能够同时发送给订阅该音讯的消费者，
• 上图中的X示意交换机，应用的扇形交换机(fanout),它将发送的音讯发送到所有绑定交换机的队列。

代码示例

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

  @Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");}@Beanpublic Queue psFirstQueue() {return new Queue("psFirstQueue");}@Beanpublic Queue psSecondQueue() {return new Queue("psSecondQueue");}@Beanpublic Queue psThirdQueue() {return new Queue("psThirdQueue");}@Beanpublic Binding routingFirstBinding() {return BindingBuilder.bind(psFirstQueue()).to(fanoutExchange());}@Beanpublic Binding routingSecondBinding() {return BindingBuilder.bind(psSecondQueue()).to(fanoutExchange());}@Beanpublic 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

发送一条音讯，绑定的队列都能接管到音讯。

4. 路由模式

依据routingKey有选择性的接管音讯

特点

• 每个队列依据不同routingKey绑定交换机
• 音讯发送到交换机后通过routingKey发送给特定的队列，而后传到消费者生产。
• 替换由扇形交换机(fanout)改成直连交换机(direct)。

代码示例

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

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

• 创立一个交换机，依据不同的路由规定匹配不同的队列routingExchange,依据不同的routingKey绑定不同的队列：

  • firstRouting路由键绑定routingFirstQueue队列。
  • secondRouting路由键绑定routingSecondQueue队列。
  • thirdRouting路由键绑定routingThirdQueue队列。

• 生产音讯:

  @GetMapping("/routing-first")public String routingFirst() {  // 应用不同的routingKey 转发到不同的队列  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队列生产了。

5. 主题模式

基于某个主题接管音讯

特点

路由模式发送的音讯，是须要指定固定的routingKey，如果想要针对一类路由。
比方：

• 只接管以.com 结尾的音讯。
• www.结尾的音讯。

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

代码示例

• 创立交换机和队列：

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

• 应用通配符绑定交换机和交换机：

@Beanpublic Binding topicFirstBind() {    // .com 为结尾    return BindingBuilder.bind(topicFirstQueue()).to(topicExchange()).with("*.com");}@Beanpublic Binding topicSecondBind() {    // www.为结尾    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(主题)

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

• * 示意能够匹配仅一个。
• # 示意能够匹配零个或多个。

总结

整合SpringBoot实现RabbitMQ六种工作模式，并具体解说RabbitMQ六种工作模式：

• 简略模式

  • 无需创立交换机，匹配生产端和生产的routingKey即可。

• 工作模式

  • 多个生产端公平竞争同一个音讯。

• 公布订阅模式

  • 一次向多个消费者发送音讯。

• 路由模式

  • 依据特定的路由键转发音讯。

• 主题模式

  • 依据通配符，匹配路由键转发音讯。

• RPC模式

  • 生产端接管生产端发送的返回值。

源码示例

• https://github.com/jeremylai7/springboot-learning/tree/master/spring-rabbitmq/src/main/java/com/jeremy/pattern

参考

• RabbitMQ简介和六种工作模式详解
• RabbitMQ 的四种交换机