springboot 整合rabbitmq 自定义消息监听容器来实现消息批量处理前言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文件中添加以下依赖：
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-amqp</artifactid></dependency>
配置文件接下来，在application.properties文件中添加以下配置：
spring.rabbitmq.host=localhostspring.rabbitmq.port=5672spring.rabbitmq.username=guestspring.rabbitmq.password=guestspring.rabbitmq.virtual-host=/# 队列名称spring.rabbitmq.listener.simple.queue-name=myqueue# 最大并发消费者数量spring.rabbitmq.listener.simple.concurrency=5# 最小数量spring.rabbitmq.listener.simple.min-concurrency=1# 最大数量spring.rabbitmq.listener.simple.max-concurrency=10# 批量处理消息的大小spring.rabbitmq.listener.simple.batch-size=50
或
spring: rabbitmq: host: localhost listener: simple: batch-size: 50 concurrency: 5 max-concurrency: 10 min-concurrency: 1 queue-name: myqueue password: guest port: 5672 username: guest virtual-host: /
编写监听器然后，我们需要创建一个监听器类，以便处理从队列中接收到的消息。以下是一个简单的示例：
@componentpublic class mylistener { @rabbitlistener(queues = "myqueue", containerfactory = "myfactory") public void handlemessage(list<mymessage> 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工厂，以便能够自定义监听容器的行为。以下是一个简单的示例：
@configurationpublic 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;// } @bean public simplerabbitlistenercontainerfactory myfactory( connectionfactory connectionfactory, platformtransactionmanager transactionmanager, messageconverter messageconverter) { simplerabbitlistenercontainerfactory factory = new simplerabbitlistenercontainerfactory(); // 并发消费者数，默认为 1 factory.setconcurrentconsumers(5); // 最大并发消费者数，默认为 1 factory.setmaxconcurrentconsumers(10); // 拒绝未确认的消息并重新将它们放回队列，默认为 true factory.setdefaultrequeuerejected(false); // 容器启动时是否自动启动，默认为 true factory.setautostartup(true); // 消息确认模式，默认为 auto factory.setacknowledgemode(acknowledgemode.manual); // 每个消费者在一次请求中预获取的消息数，默认为 1 factory.setprefetchcount(5); // 从队列中接收消息的超时时间，默认为 0，表示没有超时限制 factory.setreceivetimeout(1000); // 与容器一起使用的事务管理器。默认情况下，容器不会使用事务 factory.settransactionmanager(transactionmanager); // 消息转换器，用于将接收到的消息转换为 java 对象或将 java 对象转换为消息 factory.setmessageconverter(messageconverter); // 用于异步消息处理的线程池。默认情况下，容器使用一个简单的 simpleasynctaskexecutor factory.settaskexecutor(new simpleasynctaskexecutor()); // 在关闭容器时等待活动线程终止的时间，默认为 5000 毫秒 factory.setshutdowntimeout(10000); // 重试失败的消息之前等待的时间，默认为 5000 毫秒 factory.setrecoveryinterval(5000); // 如果消息处理器尝试监听不存在的队列，是否抛出异常。默认为 true factory.setmissingqueuesfatal(false); // 监听器容器连接工厂 factory.setconnectionfactory(connectionfactory); return factory; }}
这些属性中的大多数都是可选的，可以根据需要进行设置。根据应用程序的需求，我们可以自由地调整这些属性，以提高应用程序的性能和可靠性。
发送消息最后，我们可以编写一个简单的发送消息的代码来向队列中发送一些消息。以下是一个简单的示例：
@componentpublic class mysender { @autowired private rabbittemplate rabbittemplate; public void sendmessage(string message) { for (int i = 0; i < 100; i++) { rabbittemplate.convertandsend("myqueue", "message：" + i); } }}
以上就是springboot怎么整合rabbitmq自定义消息监听容器来实现消息批量处理的详细内容。