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

Posted ℳ₯㎕ddzོꦿ࿐

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了SpringBoot 整合RabbitMq 自定义消息监听容器来实现消息批量处理相关的知识,希望对你有一定的参考价值。

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=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.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: /

编写监听器

然后,我们需要创建一个监听器类,以便处理从队列中接收到的消息。以下是一个简单的示例:

@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 
            // 手动确认消息
            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;
//    

	@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;
    

这些属性中的大多数都是可选的,可以根据需要进行设置。根据应用程序的需求,我们可以自由地调整这些属性,以提高应用程序的性能和可靠性。

发送消息

最后,我们可以编写一个简单的发送消息的代码来向队列中发送一些消息。以下是一个简单的示例:

@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 提供了一个灵活而强大的机制,用于自定义和配置 RabbitMQ 消息监听容器。

SpringBoot2.x整合Redis缓存自定义序列化

一、导入Jar包

<!--rabbitmq-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.49</version>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
        </dependency>   

 

二、配置文件

package com.liangjian.config;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.time.Duration;

/**
 * reids配置类
 */

@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) 
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(factory);

        // 使用Jackson2JsonRedisSerialize 替换默认序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);

        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);

        // 设置value的序列化规则和 key的序列化规则
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();

        return redisTemplate;
    

    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) 
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);

        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);

        // 配置序列化(解决乱码的问题),过期时间30秒
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofSeconds(1800000))
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();

        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    

 

 

三、application.yml 添加以下配置

spring:
## Redis 配置
## Redis数据库索引(默认为0)
  redis:
    database: 0
  ## Redis服务器地址
    host: 127.0.0.1
  ## Redis服务器连接端口
    port: 6379
  ## Redis服务器连接密码(默认为空)
    password:
  ## 连接超时时间(毫秒)
   # timeout: 1800000
  ## 连接池最大连接数(使用负值表示没有限制)
    jedis:
      pool:
        max-active: 8
  ## 连接池最大阻塞等待时间(使用负值表示没有限制)
        max-wait: -1
  ## 连接池中的最大空闲连接
        max-idle: 8
  ## 连接池中的最小空闲连接
        min-idle: 0

 

四、在方法上使用@Cacheable(value = "MyRedis"/*上一级包名*/, key = "redis")进行添加缓存

(方法必须有返回值,返回值就是存入Redis的数据)

 

五、删除缓存:

  方法一:删除指定Key

 

@Caching(evict = 
@CacheEvict(value = "MyRedis",key=“redis”)
)

方法二:删除指定文件下(value值)所有的Key

 

@Caching(evict = 
@CacheEvict(value = "MyRedis",allEntries=true/*表示删除MyRedis文件下所有的缓存*/)
)



六、有时候通过注解方式无法满足需求,可以使用RedisTemplate对象进行存储,关键代码如下:
 //获取redisTemplate对象
 @Autowired
 private RedisTemplate redisTemplate;

//新增 参数分别为:Key名、存入的数据、过期时间、过期时间的类型(TimeUnit.SECONDS 是以秒计算)
redisTemplate.opsForValue().set(“token”,admins,1800, TimeUnit.SECONDS);

//获取 参数:Key值
redisTemplate.opsForValue().get(“token”)

//删除 参数:Key值
redisTemplate.delete("token");


 




以上是关于SpringBoot 整合RabbitMq 自定义消息监听容器来实现消息批量处理的主要内容,如果未能解决你的问题,请参考以下文章

springBoot整合RabbitMQ(新手整合请勿喷)

SpringBoot2.x整合Redis缓存自定义序列化

springboot 整合RabbitMQ yml配置文件配置交换机 队列信息

SpringBoot日记——MQ消息队列整合

springboot整合rabbitmq(topic主题模式)

SpringBoot系列5SpringBoot整合RabbitMQ