RabbitMQ实现消息发送接收——实战篇(路由模式)

本篇博文将带领大家一起学习rabbitMQ如何进行消息发送接收，我也是在写项目的时候边学边写，有不足的地方希望在评论区留下你的建议，我们一起讨论学习呀~

需求背景

先说一下我的项目需求背景，社区之间可以进行物资借用，当有社区提交物资借用申请时，需要通过RabbitMQ将这条消息发送到被借用物资的社区，同时在界面进行提示。

先把依赖引入一下

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

application.yml做好配置：

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guest

工具类实现

先选择以何种方式进行消息发送，这里根据需求我选择使用RabbitMQ的路由模式进行消息发送，先来配置一下相应工具类：

先配置RabbitMQ的配置类

/*** @Title: RabbitMQConfig* @Author yinan* @Package com.yinan.config.RabbitConfig* @Date 2024/12/13 13:58* @description: RabbitMQ配置类*/
@Configuration
public class RabbitMQConfig {@Beanpublic RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {return new RabbitAdmin(connectionFactory);}//    声明一个交换机@Beanpublic DirectExchange borrowMaterialExchange(){return new DirectExchange("borrow_material_exchange");}//    动态绑定队列时使用的方法（具体绑定逻辑在下面的监听器中实现）
//    @Bean
//    public Queue communityQueue(){
//        return new Queue("communityQueue");
//    }@Beanpublic Jackson2JsonMessageConverter messageConverter() {return new Jackson2JsonMessageConverter();}@Beanpublic AmqpTemplate amqpTemplate(ConnectionFactory connectionFactory) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMessageConverter(messageConverter());return rabbitTemplate;}
}

为了确保消息发送和接收时都以 JSON 格式处理，可以在 Spring 配置中添加 Jackson2JsonMessageConverter。这样，发送端会将 MaterialBorrowing 对象序列化为 JSON，接收端会自动将 JSON 反序列化回 MaterialBorrowing 对象。

绑定交换机和对应队列

