乐趣区

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

jiezi

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
  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）模式

最简略的音讯发送

特点

  • 生产者是消费者是一一对应，也叫做 点对点模式, 生产者发送音讯通过队列间接发送给消费者。
  • 生产者和消费者在发送和接管音讯时，只须要指定队列名称，而不须要指定 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 的四种交换机
退出移动版