springboot 集成 kafka批量消费数据

新建一个maven工程，添加kafka依赖

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>

yaml配置文件

spring:kafka:bootstrap-servers: ${local_host_ip}:9092producer:# producer 生产者retries:0# 重试次数acks:1# 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)batch-size:100000# 批量处理的最大大小 单位 bytebuffer-memory:33554432# 生产端缓冲区大小key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:# consumer消费者group-id: javagroup # 默认的消费组IDenable-auto-commit:false# 是否自动提交offset#auto-commit-interval: 100  # 提交offset延时(接收到消息后多久提交offset)# earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费# latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据# none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常auto-offset-reset: latest
          max-poll-records:1000#批量消费每次最多消费多少条消息key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        listener:enable-auto-commit:falseack-mode: manual_immediate
          max-poll-records:1000# 消费者监听的topic不存在时，项目会报错，设置为falsemissing-topics-fatal:falsetype: batch 

/**
     * 手动提交批量消费
     *
     * @param list (list 的大小和 batch-size（单位） 有关)
     * @param consumer
     */@KafkaListener(topics ="topic_test", groupId ="topic_test_group")publicvoidrecieveRealTimNta(List<Message<String>> list,Consumer consumer){try{//业务模块
            consumer.commitAsync();}catch(Exception e){
            log.error("安全事件-告警数据解析 -nta  消费失败");}finally{
            consumer.commitSync();}}