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

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

前提概念

生产者

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

队列

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

消费者

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

SpringBoot整合RabbitMQ根本配置

增加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: 192.168.3.19
    port: 5672
    username: admin
    password: 123456

音讯生产

生产端发送音讯，调用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

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

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 news
second:this is news

两个消费者，轮流生产音讯。相似nginx负载平衡。

3. 公布订阅模式

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

特点

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

代码示例

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

@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");
}

@Bean
public Queue psFirstQueue() {
return new Queue("psFirstQueue");
}

@Bean
public 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

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

4. 路由模式

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

生产音讯:

@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绑定惟一的队列，而主题模式的是应用通配符匹配一个或者多个队列。

代码示例

创立交换机和队列：

@Bean
public 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");
}

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

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

@Bean
public 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 的四种交换机

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

前提概念

生产者

队列

消费者

SpringBoot整合RabbitMQ根本配置

1. 简略（simple）模式

特点

代码示例

2. 工作模式

特点

代码示例

3. 公布订阅模式

特点

代码示例

4. 路由模式

特点

代码示例

5. 主题模式

特点

代码示例

6. RPC模式

特点

代码示例

交换机类型

Direct Exchange（直连）

Fanout Exchange(扇形)

Topic Exchange(主题)

总结

源码示例

参考

评论

发表回复取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

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

前提概念

生产者

队列

消费者

SpringBoot整合RabbitMQ根本配置

1. 简略（simple）模式

特点

代码示例

2. 工作模式

特点

代码示例

3. 公布订阅模式

特点

代码示例

4. 路由模式

特点

代码示例

5. 主题模式

特点

代码示例

6. RPC模式

特点

代码示例

交换机类型

Direct Exchange（直连）

Fanout Exchange(扇形)

Topic Exchange(主题)

总结

源码示例

参考

评论

发表回复 取消回复

更多文章

DDN HPC 存储硬件架构设计深度分析

探秘IO500：从Lustre并行文件系统出发，开启HPC存储性能新征程

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

发表回复取消回复