主页 > 其他 >

SpringBoot中RabbitMQ的使用

其他
2025-09-15 11:24:01

引入依赖

添加配置

Simple（简单模式）

生产者代码

消费者代码

?编辑

Work Queue（工作队列）

生产者代码

消费者代码

Publish/Subscribe（发布/订阅）

生产者代码

消费者代码

Routing（路由模式）

生产者代码

消费者代码

Topics（通配符模式）

生产者代码

消费者代码

常见问题

交换机与队列的绑定

交换机类型错误

队列属性错误

在 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** 时，需要指定监听的队列

不要忘记添加@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，就直接将消息丢弃了

当消息发送成功，但队列中却没有消息时，就需要检查队列和交换机的绑定关系了

交换机类型错误