东阳高端营销型网站建设品牌,茌平网站开发,wordpress 建站插件,常见的网络营销平台有哪些SpringBoot 整合RabbitMq 自定义消息监听容器来实现消息批量处理前言添加依赖配置文件编写监听器创建SimpleRabbitListenerContainerFactory发送消息前言
RabbitMQ是一种常用的消息队列#xff0c;Spring Boot对其进行了深度的整合#xff0c;可以快速地实现消息的发送和接收…
SpringBoot 整合RabbitMq 自定义消息监听容器来实现消息批量处理前言添加依赖配置文件编写监听器创建SimpleRabbitListenerContainerFactory发送消息前言
RabbitMQ是一种常用的消息队列Spring Boot对其进行了深度的整合可以快速地实现消息的发送和接收。在RabbitMQ中消息的发送和接收都是异步的因此需要使用监听器来监听消息的到来。Spring Boot中提供了默认的监听器容器但是有时候我们需要自定义监听器容器来满足一些特殊的需求比如批量获取数据。
在本文中我们将使用Spring Boot来整合RabbitMQ并自定义一个监听器容器实现批量获取数据的功能。 前置条件 在开始之前您需要具备以下条件
已经安装好RabbitMQ服务器并启动。已经创建好要使用的队列。已经熟悉了Spring Boot和RabbitMQ的基本知识。
环境准备 在开始之前我们需要准备好以下环境
JDK 1.8或以上版本Spring Boot 2.5.0或以上版本RabbitMQ 3.8.0或以上版本
添加依赖
首先在pom.xml文件中添加以下依赖
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency配置文件
接下来在application.properties文件中添加以下配置
spring.rabbitmq.hostlocalhost
spring.rabbitmq.port5672
spring.rabbitmq.usernameguest
spring.rabbitmq.passwordguest
spring.rabbitmq.virtual-host/# 队列名称
spring.rabbitmq.listener.simple.queue-namemyQueue# 最大并发消费者数量
spring.rabbitmq.listener.simple.concurrency5# 最小数量
spring.rabbitmq.listener.simple.min-concurrency1# 最大数量
spring.rabbitmq.listener.simple.max-concurrency10# 批量处理消息的大小
spring.rabbitmq.listener.simple.batch-size50或
spring:rabbitmq:host: localhostlistener:simple:batch-size: 50concurrency: 5max-concurrency: 10min-concurrency: 1queue-name: myQueuepassword: guestport: 5672username: guestvirtual-host: /编写监听器
然后我们需要创建一个监听器类以便处理从队列中接收到的消息。以下是一个简单的示例
Component
public class MyListener {RabbitListener(queues myQueue, containerFactory myFactory)public void handleMessage(ListMyMessage messages, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag)throws IOException {try {// 处理消息System.out.println(Received messages.size() messages);for (Message message : messages) {// 处理消息System.out.println(Received message: new String(message.getBody()));}channel.basicAck(messages.get(messages.size() - 1).getMessageProperties().getDeliveryTag(), true);} finally {// 手动确认消息channel.basicAck(deliveryTag, true);}}
}在上面的代码中我们使用了RabbitListener注解来指定要监听的队列名称同时也指定了使用myFactory工厂来创建监听容器。在这个监听器中我们简单地打印了接收到的消息。
创建SimpleRabbitListenerContainerFactory
接下来我们需要创建一个SimpleRabbitListenerContainerFactory工厂以便能够自定义监听容器的行为。以下是一个简单的示例
Configuration
public class RabbitMQConfig {// Bean
// public SimpleRabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory) {
// SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory();
// factory.setConnectionFactory(connectionFactory);
// factory.setConcurrentConsumers(1);
// factory.setMaxConcurrentConsumers(10);
// factory.setBatchListener(true);
// factory.setBatchSize(50);
// return factory;
// }Beanpublic SimpleRabbitListenerContainerFactory myFactory(ConnectionFactory connectionFactory,PlatformTransactionManager transactionManager,MessageConverter messageConverter) {SimpleRabbitListenerContainerFactory factory new SimpleRabbitListenerContainerFactory();// 并发消费者数默认为 1factory.setConcurrentConsumers(5);// 最大并发消费者数默认为 1factory.setMaxConcurrentConsumers(10);// 拒绝未确认的消息并重新将它们放回队列默认为 truefactory.setDefaultRequeueRejected(false);// 容器启动时是否自动启动默认为 truefactory.setAutoStartup(true);// 消息确认模式默认为 AUTOfactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 每个消费者在一次请求中预获取的消息数默认为 1factory.setPrefetchCount(5);// 从队列中接收消息的超时时间默认为 0表示没有超时限制factory.setReceiveTimeout(1000);// 与容器一起使用的事务管理器。默认情况下容器不会使用事务factory.setTransactionManager(transactionManager);// 消息转换器用于将接收到的消息转换为 Java 对象或将 Java 对象转换为消息factory.setMessageConverter(messageConverter);// 用于异步消息处理的线程池。默认情况下容器使用一个简单的 SimpleAsyncTaskExecutorfactory.setTaskExecutor(new SimpleAsyncTaskExecutor());// 在关闭容器时等待活动线程终止的时间默认为 5000 毫秒factory.setShutdownTimeout(10000);// 重试失败的消息之前等待的时间默认为 5000 毫秒factory.setRecoveryInterval(5000);// 如果消息处理器尝试监听不存在的队列是否抛出异常。默认为 truefactory.setMissingQueuesFatal(false);// 监听器容器连接工厂factory.setConnectionFactory(connectionFactory);return factory;}
}这些属性中的大多数都是可选的可以根据需要进行设置。根据应用程序的需求我们可以自由地调整这些属性以提高应用程序的性能和可靠性。
发送消息
最后我们可以编写一个简单的发送消息的代码来向队列中发送一些消息。以下是一个简单的示例
Component
public class MySender {Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMessage(String message) {for (int i 0; i 100; i) {rabbitTemplate.convertAndSend(myQueue, message i);}}
}总之 SimpleRabbitListenerContainerFactory 提供了一个灵活而强大的机制用于自定义和配置 RabbitMQ 消息监听容器。
