Spring Boot 中 RabbitMQ 的使用
目录
在 RabbitMQ 的工作模式_发布订阅和工作模式的区别-CSDN博客 中,我们学习了 RabbitMQ 的 7 种工作模式,接下来,在本篇文章中,我们就来在 Spring Boot 中实现常见的工作模式(由于 RPC 模式 和 发布确认模式 使用较少,因此在这里就不进行介绍了),进而学习在 Spring Boot 中如何使用 RabbitMQ
在编写代码之前,我们需要先创建项目,引入依赖,并配置基本信息
引入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
或是在创建项目时,添加依赖:
接着,需要添加配置
添加配置
spring:
rabbitmq:
host: 49.232.238.62
port: 5672 #默认为5672
username: admin
password: 123456
virtual-host: test02
或是:
spring:
rabbitmq:
addresses: amqp://admin:123456@49.232.238.62:5672/test02
在这里,使用的虚拟机为 test02,因此,需要创建虚拟机 test02:
创建成功:
接下来,就可以开始编写代码了
Simple(简单模式)
在简单模式下,只有一个生产者和一个消费者,生产者生产的消息存储到队列后,由这个消费者消费
我们先来实现生产者代码
生产者代码
为了方便进行测试,我们通过 接口 来发送消息
创建 Constants 类,定义 简单模式 下使用的队列名称:
public class Constants {
// 简单模式
public static final String SIMPLE_QUEUE = "simple.queue";
}
接着,需要声明队列
创建 RabbitMQConfig 类,创建 简单模式 下使用的队列:
注意不要导错包了,当前我们使用的队列位于 org.springframework.amqp.core 包下
使用 QueueBuilder 声明队列:
import com.example.springrabbitmq.constant.Constants;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
// 简单模式
@Bean("simpleQueue")
public Queue simpleQueue() {
return QueueBuilder.durable(Constants.SIMPLE_QUEUE).build();
}
}
由于声明的队列来自于第三方的包,且后续 工作模式、发布定义模式等使用的队列也在 RabbitMQConfig 中定义,因此需要使用 @Bean 注解
其中 durable 表示 持久化,调用 durable 方法,表示创建一个持久化队列
接着,调用 build 方法创建 Bean
不要忘记添加 @Configuration 注解
在这里,我们直接使用 内置的交换机 来进行路由,因此,不需要声明交换机 以及 绑定交换机和队列
接着,我们创建 ProducerController 类,实现 简单模式 下的生产者:
在 Spring 中,使用 RabbitTemplate 来操作 RabbitMQ,因此,我们将它注入进来:
@RestController
@RequestMapping("/producer")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
}
接着,使用 convertAndSend 方法来发送消息:
public void convertAndSend(String exchange, String routingKey, final Object object) throws AmqpException
exchange:交换机名称,使用内置的交换机时,为 ""
routingKey:路由键,用内置交换机时,routingKey 为队列名称
object:要发送的消息
@RequestMapping("simple")
public String simple() {
for (int i = 0; i < 20; i++) {
String message = "simple... " + i;
rabbitTemplate.convertAndSend("", Constants.SIMPLE_QUEUE, message);
}
return "OK";
}
完整代码:
import com.example.springrabbitmq.constant.Constants;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/producer")
public class ProducerController {
@Autowired
private RabbitTemplate rabbitTemplate;
@RequestMapping("simple")
public String simple() {
for (int i = 0; i < 20; i++) {
String message = "simple... " + i;
rabbitTemplate.convertAndSend("", Constants.SIMPLE_QUEUE, message);
}
return "OK";
}
}
运行代码,观察 Queues and Streams:
此时并未创建队列
访问 127.0.0.1:8080/producer/simple,再次观察:
simple.queue 队列被创建,且队列中已有 20 条消息,也就是说,当我们运行程序时,队列并不会立即创建,而是当我们访问接口,要向队列中发送消息时,才会创建队列
我们查看其中一条消息:
消息正确发送
接下来,我们继续实现消费者代码
消费者代码
消费者需要 监听队列,当队列中有消息时,从队列中获取消息并消费,因此,我们创建监听类 SimpleListener
public class SimpleListener {
}
接下来,如何监听队列中的消息呢?
Spring 为我们提供了 @RabbitListener 注解,用于监听 RabbitMQ 队列,通过这个注解,我们可以定义一个方法,以便从队列中接收消息
@RabbitListener 支持多种参数类型,这些参数类型代表了从队列接收到的消息和相关信息
常见参数类型:
String:消息内容
Message(org.springframework.amqp.core.Message):Spring AMQP 的 Message 类,返回原始的消息体以及消息属性(如:消息ID、内容、队列信息等)
Channel(com.rabbitmq.client.Channel):RabbitMQ 的通道对象,可以用于进行更高级的操作,如手动确认消息
接下来,我们就来实现消费者监听队列:
使用 @RabbitListener 时,需要指定监听的队列
import com.example.springrabbitmq.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class SimpleListener {
@RabbitListener(queues = Constants.SIMPLE_QUEUE)
public void simpleListener(Message message) {
System.out.println("listener 从队列 " + Constants.SIMPLE_QUEUE +" 中接收到消息:" + message);
}
}
不要忘记添加 @Component,将其交给 Spring 管理
我们再次运行代码,观察打印的日志:
队列中的 20 条消息瞬间就被消费者消费掉了
可以看到,Message 中包含了 交换机、RoutingKey、deliverTage、监听的队列等信息
且观察 Connection 和 Channel
此时有一个 Connection:
两个 Channel:
哪一个是消费者的哪一个是生产者的呢?
我们可以从 Message rates 中确定,发布消息的是生产者,消费消息的是消费者
也可以根据 consumer 确定:
而当有多个消费者时,可以通过 ConsumerTag 来确定不同的消费者:
接下来,我们继续学习工作队列模式
Work Queue(工作队列)
工作队列模式下,有一个生产者和多个消费者,多个消费者共同消费消息
首先在 Constants 中声明工作队列模式下使用的队列:
// 工作模式
public static final String WORK_QUEUE = "work.queue";
在 RabbitMQConfig 中声明队列:
// 工作队列模式
@Bean("workQueue")
public Queue workQueue() {
return QueueBuilder.durable(Constants.WORK_QUEUE).build();
}
我们仍然使用内置的交换机进行路由,因此,也就不需要声明交换机和绑定交换机和队列
接下来, 我们就来实现生产者
生产者代码
工作队列模式与简单模式的区别在于工作模式下有多个消费者,因此生产者的消费代码与简单模式相同
在 ProducerController 中发送消息:
@RequestMapping("/work")
public String work() {
for (int i = 0; i < 20; i++) {
String message = "work... " + i;
rabbitTemplate.convertAndSend("", Constants.WORK_QUEUE, message);
}
return "OK";
}
运行,并访问 http://127.0.0.1:8080/producer/work,观察结果:
成功创建队列,且 20 条消息成功发送
接着,我们继续实现消费者代码
消费者代码
消费者的代码也与简单模式下的代码相同,只是在这里我们需要创建两个消费者:
我们可以定义 WorkListener1 和 WorkListener2,分别定义一个方法来监听:
import com.example.springrabbitmq.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class WorkListener1 {
@RabbitListener(queues = Constants.WORK_QUEUE)
public void queueListener1(Message message) {
System.out.println("listener1 从队列 " + Constants.SIMPLE_QUEUE +" 中接收到消息:" + message);
}
}
import com.example.springrabbitmq.constant.Constants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
public class WorkListener2 {
@RabbitListener(queues = Constants.WORK_QUEUE)
public void queueListener2(Message message) {
System.out.println("listener2 从队列 " + Constants.SIMPLE_QUEUE +" 中接收到消息:" + message);
}
}
再次运行,观察结果:
可以看到两个消费者共同消费了这 20 条消息,且它们的 deliveryTag 都是从 1 开始计数的
接下来,我们继续学习 发布定义模式
Publish/Subscribe(发布/订阅)
在发布/订阅模式中,每个消费者收到的消息都是相同的,且多了 Exchange 角色
在前面 简单模式 和 工作队列模式 下,图中都没有出现交换机,但实际上,生产者生产的消息都是先发送到交换机,然后再路由到队列中的。在前两种模式下,直接使用 RabbitMQ 提供的内置交换机就可以实现,因此,并没有突出交换机的存在,但实际上生产者生产的消息不会直接投递到队列中
因此,在 发布/订阅 模式下,我们就需要声明交换机,并绑定队列和交换机
先在 Constants 类中声明 发布订阅模式 下使用的交换机和队列:
// 发布订阅模式
public static final String PUBLISH_CHANGE = "fanout";
public static final String PUBLISH_QUEUE_1 = "publish.queue.1";
public static final String PUBLISH_QUEUE_2 = "publish.queue.2";
接着,在 RabbitMQConfig 中声明队列:
// 发布订阅模式
@Bean("fanoutQueue1")
public Queue fanoutQueue1() {
return QueueBuilder.durable(Constants.PUBLISH_QUEUE_1).build();
}
@Bean("fanoutQueue2")
public Queue fanoutQueue2() {
return QueueBuilder.durable(Constants.PUBLISH_QUEUE_2).build();
}
还需要声明交换机
在 发布订阅模式 下,交换机的类型为 Fanout(广播)
在 Spring 中,使用 FanoutExchange 表示广播类型的交换机
// 声明交换机
@Bean("fanoutExchange")
public FanoutExchange fanoutExchange() {
return ExchangeBuilder.fanoutExchange(Constants.PUBLISH_CHANGE).durable(true).build();
}
使用 ExchangeBuilder 的 fanoutExchange 方法创建广播类型的交换机
接着,需要将队列和交换机进行绑定
在 Spring 中,使用 Binding(org.springframework.amqp.core.Binding)表示交换机与队列的绑定关系
使用 BindingBuilder 进行绑定:
但此时队列不能自动注入, 因为此时有多个 Queue 类型的队列
我们可以使用 @Qualifier 注解指定我们要绑定的队列(fanoutExchange 由于只有一个,可以不指定):
// 绑定交换机和队列
@Bean("fanoutQueueBinding1")
public Binding fanoutQueueBinding1(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange,@Qualifier("fanoutQueue1") Queue queue) {
return BindingBuilder.bind(queue).to(fanoutExchange);
}
@Bean("fanoutQueueBinding2")
public Binding fanoutQueueBinding2(@Qualifier("fanoutExchange") FanoutExchange fanoutExchange, @Qualifier("fanoutQueue2") Queue queue) {
return BindingBuilder.bind(queue).to(fanoutExchange);
}
接着,我们就可以使用接口发送消息了
生产者代码
在 ProducerController 中使用 rabbitTemplate 发送消息
@RequestMapping("/fanout")
public String fanout() {
for (int i = 0; i < 20; i++) {
String message = "fanout ... " + i;
rabbitTemplate.convertAndSend(Constants.PUBLISH_CHANGE, "", message);
}
return "OK";
}
注意,发布订阅模式下,routingKey 为空,表示所有队列都可以收到消息
运行程序,访问 http://127.0.0.1:8080/producer/fanout,并观察结果
两个队列中都有 20 条消息,消费发送成功
接着,我们继续编写消费者代码
消费者代码
发布订阅模式下,消费者的代码与工作模式下的代码相同
为了方便观察,在这里就直接将两个消费者写在一个类中:
@Component
public class FanoutListener {
@RabbitListener(queues = Constants.PUBLISH_QUEUE_1)
public void queueListener1(String message) {
System.out.println("listener1 从队列 " + Constants.PUBLISH_QUEUE_1 +" 中接收到消息:" + message);
}
@RabbitListener(queues = Constants.PUBLISH_QUEUE_2)
public void queueListener2(String message) {
System.out.println("listener2 从队列 " + Constants.PUBLISH_QUEUE_2 +" 中接收到消息:" + message);
}
}
由于我们只关心消息的内容,因此,可以只使用 String 类型来接收消息
再次运行,观察打印的日志:
两个消费者都消费了 20 条相同的消息
我们继续看路由模式
Routing(路由模式)
Routing 模式下,队列与交换机之间的绑定,不再是任意的绑定了,而是需要指定一个 BindingKey
生产者在向 交换机 发送消息时,也需要指定消息的 RoutingKey
交换机不会将消息发送给每一个绑定的 key,而是会根据消息的 RoutingKey 进行判断,只有队列绑定时的 BindingKey 和发送消息的 RoutingKey 完全一致时,才会接收消息
先在 Constants 中声明路由模式下使用的队列和交换机:
// 路由模式
public static final String ROUTING_CHANGE = "routing";
public static final String ROUTINT_QUEUE_1 = "routing.queue.1";
public static final String ROUTINT_QUEUE_2 = "routing.queue.2";
接着,在 RabbitMQConfig 中声明队列:
// 路由模式
@Bean("routingQueue1")
public Queue routingQueue1() {
return QueueBuilder.durable(Constants.ROUTINT_QUEUE_1).build();
}
@Bean("routingQueue2")
public Queue routingQueue2() {
return QueueBuilder.durable(Constants.ROUTINT_QUEUE_2).build();
}
声明交换机:
注意,路由模式下,交换机类型为 Direct(定向)
在 Spring 中,使用 DirectExchange表示广播类型的交换机
// 声明交换机
@Bean("routingExchange")
public DirectExchange routingExchange() {
return ExchangeBuilder.directExchange(Constants.ROUTING_CHANGE).durable(true).build();
}
使用 ExchangeBuilder 的 directExchange 方法创建通配符类型的交换机
绑定队列和交换机:
// 绑定交换机和队列
@Bean("routingQueueBinding1")
public Binding routingQueueBinding1(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue1") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with("a");
}
@Bean("routingQueueBinding2")
public Binding routingQueueBinding2(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue2") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with("a");
}
@Bean("routingQueueBinding3")
public Binding routingQueueBinding3(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue2") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with("b");
}
@Bean("routingQueueBinding4")
public Binding routingQueueBinding4(@Qualifier("routingExchange") DirectExchange directExchange, @Qualifier("routingQueue2") Queue queue) {
return BindingBuilder.bind(queue).to(directExchange).with("c");
}
在绑定交换机时,需要调用 with 方法指定 BindingKey
之后,就可以发送消息了
生产者代码
在 ProducerController 中使用 rabbitTemplate 发送消息
在 路由模式 下,需要指定 RoutingKey,为了测试不同 RoutingKey 发送消息的情况,我们在路径中指定 RoutingKey
@RequestMapping("/routing/{routingKey}")
public String direct(@PathVariable("routingKey") String routingKey) {
rabbitTemplate.convertAndSend(Constants.ROUTING_CHANGE, routingKey, "routing test... " +routingKey);
return "OK";
}
运行程序,访问 http://127.0.0.1:8080/producer/routing/a 、http://127.0.0.1:8080/producer/routing/b 和 http://127.0.0.1:8080/producer/routing/c
观察结果:
routing.queue.1 中只有 1 条消息,routing.queue.2 中有 3 条消息
只有当 routingKey 为 a 时,才能与 routing.queue.1 的 BindingKey 匹配,而 routingKey 为 a、b、c 时,都能与 routing.queue.2 的 BindingKey 相匹配,因此routing.queue.1 中只有 1 条消息,routing.queue.2 中有 3 条消息
我们查看 routing.queue.2 中的消息:
接着,我们继续实现消费者代码
消费者代码
消费者的代码与 发布订阅模式 的相同,因此我们只需要修改消费的队列即可:
@Component
public class RoutingListener {
@RabbitListener(queues = Constants.ROUTINT_QUEUE_1)
public void queueListener1(String message) {
System.out.println("listener1 从队列 " + Constants.ROUTINT_QUEUE_1 +" 中接收到消息:" + message);
}
@RabbitListener(queues = Constants.ROUTINT_QUEUE_2)
public void queueListener2(String message) {
System.out.println("listener2 从队列 " + Constants.ROUTINT_QUEUE_2 +" 中接收到消息:" + message);
}
}
再次运行,观察日志:
我们继续看 通配符模式
Topics(通配符模式)
相比于 routing 模式,topics 类型的交换机在匹配规则上进行了扩展,BindingKey 支持通配符匹配
RoutingKey 是一系列由 . 分割的单词,如 user.name、work.abc等
BindingKey 也和 RoutingKey 一样,由 . 分割的字符串
在 BindingKey 中可以存在两种特殊的字符串,用于模糊匹配:
* :表示能够匹配任意一个单词
#:表示能够匹配任意多个单词(可以为 0 个)
先在 Constants 中声明通配符模式下使用的队列和交换机:
// 通配符模式
public static final String TOPICS_CHANGE = "topics";
public static final String TOPICS_QUEUE_1 = "topics.queue.1";
public static final String TOPICS_QUEUE_2 = "topics.queue.2";
接着,在 RabbitMQConfig 中声明队列:
// 通配符模式
@Bean("topicsQueue1")
public Queue topicsQueue1() {
return QueueBuilder.durable(Constants.TOPICS_QUEUE_1).build();
}
@Bean("topicsQueue2")
public Queue topicsQueue2() {
return QueueBuilder.durable(Constants.TOPICS_QUEUE_2).build();
}
声明交换机:
注意,通配符模式下,交换机类型为 Topics(通配符)
在 Spring 中,使用 TopicExchange 表示通配符类型的交换机
@Bean("topicsExchange")
public TopicExchange topicExchange() {
return ExchangeBuilder.topicExchange(Constants.TOPICS_CHANGE).durable(true).build();
}
使用 ExchangeBuilder 的 topicExchange 方法创建通配符类型的交换机
绑定队列和交换机:
// 绑定交换机和队列
@Bean("topicsQueueBinding1")
public Binding topicsQueueBinding1(@Qualifier("topicsExchange") TopicExchange topicExchange, @Qualifier("topicsQueue1") Queue queue) {
return BindingBuilder.bind(queue).to(topicExchange).with("*.*.a");
}
@Bean("topicsQueueBinding2")
public Binding topicsQueueBinding2(@Qualifier("topicsExchange") TopicExchange topicExchange, @Qualifier("topicsQueue2") Queue queue) {
return BindingBuilder.bind(queue).to(topicExchange).with("*.b.*");
}
@Bean("topicsQueueBinding3")
public Binding topicsQueueBinding3(@Qualifier("topicsExchange") TopicExchange topicExchange, @Qualifier("topicsQueue2") Queue queue) {
return BindingBuilder.bind(queue).to(topicExchange).with("c.#");
}
调用 with 方法指定 BindingKey
接下来,就可以发送消息了
生产者代码
在 ProducerController 中使用 rabbitTemplate 发送消息
同样的,在 通配符模式 下,需要指定 RoutingKey,为了测试不同 RoutingKey 发送消息的情况,我们在路径中指定 RoutingKey
@RequestMapping("/topics/{topicsKey}")
public String topics(@PathVariable("topicsKey") String topicsKey) {
rabbitTemplate.convertAndSend(Constants.TOPICS_CHANGE, topicsKey, "topics test... " +topicsKey);
return "OK";
}
运行程序,并访问 http://127.0.0.1:8080/producer/topics/a.b.a 、http://127.0.0.1:8080/producer/topics/c.work 和 http://127.0.0.1:8080/producer/topics/a.a.a
观察结果:
topics.queue.1 和 topics.queue.2 中都有两条消息,我们先来看 topics.queue.1:
topics.queue.2 中的消息:
当 topicsKey为 a.b.a 时,能与 topics.queue.1 的 BindingKey(*.*.a) 匹配,也能与 topics.queue.2 的 BindingKey(*.b.*) 匹配,因此,消息会被路由到两个队列中
topicsKey 为 c.work 时,只能与 topics.queue.2 的 BindingKey(c.#) 相匹配
topicsKey 为 a.a.a 时,只能与 topics.queue.1 的 BindingKey(*.*.a) 相匹配
我们继续实现消费者代码
消费者代码
消费者的代码与 路由模式 的相同,因此我们只需要修改消费的队列即可:
@Component
public class TopicsListener {
@RabbitListener(queues = Constants.TOPICS_QUEUE_1)
public void queueListener1(String message) {
System.out.println("listener1 [" + Constants.TOPICS_QUEUE_1 +"] 接收到消息: " + message);
}
@RabbitListener(queues = Constants.TOPICS_QUEUE_2)
public void queueListener2(String message) {
System.out.println("listener2 [" + Constants.TOPICS_QUEUE_2 +"] 接收到消息: " + message);
}
}
再次运行,观察日志:
常见问题
在我们编写代码的过程中,可能会出现一些问题,接下来,我们就一起来看看常见的错误类型
交换机与队列的绑定
例如,在 通配符模式 下,交换机类型绑定为 Direct(定向)模式:
此时,若使用了 @Qualifier 注解,则会直接报错,显示类型不匹配
但是,若未使用 @Qualifier 注解,则不会报错,程序也能正常运行
我们观察交换机 routing:
可以看到 topics.queue.1 与 routing 交换机进行了绑定,我们再访问 , 消息成功发送
但 topics.queue.1 队列中始终没有消息,这是因为 topics.queue.1 此时并未与 topics 交换机进行绑定,topics 交换机在接收到消息后,发现没有匹配的 BindingKey,就直接将消息丢弃了
当消息发送成功,但队列中却没有消息时,就需要检查队列和交换机的绑定关系了
交换机类型错误
在我们定义交换机时,可能会一不小心将交换机的名称写错:
我们运行程序,观察日志:
我们来看 ERROR 信息显示在 虚拟机 test02 上的交换机 fanout 接收到的类型为 topic,但当前 fanout 是 fanout 类型的
fanout交换机 是 fanout 类型的,但是此时将其当做 topic 类型来使用,也就报错了
若一个交换机已经存在,则不能再修改其类型了,若我们需要进行修改,则需要将之前已经存在的交换机删除
队列属性错误
例如,topics.queue.1 是一个持久化的队列,但此时我们将其声明为 非持久化 的:
再次运行:
也显示了 ERROR 信息,提示 在虚拟机 test02 上的队列 topics.queue.1 是持久化的,接收到的参数是 false,但当前为 true
也就是说,我们尝试修改队列的持久化参数,此时是不被允许的
topics.queue.1 是持久化的:
D 表示 durable 为 true
若我们需要将其修改为非持久化的,需要先将已经存在的 topics.queue.1 队列删除:
此时再次运行程序,访问http://127.0.0.1:8080/producer/topics/a.a.a
观察结果:
此时,topics.queue.1 就是非持久化的了
当交换机、队列创建完成时,其属性是不能发生变化的,若需要修改,则需要将当前交换机或队列删除,然后重新声明
原文地址:https://blog.csdn.net/2301_76161469/article/details/143989679
免责声明:本站文章内容转载自网络资源,如本站内容侵犯了原著者的合法权益,可联系本站删除。更多内容请关注自学内容网(zxcms.com)!