/*** @Title: RabbitMQBindRoutingConfig* @Author yinan* @Package com.yinan.config.RabbitConfig* @Date 2024/12/13 14:21* @description:  动态绑定路由配置*/
@Component
@Slf4j
public class RabbitMQBindRoutingConfig {@Autowiredprivate DirectExchange borrowMaterialExchange;@Autowiredprivate RabbitAdmin rabbitAdmin;/*** 以社区ID为路由键，为指定社区动态创建队列并绑定到交换机* @param communityId 社区ID*/public void bindRouting(String communityId){
//        创建队列Queue queue = new Queue("queue_" + communityId);
//        动态绑定交换机和指定队列Binding binding = BindingBuilder.bind(queue).to(borrowMaterialExchange).with(communityId);rabbitAdmin.declareExchange(borrowMaterialExchange);rabbitAdmin.declareBinding(binding);log.info("队列绑定成功，社区ID----》" + communityId + "，队列名称----》" + queue.getName() + "，交换机名称----》" + borrowMaterialExchange.getName());}}

动态声明队列

@Configuration
@Slf4j
public class QueueDeclareConfig {@Autowiredprivate RabbitAdmin rabbitAdmin;public void dynamicDeclareQueue(String communityId){String queueName = String.format("queue_%s",communityId);Queue queue = new Queue(queueName,true);rabbitAdmin.declareQueue(queue);log.info("队列声明成功");}
}

在创建声明队列的时候，我们希望的是根据我们的规则在调用接口的时候去创建指定名称的队列，所以可以使用动态声明对列而不是直接在平台上进行配置。

消息发送

@Component
@Slf4j
public class MessageSendConfig {@Autowiredprivate AmqpTemplate amqpTemplate;public void sendMessage(Object message,String communityId){System.out.println("发送消息:" + message);amqpTemplate.convertAndSend("borrow_material_exchange",communityId, message);log.info("发送消息成功------->"+message);}}

消息接收(动态声明与监听结合),这里你可以先思考一下为什么要用这种方式实现消息接收，而不是使用@RabbitListener去动态获取某个队列接收消息。

/*** @Title: MessageRecieveConfig* @Author yinan* @Package com.yinan.config.RabbitConfig* @Date 2024/12/13 12:53* @description: 动态监听接收消息*/
@Service
@Slf4j
public class MessageRecieveConfig<T> {private final ConnectionFactory connectionFactory;public MessageRecieveConfig(ConnectionFactory connection) {this.connectionFactory = connection;}public void recieveMessage(String communityId,Class<T> objectType){String queueName = String.format("queue_%s",communityId);
//        创建监听容器SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);container.setQueueNames(queueName);
//        处理消息消费逻辑container.setMessageListener(message -> {try {// 将字节数组转换为字符串String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);System.out.println("接收到的消息：" + messageBody);// 如果需要将消息解析为对象（例如 MaterialBorrowing）ObjectMapper objectMapper = new ObjectMapper();T result = objectMapper.readValue(messageBody, objectType);System.out.println("反序列化后的消息：" + result);} catch (Exception e) {log.error("处理消息时发生错误：", e);}});// 确保自动确认container.setAcknowledgeMode(AcknowledgeMode.AUTO);container.start();log.info("动态监听已启动，监听队列------->"+queueName);}}

在你的项目中分别调用就行了，需要注意的是你必须确保在消息发送的时候你的队列已经创建完成且和对应交换机进行了绑定，不然可能会导致消息发送失败。

ok，我们启动项目

你会发现你的项目根本启动不起来，原因是因为对于 Spring AMQP 的监听器来说，必须确保监听的队列已经存在于 RabbitMQ 中，否则会抛出类似 DeclarationException 的错误。

所以我们考虑可以通过动态声明队列，在程序运行时确保 RabbitMQ 上创建好所需的队列。

动态声明队列的含义

动态声明队列是指程序在运行时，通过代码检查或创建 RabbitMQ 中尚不存在的队列，而不是手动预先配置好所有队列。这种方式可以自动帮你在 RabbitMQ 中创建所需的队列，而无需手动操作。

这里说一下为什么需要动态绑定队列而不直接使用@RabbitListener？

为什么使用 `SimpleMessageListenerContainer` 动态绑定队列

SimpleMessageListenerContainer 不需要在项目启动时绑定队列。你可以在用户调用接口时动态创建队列，并动态监听它。
特点：
- 队列在用户调用接口时才会被动态创建（通过 RabbitAdmin 或其他机制）。
- 动态创建队列和监听时，项目启动时不会尝试绑定不存在的队列，因此不会报错。
适用场景：非常适合动态队列需求，比如队列名依赖用户输入或业务逻辑，且不想在项目启动时绑定固定的队列。

使用 @RabbitListener 的情况

@RabbitListener 会在项目启动时绑定到指定的队列。

要求：如果绑定的队列在 RabbitMQ 中不存在，项目启动时就会抛出异常，类似 DeclarationException，这也就是上面为什么会报错的原因。
解决办法：
提前创建队列：在 RabbitMQ 中手动创建队列，或通过 RabbitAdmin 在项目启动时自动创建队列。
动态队列名：如果队列名是动态的，可以结合 SpEL 表达式，但队列仍然需要在项目启动时确保存在。

SpEL 表达式

如果你的需求中已经确定队列已经创建好的，但是需要动态去获取队列，可以使用如下形式：

@RabbitListener(queues = "#{T(java.lang.String).format('queue_%s', 'borrowedCommunityId')}")

这个表达式是 Spring AMQP 中用于动态指定队列名称的 SpEL 表达式（Spring Expression Language），它的作用就是会动态生成一个队列名称，基于你传入的参数构造队列名。

详解

1. 关键部分解析

T(java.lang.String)：

T 是 SpEL 用于引用 Java 类 的方式。
java.lang.String 是目标 Java 类，表明你可以调用 String 类的静态方法。

.format()：

String.format() 是 Java 中的静态方法，用于格式化字符串。
格式化字符串的格式是 'queue_%s'，%s 是占位符，用于拼接动态内容。

'queue_%s'：

这是格式化字符串的模板。%s 表示字符串占位符。

动态参数（例如 borrowedCommunityId）：

它会替换 %s，生成队列名。例如，当 borrowedCommunityId 的值是 123 时，结果是：queue_123。

2. 具体实例

假设 borrowedCommunityId = "123"：
String result = String.format("queue_%s", "123");
System.out.println(result); // 输出：queue_123
在 SpEL 中，这等同于：
queues = "#{T(java.lang.String).format('queue_%s', '123')}"
这会动态生成队列名称为 queue_123。

为什么用 SpEL？

Spring AMQP 的 @RabbitListener 注解中，queues 参数支持 SpEL 表达式。这使得我们可以动态决定要监听的队列，而不是写死某个固定的队列名称。

实际应用场景

就比如在我的代码中，可能有多个社区队列，例如：

queue_123（社区 ID 为 123 的队列）
queue_456（社区 ID 为 456 的队列）

使用 queues = "#{T(java.lang.String).format('queue_%s', borrowedCommunityId)}"，可以动态生成不同社区的队列名称，从而实现按社区路由的功能。