java

关注公众号 jb51net

关闭
首页 > 软件编程 > java > SpringBoot整合RabbitMQ

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

作者:小码code

这篇文章主要介绍了SpringBoot整合RabbitMQ实现六种工作模式,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下

RabbitMQ主要有六种种工作模式,本文整合SpringBoot分别介绍工作模式的实现。

前提概念

生产者

消息生产者或者发送者,使用P表示:

队列

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

消费者

消费的消费者或者接收者,使用C表示,如果有多个消费者也可以用C1C2表示:

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. 发布订阅模式

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

特点

代码示例

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

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

生产端:

@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有选择性的接收消息

特点

代码示例

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

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

生产消息:

@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,如果想要针对一类路由。
比如:

主题模式就派上场了,路由模式主题模式类似,路由模式是设置特定的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.#");
}

通配符有两种,*#,

比如:

例如: taobao.comwww.taobao.comwww.jd.com

例如: taobao.comjd.com。多个字段是无法匹配的,比如www.taobao.comcn.taobao.com

例如www.taobaowww.jd

例如:www.taobaowww.jd。多个字段是无法匹配的,比如www.taobao.comwww.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.comwww.jd。而*.com只能匹配一个以.com结尾的路由键,例如taobao.com,而无法匹配www.taobao.com

6. RPC模式

消息有返回值

特点

代码示例

消费端添加返回值:

@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 Exchange(直连)

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

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

Fanout Exchange(扇形)

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

Topic Exchange(主题)

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

总结

整合SpringBoot实现RabbitMQ六种工作模式,并详细讲解RabbitMQ六种工作模式:

无需创建交换机,匹配生产端和消费的routingKey即可。

多个消费端公平竞争同一个消息。

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

根据特定的路由键转发消息。

根据通配符,匹配路由键转发消息。

生产端接收消费端发送的返回值。

源码示例

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

参考文献

RabbitMQ简介和六种工作模式

详解RabbitMQ 的四种交换机

到此这篇关于SpringBoot整合RabbitMQ实现六种工作模式的文章就介绍到这了,更多相关SpringBoot整合RabbitMQ内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
阅读全文