SpringBoot?整合RabbitMq?自定義消息監(jiān)聽容器來實(shí)現(xiàn)消息批量處理
SpringBoot 整合RabbitMq 自定義消息監(jiān)聽容器來實(shí)現(xiàn)消息批量處理
前言
RabbitMQ是一種常用的消息隊(duì)列,Spring Boot對其進(jìn)行了深度的整合,可以快速地實(shí)現(xiàn)消息的發(fā)送和接收。在RabbitMQ中,消息的發(fā)送和接收都是異步的,因此需要使用監(jiān)聽器來監(jiān)聽消息的到來。Spring Boot中提供了默認(rèn)的監(jiān)聽器容器,但是有時候我們需要自定義監(jiān)聽器容器,來滿足一些特殊的需求,比如批量獲取數(shù)據(jù)。

在本文中,我們將使用Spring Boot來整合RabbitMQ,并自定義一個監(jiān)聽器容器,實(shí)現(xiàn)批量獲取數(shù)據(jù)的功能。
前置條件:
在開始之前,您需要具備以下條件:
- 已經(jīng)安裝好RabbitMQ服務(wù)器并啟動。
- 已經(jīng)創(chuàng)建好要使用的隊(duì)列。
- 已經(jīng)熟悉了Spring Boot和RabbitMQ的基本知識。
環(huán)境準(zhǔn)備:
在開始之前,我們需要準(zhǔn)備好以下環(huán)境:
- 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=localhost spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.virtual-host=/ # 隊(duì)列名稱 spring.rabbitmq.listener.simple.queue-name=myQueue # 最大并發(fā)消費(fèi)者數(shù)量 spring.rabbitmq.listener.simple.concurrency=5 # 最小數(shù)量 spring.rabbitmq.listener.simple.min-concurrency=1 # 最大數(shù)量 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: /編寫監(jiān)聽器
然后,我們需要創(chuàng)建一個監(jiān)聽器類,以便處理從隊(duì)列中接收到的消息。以下是一個簡單的示例:
@Component
public 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 {
// 手動確認(rèn)消息
channel.basicAck(deliveryTag, true);
}
}
}在上面的代碼中,我們使用了@RabbitListener注解來指定要監(jiān)聽的隊(duì)列名稱,同時也指定了使用myFactory工廠來創(chuàng)建監(jiān)聽容器。在這個監(jiān)聽器中,我們簡單地打印了接收到的消息。
創(chuàng)建SimpleRabbitListenerContainerFactory
接下來,我們需要創(chuàng)建一個SimpleRabbitListenerContainerFactory工廠,以便能夠自定義監(jiān)聽容器的行為。以下是一個簡單的示例:
@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;
// }
@Bean
public SimpleRabbitListenerContainerFactory myFactory(
ConnectionFactory connectionFactory,
PlatformTransactionManager transactionManager,
MessageConverter messageConverter) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
// 并發(fā)消費(fèi)者數(shù),默認(rèn)為 1
factory.setConcurrentConsumers(5);
// 最大并發(fā)消費(fèi)者數(shù),默認(rèn)為 1
factory.setMaxConcurrentConsumers(10);
// 拒絕未確認(rèn)的消息并重新將它們放回隊(duì)列,默認(rèn)為 true
factory.setDefaultRequeueRejected(false);
// 容器啟動時是否自動啟動,默認(rèn)為 true
factory.setAutoStartup(true);
// 消息確認(rèn)模式,默認(rèn)為 AUTO
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
// 每個消費(fèi)者在一次請求中預(yù)獲取的消息數(shù),默認(rèn)為 1
factory.setPrefetchCount(5);
// 從隊(duì)列中接收消息的超時時間,默認(rèn)為 0,表示沒有超時限制
factory.setReceiveTimeout(1000);
// 與容器一起使用的事務(wù)管理器。默認(rèn)情況下,容器不會使用事務(wù)
factory.setTransactionManager(transactionManager);
// 消息轉(zhuǎn)換器,用于將接收到的消息轉(zhuǎn)換為 Java 對象或?qū)?Java 對象轉(zhuǎn)換為消息
factory.setMessageConverter(messageConverter);
// 用于異步消息處理的線程池。默認(rèn)情況下,容器使用一個簡單的 SimpleAsyncTaskExecutor
factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
// 在關(guān)閉容器時等待活動線程終止的時間,默認(rèn)為 5000 毫秒
factory.setShutdownTimeout(10000);
// 重試失敗的消息之前等待的時間,默認(rèn)為 5000 毫秒
factory.setRecoveryInterval(5000);
// 如果消息處理器嘗試監(jiān)聽不存在的隊(duì)列,是否拋出異常。默認(rèn)為 true
factory.setMissingQueuesFatal(false);
// 監(jiān)聽器容器連接工廠
factory.setConnectionFactory(connectionFactory);
return factory;
}
}
這些屬性中的大多數(shù)都是可選的,可以根據(jù)需要進(jìn)行設(shè)置。根據(jù)應(yīng)用程序的需求,我們可以自由地調(diào)整這些屬性,以提高應(yīng)用程序的性能和可靠性。
發(fā)送消息
最后,我們可以編寫一個簡單的發(fā)送消息的代碼來向隊(duì)列中發(fā)送一些消息。以下是一個簡單的示例:
@Component
public class MySender {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMessage(String message) {
for (int i = 0; i < 100; i++) {
rabbitTemplate.convertAndSend("myQueue", "message:" + i);
}
}
}總之, SimpleRabbitListenerContainerFactory 提供了一個靈活而強(qiáng)大的機(jī)制,用于自定義和配置 RabbitMQ 消息監(jiān)聽容器。
到此這篇關(guān)于SpringBoot 整合RabbitMq 自定義消息監(jiān)聽容器來實(shí)現(xiàn)消息批量處理的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMq消息批量處理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Java 進(jìn)階使用 Lambda 表達(dá)式實(shí)現(xiàn)超強(qiáng)的排序功能
今天要說的是第二種排序方式,在內(nèi)存中實(shí)現(xiàn)數(shù)據(jù)排序。這篇文章主要介紹了Java 進(jìn)階使用 Lambda 表達(dá)式實(shí)現(xiàn)超強(qiáng)的排序功能,需要的朋友可以參考下2021-11-11
SpringBoot RestTemplate 簡單包裝解析
這篇文章主要介紹了SpringBoot RestTemplate 簡單包裝解析,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2019-08-08
java實(shí)現(xiàn)人員信息管理系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)人員信息管理系統(tǒng),文中示例代碼介紹的非常詳細(xì),具有一定的參考價值,感興趣的小伙伴們可以參考一下2022-02-02

