Springboot环境下RabbitMQ的批量消费(consumer-batch-enabled)

Springboot环境下RabbitMQ的批量消费(consumer-batch-enabled) 概述:

spring-rabbit2.2版本后添加了批量消费, 网上没找到相关的用法,特此记录一下

环境:

jdk1.8 RabbitMQ3.8.3 新建springboot项目

配置:

引入amqp,如下

    org.springframework.boot
    spring-boot-starter-amqp

修改application.yml配置如下

spring:
  rabbitmq:
    host: mymq.net
    virtual-host: /
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual
        prefetch: 250
        consumer-batch-enabled: true    #开启批量消费
        batch-size: 16                  #每次批量消费大小

指定批处理MessageListenerConverter(这里必须设置)

@Slf4j
@Component
public class BatchConfig {

    @Bean
    public String cusSimpleRabbitListenerContainerFactory(SimpleRabbitListenerContainerFactory containerFactory) {
        log.info("设置批量-----");
        containerFactory.setBatchListener(true);
        return "cusSimpleRabbitListenerContainerFactory";
    }
}

配置消费者

@Slf4j
@Component
public class CallCompleteConsumer {

    @RabbitHandler
    @RabbitListener(queues = "my-consume-queue")
    public void onMessage(List messages, Channel channel) {

        final List messageList = messages.stream()
                .map(each -> new String(each.getBody()))
                .collect(Collectors.toList());
        log.info("[onMessage] {}: {}", messages.size(), messageList);
        try {
            TimeUnit.MILLISECONDS.sleep(10);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            for (Message message : messages) {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

运行:

先创建好对列, 修改参数,启动即可

总结:

    消息批处理比较适合像处理结果批量入库批量 *** 作效率高于单独 *** 作的时候, 如处理结果批量更新到Mysql, 这里说一下ack与Kafka批处理的区别, Kafka批处理ack的时候提交的是offset, 而这里的批处理只针对单个消息的tag, 必须每个都要ack.