当前位置：首页 > news >正文

RabbitMq(五) -- SpringBoot整合 RabbitMQ 完整实现

news 2026/7/31 9:22:43

1. 概述

RabbitMQ 是基于 AMQP 协议的消息中间件，在分布式系统中常用于解耦、异步处理、流量削峰等场景。SpringBoot 通过 spring-boot-starter-amqp 提供了对 RabbitMQ 的深度整合，让开发者能够以声明式或编程式的方式轻松操作消息队列。

本文将详细讲解：

队列创建的两种方式：@Bean 方式与 AmqpAdmin 方式
消息接收的两种方式：@RabbitListener 与 @RabbitHandler

每个知识点均配有详细的文字解释和完整的代码示例，参数含义会逐一说明。

2. 环境准备

2.1 添加依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

2.2 配置连接信息

spring:rabbitmq:host:127.0.0.1# RabbitMQ 服务地址port:5672# 连接端口（注意不是管理界面端口 15672）username:guest# 用户名password:guest# 密码virtual-host:/# 虚拟主机，默认为 /# 可选：连接超时、重试等配置connection-timeout:5000publisher-confirm-type:correlated# 开启发送确认publisher-returns:true# 开启消息返回

3. 队列创建的两种方式

3.1 方式一：使用 @Bean 声明式创建

通过在 @Configuration 类中定义 Queue、Exchange、Binding 的 Bean，Spring 容器启动时会自动在 RabbitMQ 服务端创建这些组件。

3.1.1 基本示例

