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

在本文中,我們將使用Spring Boot來(lái)整合RabbitMQ,并自定義一個(gè)監(jiān)聽(tīng)器容器,實(shí)現(xiàn)批量獲取數(shù)據(jù)的功能。
前置條件:
在開(kāi)始之前,您需要具備以下條件:
- 已經(jīng)安裝好RabbitMQ服務(wù)器并啟動(dòng)。
- 已經(jīng)創(chuàng)建好要使用的隊(duì)列。
- 已經(jīng)熟悉了Spring Boot和RabbitMQ的基本知識(shí)。
環(huán)境準(zhǔn)備:
在開(kāi)始之前,我們需要準(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>配置文件
接下來(lái),在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)聽(tīng)器
然后,我們需要?jiǎng)?chuàng)建一個(gè)監(jiān)聽(tīng)器類,以便處理從隊(duì)列中接收到的消息。以下是一個(gè)簡(jiǎn)單的示例:
@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 {
// 手動(dòng)確認(rèn)消息
channel.basicAck(deliveryTag, true);
}
}
}在上面的代碼中,我們使用了@RabbitListener注解來(lái)指定要監(jiān)聽(tīng)的隊(duì)列名稱,同時(shí)也指定了使用myFactory工廠來(lái)創(chuàng)建監(jiān)聽(tīng)容器。在這個(gè)監(jiān)聽(tīng)器中,我們簡(jiǎn)單地打印了接收到的消息。
創(chuàng)建SimpleRabbitListenerContainerFactory
接下來(lái),我們需要?jiǎng)?chuàng)建一個(gè)SimpleRabbitListenerContainerFactory工廠,以便能夠自定義監(jiān)聽(tīng)容器的行為。以下是一個(gè)簡(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);
// 容器啟動(dòng)時(shí)是否自動(dòng)啟動(dòng),默認(rèn)為 true
factory.setAutoStartup(true);
// 消息確認(rèn)模式,默認(rèn)為 AUTO
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
// 每個(gè)消費(fèi)者在一次請(qǐng)求中預(yù)獲取的消息數(shù),默認(rèn)為 1
factory.setPrefetchCount(5);
// 從隊(duì)列中接收消息的超時(shí)時(shí)間,默認(rèn)為 0,表示沒(méi)有超時(shí)限制
factory.setReceiveTimeout(1000);
// 與容器一起使用的事務(wù)管理器。默認(rèn)情況下,容器不會(huì)使用事務(wù)
factory.setTransactionManager(transactionManager);
// 消息轉(zhuǎn)換器,用于將接收到的消息轉(zhuǎn)換為 Java 對(duì)象或?qū)?Java 對(duì)象轉(zhuǎn)換為消息
factory.setMessageConverter(messageConverter);
// 用于異步消息處理的線程池。默認(rèn)情況下,容器使用一個(gè)簡(jiǎn)單的 SimpleAsyncTaskExecutor
factory.setTaskExecutor(new SimpleAsyncTaskExecutor());
// 在關(guān)閉容器時(shí)等待活動(dòng)線程終止的時(shí)間,默認(rèn)為 5000 毫秒
factory.setShutdownTimeout(10000);
// 重試失敗的消息之前等待的時(shí)間,默認(rèn)為 5000 毫秒
factory.setRecoveryInterval(5000);
// 如果消息處理器嘗試監(jiān)聽(tīng)不存在的隊(duì)列,是否拋出異常。默認(rèn)為 true
factory.setMissingQueuesFatal(false);
// 監(jiān)聽(tīng)器容器連接工廠
factory.setConnectionFactory(connectionFactory);
return factory;
}
}
這些屬性中的大多數(shù)都是可選的,可以根據(jù)需要進(jìn)行設(shè)置。根據(jù)應(yīng)用程序的需求,我們可以自由地調(diào)整這些屬性,以提高應(yīng)用程序的性能和可靠性。
發(fā)送消息
最后,我們可以編寫一個(gè)簡(jiǎn)單的發(fā)送消息的代碼來(lái)向隊(duì)列中發(fā)送一些消息。以下是一個(gè)簡(jiǎn)單的示例:
@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 提供了一個(gè)靈活而強(qiáng)大的機(jī)制,用于自定義和配置 RabbitMQ 消息監(jiān)聽(tīng)容器。
到此這篇關(guān)于SpringBoot 整合RabbitMq 自定義消息監(jiān)聽(tīng)容器來(lái)實(shí)現(xiàn)消息批量處理的文章就介紹到這了,更多相關(guān)SpringBoot RabbitMq消息批量處理內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Java基礎(chǔ)鞏固小項(xiàng)目點(diǎn)菜系統(tǒng)的實(shí)現(xiàn)
這篇文章主要介紹了一個(gè)Java小項(xiàng)目點(diǎn)菜系統(tǒng)的實(shí)現(xiàn),主要是用的集合,適合正在學(xué)習(xí)Java的朋友拿來(lái)實(shí)戰(zhàn)練手,感興趣的朋友快來(lái)看看吧2022-03-03
淺談SpringBoot中的@Conditional注解的使用
這篇文章主要介紹了淺談SpringBoot中的@Conditional注解的使用,小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2019-04-04
Java實(shí)現(xiàn)優(yōu)雅的參數(shù)校驗(yàn)方法詳解
這篇文章主要為大家詳細(xì)介紹了Java語(yǔ)言如何實(shí)現(xiàn)優(yōu)雅的參數(shù)校驗(yàn),文中的示例代碼講解詳細(xì),對(duì)我們學(xué)習(xí)Java有一定是幫助,需要的可以參考一下2022-06-06
SpringBoot關(guān)閉druid的頁(yè)面和添加密碼驗(yàn)證方式
這篇文章主要介紹了SpringBoot關(guān)閉druid的頁(yè)面和添加密碼驗(yàn)證方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-05-05
Java多線程編程之讀寫鎖ReadWriteLock用法實(shí)例
這篇文章主要介紹了Java多線程編程之讀寫鎖ReadWriteLock用法實(shí)例,本文直接給出編碼實(shí)例,需要的朋友可以參考下2015-05-05
java8中parallelStream性能測(cè)試及結(jié)果分析
本篇文章給大家用代碼實(shí)例做了segmentfaultjava8中parallelStream性能測(cè)試,并對(duì)測(cè)試結(jié)果做了說(shuō)明,需要的朋友學(xué)習(xí)下吧。2018-01-01
Spring Security單項(xiàng)目權(quán)限設(shè)計(jì)過(guò)程解析
這篇文章主要介紹了Spring Security單項(xiàng)目權(quán)限設(shè)計(jì)過(guò)程解析,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11
解決使用httpclient傳遞json數(shù)據(jù)亂碼的問(wèn)題
這篇文章主要介紹了解決使用httpclient傳遞json數(shù)據(jù)亂碼的問(wèn)題,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-01-01
Mybatis多個(gè)字段模糊匹配同一個(gè)值的案例
這篇文章主要介紹了Mybatis多個(gè)字段模糊匹配同一個(gè)值的案例,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-09-09

