springboot整合kafaka批量消费

Spring Boot整合Kafka批量消费可以通过设置Kafka监听器的批量消费配置来实现。具体步骤如下：

1. 在Spring Boot项目中添加Kafka依赖：

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

2. 创建Kafka消费者配置类：

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 设置批量消费
        factory.setBatchListener(true);
        // 批量消费每次最多拉取10条消息
        factory.setBatchListenerForCompactedTopics(false);
        factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.BATCH);
        return factory;
    }
}

3. 创建Kafka消息监听器：

@Service
public class KafkaConsumerListener {

    @KafkaListener(topics = "${spring.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group-id}")
    public void onMessage(List<String> messages) {
        for (String message : messages) {
            // 处理消息
        }
    }
}

其中，`@KafkaListener`注解的`List<String>`类型的参数表示批量消费的消息列表。

4. 配置application.yml文件：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      topic: test-topic
    consumer:
      topic: test-topic
      group-id: test-group

以上就是Spring Boot整合Kafka批量消费的基本配置。需要注意的是，批量消费也会存在消息重复消费的问题，需要根据具体业务场景进行处理。