importorg.springframework.amqp.core.*;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassRabbitMQBeanConfig{// 1. 创建队列@BeanpublicQueuemyQueue(){// 参数详解：// name: 队列名称，必须唯一// durable: 是否持久化（true: 重启后队列依然存在）// exclusive: 是否独占（true: 仅当前连接可用，连接关闭后队列删除）// autoDelete: 是否自动删除（最后一个消费者取消后删除队列）// 还有另一种构造方式：new Queue(name, durable, exclusive, autoDelete, arguments)returnnewQueue("myQueue",true,false,false);}// 2. 创建交换机（以 DirectExchange 为例）@BeanpublicDirectExchangemyExchange(){// 参数详解：// name: 交换机名称// durable: 是否持久化// autoDelete: 是否自动删除// arguments: 扩展参数returnnewDirectExchange("myExchange",true,false);}// 3. 创建绑定关系@BeanpublicBindingmyBinding(){returnBindingBuilder.bind(myQueue())// 绑定哪个队列.to(myExchange())// 绑定到哪个交换机.with("myRoutingKey");// 路由键}}

3.1.2 带参数的队列（死信队列、TTL 等）

@BeanpublicQueuettlQueue(){Map<String,Object>args=newHashMap<>();// 设置队列中消息的 TTL（毫秒）args.put("x-message-ttl",60000);// 设置死信交换机args.put("x-dead-letter-exchange","deadLetterExchange");// 设置死信路由键args.put("x-dead-letter-routing-key","deadLetterKey");// 设置队列最大长度（条数）args.put("x-max-length",1000);// 设置队列最大容量（字节）args.put("x-max-length-bytes",10485760);returnnewQueue("ttlQueue",true,false,false,args);}

3.2 方式二：使用 AmqpAdmin 动态创建

AmqpAdmin 是 Spring 提供的用于管理 RabbitMQ 组件的接口，可以在运行时动态声明队列、交换机和绑定关系，适合需要根据业务逻辑动态创建临时队列的场景。

3.2.1 AmqpAdmin 常用方法

importorg.springframework.amqp.core.*;importorg.springframework.amqp.rabbit.core.RabbitAdmin;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;@ServicepublicclassDynamicQueueService{@AutowiredprivateAmqpAdminamqpAdmin;// 动态创建队列publicvoidcreateQueue(StringqueueName,booleandurable,booleanexclusive,booleanautoDelete){Queuequeue=newQueue(queueName,durable,exclusive,autoDelete);amqpAdmin.declareQueue(queue);}// 动态创建交换机publicvoidcreateExchange(StringexchangeName,Stringtype,booleandurable){AbstractExchangeexchange;switch(type){case"direct":exchange=newDirectExchange(exchangeName,durable,false);break;case"topic":exchange=newTopicExchange(exchangeName,durable,false);break;case"fanout":exchange=newFanoutExchange(exchangeName,durable,false);break;case"headers":exchange=newHeadersExchange(exchangeName,durable,false);break;default:thrownewIllegalArgumentException("不支持的类型: "+type);}amqpAdmin.declareExchange(exchange);}// 动态创建绑定publicvoidcreateBinding(StringqueueName,StringexchangeName,StringroutingKey){Bindingbinding=newBinding(queueName,Binding.DestinationType.QUEUE,exchangeName,routingKey,null// arguments);amqpAdmin.declareBinding(binding);}// 删除队列publicbooleandeleteQueue(StringqueueName){returnamqpAdmin.deleteQueue(queueName);}// 清空队列消息（不删除队列）publicvoidpurgeQueue(StringqueueName){amqpAdmin.purgeQueue(queueName,false);}}

3.2.2 动态创建带参数的队列

publicvoidcreateQueueWithArgs(StringqueueName){Map<String,Object>args=newHashMap<>();args.put("x-message-ttl",30000);// 30秒过期args.put("x-max-priority",10);// 支持优先级Queuequeue=newQueue(queueName,true,false,false,args);amqpAdmin.declareQueue(queue);}

3.3 两种方式的对比：

4. 消息发送（RabbitTemplate）

在讲解消息接收之前，先了解如何发送消息，以便测试接收功能。

importorg.springframework.amqp.rabbit.core.RabbitTemplate;importorg.springframework.beans.factory.annotation.Autowired;importorg.springframework.stereotype.Service;@ServicepublicclassMessageSender{@AutowiredprivateRabbitTemplaterabbitTemplate;// 简单发送字符串publicvoidsend(Stringexchange,StringroutingKey,Stringmessage){rabbitTemplate.convertAndSend(exchange,routingKey,message);}// 发送带消息属性的消息publicvoidsendWithProperties(Stringexchange,StringroutingKey,Stringmessage){MessagePropertiesproperties=newMessageProperties();properties.setContentType("text/plain");properties.setPriority(5);// 优先级properties.setExpiration("60000");// 消息过期时间（毫秒）properties.setHeader("customKey","customValue");Messagemsg=MessageBuilder.withBody(message.getBytes(StandardCharsets.UTF_8)).andProperties(properties).build();rabbitTemplate.send(exchange,routingKey,msg);}// 发送 Java 对象（需要配置 Jackson 转换器）publicvoidsendObject(Stringexchange,StringroutingKey,Objectobj){rabbitTemplate.convertAndSend(exchange,routingKey,obj);}}

5. 消息接收的两种方式

5.1 方式一：@RabbitListener 注解

@RabbitListener 可以标注在方法或类上，用于声明一个消息监听容器，自动监听指定队列。

5.1.1 基础用法

importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassBasicListener{// 监听单个队列，自动 ACK（默认）@RabbitListener(queues="myQueue")publicvoidlistenSimple(Stringmessage){System.out.println("收到消息: "+message);}// 监听多个队列@RabbitListener(queues={"queueA","queueB"})publicvoidlistenMultiple(Stringmessage){System.out.println("从多个队列收到: "+message);}// 使用路由键动态创建队列（需要交换机已存在）@RabbitListener(bindings=@QueueBinding(value=@Queue(value="autoQueue",durable="true"),exchange=@Exchange(value="myExchange",type=ExchangeTypes.DIRECT),key="myRoutingKey"))publicvoidlistenWithBinding(Stringmessage){System.out.println("通过 @QueueBinding 自动创建: "+message);}}

5.1.2 @RabbitListener 常用参数详解

5.1.3 手动确认消息（ACK）

默认情况下，消息消费成功后会自动 ACK。若需要手动确认，需配置 acknowledge-mode 为 MANUAL，并在监听方法中引入 Channel 和 deliveryTag。

importcom.rabbitmq.client.Channel;importorg.springframework.amqp.core.Message;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@ComponentpublicclassManualAckListener{@RabbitListener(queues="myQueue",containerFactory="manualAckContainerFactory")publicvoidlistenManual(Messagemessage,Channelchannel)throwsException{longdeliveryTag=message.getMessageProperties().getDeliveryTag();try{// 业务处理System.out.println("处理消息: "+newString(message.getBody()));// 手动确认（单条）channel.basicAck(deliveryTag,false);}catch(Exceptione){// 拒绝并重新入队（第二个参数 true 表示重新入队）channel.basicNack(deliveryTag,false,true);}}}

需要配置 containerFactory：

@BeanpublicRabbitListenerContainerFactory<?>manualAckContainerFactory(ConnectionFactoryconnectionFactory){SimpleRabbitListenerContainerFactoryfactory=newSimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手动确认returnfactory;}

5.2 方式二：@RabbitListener + @RabbitHandler 组合

当一个队列中包含多种类型（或格式）的消息时，可以在类上标注 @RabbitListener，然后在多个方法上标注 @RabbitHandler，根据消息类型自动路由到对应方法。

5.2.1 根据消息类型分发

importorg.springframework.amqp.rabbit.annotation.RabbitHandler;importorg.springframework.amqp.rabbit.annotation.RabbitListener;importorg.springframework.stereotype.Component;@Component@RabbitListener(queues="multiTypeQueue")publicclassMultiTypeListener{// 处理 String 类型@RabbitHandlerpublicvoidhandleString(Stringmessage){System.out.println("字符串消息: "+message);}// 处理 Integer 类型@RabbitHandlerpublicvoidhandleInteger(Integernumber){System.out.println("整数消息: "+number);}// 处理自定义 User 对象（需要配置 Jackson 转换器）@RabbitHandlerpublicvoidhandleUser(Useruser){System.out.println("用户对象: "+user.getName()+", "+user.getAge());}// 处理 byte[] 类型（原始消息体）@RabbitHandlerpublicvoidhandleBytes(byte[]bytes){System.out.println("字节消息: "+newString(bytes));}// 处理 Message 原始对象（包含属性）@RabbitHandlerpublicvoidhandleRawMessage(Messagemessage){Stringbody=newString(message.getBody());StringcontentType=message.getMessageProperties().getContentType();System.out.println("原始消息 - 类型: "+contentType+", 内容: "+body);}}

5.2.2 @RabbitHandler 方法参数匹配规则

根据方法参数的 Java 类型进行匹配（需要 MessageConverter 能将消息体转换成对应类型）
如果存在多个匹配的方法，会按照最具体的类型进行选择
可以通过 @Payload 和 @Headers 注解精确控制

@RabbitHandlerpublicvoidhandleWithAnnotations(@PayloadUseruser,@HeadersMap<String,Object>headers){System.out.println("Payload: "+user);System.out.println("Headers: "+headers);}

5.2.3 配置 Jackson 消息转换器（支持对象）

importorg.springframework.amqp.support.converter.Jackson2JsonMessageConverter;importorg.springframework.amqp.support.converter.MessageConverter;importorg.springframework.context.annotation.Bean;importorg.springframework.context.annotation.Configuration;@ConfigurationpublicclassRabbitMQConverterConfig{@BeanpublicMessageConverterjacksonMessageConverter(){returnnewJackson2JsonMessageConverter();}}

配置后，发送对象时会自动转为 JSON，接收时会自动反序列化为对应的 Java 类型。

6. 完整示例：综合运用

6.1 配置类（使用 @Bean 创建队列）

@ConfigurationpublicclassFullConfig{@BeanpublicQueueorderQueue(){returnnewQueue("order.queue",true);}@BeanpublicDirectExchangeorderExchange(){returnnewDirectExchange("order.exchange",true,false);}@BeanpublicBindingorderBinding(){returnBindingBuilder.bind(orderQueue()).to(orderExchange()).with("order.create");}}

6.2 发送者

@ServicepublicclassOrderSender{@AutowiredprivateRabbitTemplaterabbitTemplate;publicvoidsendOrder(Orderorder){rabbitTemplate.convertAndSend("order.exchange","order.create",order);}}

6.3 接收者（混合使用两种接收方式）

@ComponentpublicclassOrderListener{// 方式一：简单接收@RabbitListener(queues="order.queue")publicvoidhandleOrder(Orderorder){System.out.println("收到订单: "+order.getOrderId());}// 方式二：多类型处理器@Component@RabbitListener(queues="order.queue")publicstaticclassMultiHandler{@RabbitHandlerpublicvoidonOrder(Orderorder){System.out.println("Handler 处理订单: "+order.getOrderId());}@RabbitHandlerpublicvoidonString(Stringstr){System.out.println("意外收到字符串: "+str);}}}

7. 常见问题与注意事项

队列是否存在：使用 @RabbitListener 时，若队列不存在且没有 bindings 自动创建，启动会报错。建议使用 @Bean 预先声明或 bindings 属性。
消息转换器：如果发送和接收使用不同的消息转换器，可能导致类型转换异常。全局统一配置 MessageConverter 为 Jackson2JsonMessageConverter 可避免大部分问题。
并发消费：concurrency 参数可以增加消费者实例，提高吞吐量，但要注意业务幂等性。
消息确认：生产环境建议使用手动 ACK，避免业务异常导致消息丢失。
死信队列：务必为重要业务队列配置死信交换机，防止消息堆积或异常丢失后无法追溯。