Spring Boot 对rabbitmq批量处理数据的支持

ゝ一纸荒年。 2021-06-10 20:39 1300阅读 0赞

**一丶前言**

在生产中，存在一些场景，需要对数据进行批量操作。如，可以先将数据存放到redis，然后将数据进行批量写进数据库。但是使用redis，不得不面对一个数据容易丢失的问题。也可以考虑使用消息队列进行替换，在数据持久化，数据不丢失方面，消息队列确实比redis好一点，毕竟设计不一样。是不是使用消息队列，就一定好呢？不是的，首先使用消息队列，不能确保数据百分百不丢失，(如果要做到百分百不丢失，设计上就会比较复杂)，除此之外，还要面对数据重复的问题。消息丢失，消息重复，是使用消息队列必须面对的问题。

AMQP在协议上规定每次只能传送一条数据，因此做批量数据操作，需要在应用层上定义，Spring 目前已经提供 ([来源资料][Link 1])

> M-m-m. No, there is no such a functionality. Only one message can be read at a time from the queue. And it is on the protocol level. That's why we have introduced that artificial `BatchingRabbitTemplate` to batch on the application level, before protocol. – [Artem Bilan][] [Nov 30 '16 at 12:02][Nov 30 _16 at 12_02]

** 二丶spring rabbit mq 支持批量操作的版本**

关于起始版本，笔者尚未查找到佐证资料，目前笔者所使用的是2.2.2版本

批量发送

![f31adf280ccdfb6658f0ff2941601ad3.png][]

批量监听(其实不一定需要实现该接口，目前笔者的实现是使用该接口，其他可以自行查看官方文档)

![181b87f0380e2d5f23c05aed7b36e003.png][]

[官方文档资料][Link 2]

![a3a86abc3ca640cc19b93643c39145c8.png][]

** 三丶实现**

1. 简单配置测试队列

//测试批量
        public static final String BATCH_QUEUE_NAME="batch.queue";
    
        @Bean
        public Queue batchQueue(){
            return new Queue(BATCH_QUEUE_NAME);
        }

2. 配置批量发送template

@Bean("batchQueueTaskScheduler")
        public TaskScheduler batchQueueTaskScheduler(){
            TaskScheduler taskScheduler=new ThreadPoolTaskScheduler();
            return taskScheduler;
        }
    
    
        //批量处理rabbitTemplate
        @Bean("batchQueueRabbitTemplate")
        public BatchingRabbitTemplate batchQueueRabbitTemplate(ConnectionFactory connectionFactory,
                                                               @Qualifier("batchQueueTaskScheduler") TaskScheduler taskScheduler){
    
            //!!!重点： 所谓批量， 就是spring 将多条message重新组成一条message, 发送到mq, 从mq接受到这条message后，在重新解析成多条message
    
            //一次批量的数量
            int batchSize=10;
            // 缓存大小限制,单位字节，
            // simpleBatchingStrategy的策略，是判断message数量是否超过batchSize限制或者message的大小是否超过缓存限制，
            // 缓存限制，主要用于限制"组装后的一条消息的大小"
            // 如果主要通过数量来做批量("打包"成一条消息), 缓存设置大点
            // 详细逻辑请看simpleBatchingStrategy#addToBatch()
            int bufferLimit=1024; //1 K
            long timeout=10000;
    
            //注意，该策略只支持一个exchange/routingKey
            //A simple batching strategy that supports only one exchange/routingKey
            BatchingStrategy batchingStrategy=new SimpleBatchingStrategy(batchSize,bufferLimit,timeout);
            return new BatchingRabbitTemplate(connectionFactory,batchingStrategy,taskScheduler);
        }

3. 批量监听 (**注意, 批量发送和批量监听可以各自独立使用**)

a. 配置监听容器(这里是必须的!!!)

@Bean("batchQueueRabbitListenerContainerFactory")
        public SimpleRabbitListenerContainerFactory batchQueueRabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
            SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
            factory.setConnectionFactory(connectionFactory);
            //设置批量
            factory.setBatchListener(true);
            factory.setConsumerBatchEnabled(true);//设置BatchMessageListener生效
            factory.setBatchSize(10);//设置监听器一次批量处理的消息数量
            return factory;
        }

b. 配置监听器

@Slf4j
    @Component
    public class BatchQueueListener implements BatchMessageListener {
    
        //批量接收处理
        @RabbitListener(queues = RabbitMqConfig2.BATCH_QUEUE_NAME,containerFactory = "batchQueueRabbitListenerContainerFactory")
        @Override
        public void onMessageBatch(List<Message> messages) {
            log.info("batch.queue.consumer 收到{}条message", messages.size());
            if(messages.size()>0){
                log.info("第一条数据是: {}", new String(messages.get(0).getBody()));
            }
        }
    }

4. 测试

// --------------------------- 测试batch
        @Autowired
        BatchingRabbitTemplate batchQueueRabbitTemplate;
    
        @Test
        public void batchSend() throws InterruptedException {
            // 除了send(String exchange, String routingKey, Message message, CorrelationData correlationData)方法是发送单条数据
            // 其他send都是批量发送
    
            //批量发送
    
    
            long timestamp=System.currentTimeMillis();
    
            String msg;
            Message message;
            MessageProperties messageProperties=new MessageProperties();
            for(int i=0;i<1000;i++){
                msg="batch."+timestamp+"-"+i;
    
    
                message=new Message(msg.getBytes(), messageProperties);
                batchQueueRabbitTemplate.send(RabbitMqConfig2.BATCH_QUEUE_NAME,message);
    
    //            defaultRabbitTemplate.convertAndSend(RabbitMqConfig2.BATCH_QUEUE_NAME, msg.getBytes());
            }
    
            System.out.println("发送数据完毕");
            System.out.println("等待30s");
            TimeUnit.SECONDS.sleep(30); //等待消费者消费
        }

5. 输出结果

![e76fc46a7345e2e02dac03f4620810b3.png][]

解析:

设置批量监听处理的数量为10，为什么输出是100呢?

因为使用了批量发送， 配置批量发送是将10条数据压缩成1条， 批量监听收到的是压缩后的10条，解析后，变成100条，没毛病

[完整demo源码][demo]

参考资料：

[官方文档][Link 2]

[Link 1]: https://stackoverflow.com/questions/40861435/how-we-can-use-the-batchingrabbittemplate-to-read-the-messages-as-a-batch-from-t
[Artem Bilan]: https://stackoverflow.com/users/2756547/artem-bilan
[Nov 30 _16 at 12_02]: https://stackoverflow.com/questions/40861435/how-we-can-use-the-batchingrabbittemplate-to-read-the-messages-as-a-batch-from-t#comment68989437_40868638
[f31adf280ccdfb6658f0ff2941601ad3.png]: /images/20210527/fc1a162b0ca248f7a77431b50000f8ba.png
[181b87f0380e2d5f23c05aed7b36e003.png]: /images/20210527/dd5f5db600b54fa1a3c2d6bd98b7b6b3.png
[Link 2]: https://docs.spring.io/spring-amqp/reference/html/#receiving-batch
[a3a86abc3ca640cc19b93643c39145c8.png]: /images/20210527/18f15fd2c2474630bdfd2e5a314408e5.png
[e76fc46a7345e2e02dac03f4620810b3.png]: /images/20210527/dec54d46d6c44affa3a113e6f062a8f2.png
[demo]: https://gitee.com/timfruit189/rabbitmq-demo