文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

SpringBoot怎么整合RabbitMq自定义消息监听容器来实现消息批量处理

2023-07-05 21:11

关注

这篇文章主要介绍“SpringBoot怎么整合RabbitMq自定义消息监听容器来实现消息批量处理”的相关知识,小编通过实际案例向大家展示操作过程,操作方法简单快捷,实用性强,希望这篇“SpringBoot怎么整合RabbitMq自定义消息监听容器来实现消息批量处理”文章能帮助大家解决问题。

SpringBoot 整合RabbitMq 自定义消息监听容器来实现消息批量处理

前言

RabbitMQ是一种常用的消息队列,Spring Boot对其进行了深度的整合,可以快速地实现消息的发送和接收。在RabbitMQ中,消息的发送和接收都是异步的,因此需要使用监听器来监听消息的到来。Spring Boot中提供了默认的监听器容器,但是有时候我们需要自定义监听器容器,来满足一些特殊的需求,比如批量获取数据。

在本文中,我们将使用Spring Boot来整合RabbitMQ,并自定义一个监听器容器,实现批量获取数据的功能。
前置条件:
在开始之前,您需要具备以下条件:

环境准备:
在开始之前,我们需要准备好以下环境:

添加依赖

首先,在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自定义消息监听容器来实现消息批量处理”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识,可以关注编程网行业资讯频道,小编每天都会为大家更新不同的知识点。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