我把SpringKafka监听器初始化为
@Bean
public Map<String, Object> consumerConfig() {
final HashMap<String, Object> result = new HashMap<>();
result.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
result.put(GROUP_ID_CONFIG, groupId);
result.put(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
result.put(VALUE_DESERIALIZER_CLASS_CONFIG, MyKafkaJacksonRulesExecutionResultsDeserializer.class);
return result;
}
@Bean
public ConsumerFactory<Long, MessageResult> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfig());
}
@Bean
public ConcurrentKafkaListenerContainerFactory<Long, MessageResult> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Long, MessageResult> containerFactory = new ConcurrentKafkaListenerContainerFactory<>();
containerFactory.setConsumerFactory(consumerFactory());
containerFactory.setConcurrency(KAFKA_LISTENER_THREADS_COUNT);
containerFactory.getContainerProperties().setPollTimeout(KAFKA_LISTENER_POLL_TIMEOUT);
containerFactory.getContainerProperties().setAckOnError(true);
containerFactory.getContainerProperties().setAckMode(RECORD);
return containerFactory;
}用作
@KafkaListener(topics = "${spring.kafka.out-topic}")
public void processSrpResults(MessageResult result) {反序列化程序在反序列化过程中引发异常，导致无限循环，因为侦听器无法获取消息。
我怎样才能让Kafka的听众犯错误呢？
2条答案
按热度按时间jvlzgdj91#
你可以用
org.springframework.kafka.support.serializer.ErrorHandlingDeserializer2作为使用者配置中的值反序列化程序。这使您有机会优雅地处理反序列化错误，并允许您提交使用者记录。有关用法和更多详细信息，请访问https://docs.spring.io/spring-kafka/reference/html/#error-处理反序列化程序cpjpxq1n2#
我为引发异常的反序列化程序创建了一个子类。然后我在配置中使用它作为反序列化程序。然后，处理器必须处理空对象。