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

Gitana ·

更新时间:2024-09-20

· 35 次阅读

前提概念

生产者

队列

消费者

SpringBoot整合RabbitMQ基本配置添加maven依赖

1. 简单（simple）模式

2. 工作模式

生产消息：

3. 发布订阅模式

特点

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

4. 路由模式

特点

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

5. 主题模式

特点

创建交换机和队列：

6. RPC模式

特点

消费端添加返回值：

交换机类型

Direct Exchange（直连）

Fanout Exchange(扇形)

Topic Exchange(主题)

总结

源码示例

参考文献

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);
}

输出：