发生错误时如何关闭KafkaListener

当发生错误时，关闭KafkaListener可以通过以下步骤完成：

首先，需要在代码中定义一个KafkaListenerContainerFactory bean，用于创建KafkaListenerContainer。可以使用Spring Kafka提供的DefaultKafkaListenerContainerFactory类来创建。

@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    DefaultKafkaListenerContainerFactory<String, String> factory = new DefaultKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setErrorHandler(new SeekToCurrentErrorHandler());
    return factory;
}

在KafkaListener注解中，指定containerFactory属性为上一步定义的KafkaListenerContainerFactory bean的名称。

@KafkaListener(topics = "topicName", containerFactory = "kafkaListenerContainerFactory")
public void listen(ConsumerRecord<String, String> record) {
    // 处理消息
}

当发生错误时，可以使用KafkaListenerErrorHandler来处理错误。可以自定义一个实现了ErrorHandler接口的类，并在KafkaListener注解中指定errorHandler属性为该类的名称。

@Component
public class CustomKafkaListenerErrorHandler implements ErrorHandler {
    @Override
    public void handle(Exception thrownException, ConsumerRecord<?, ?> record) {
        // 处理错误
    }
}

@KafkaListener(topics = "topicName", containerFactory = "kafkaListenerContainerFactory", errorHandler = "customKafkaListenerErrorHandler")
public void listen(ConsumerRecord<String, String> record) {
    // 处理消息
}