使用 @KafkaListener 时以编程方式设置属性答案

【问题标题】：Programmatically setting properties when using @KafkaListener使用 @KafkaListener 时以编程方式设置属性
【发布时间】：2019-06-25 13:19:32
【问题描述】：

我在我的 Spring Boot 应用程序中使用了 @KafkaListener 注释，而没有创建自定义 KafkaListenerContainerFactory bean。我目前正在我的 application.yml 文件中设置 spring.kafka.consumer.value-deserializer 属性来分配我的反序列化器，并且更喜欢以编程方式执行此操作以进行编译时检查。我意识到创建我自己的KafkaListenerContainerFactory 允许我在工厂设置这个属性，但我想避免样板和一些额外的复杂性来获取 SSL 设置。

有没有一种简单的方法可以在不创建我自己的KafkaListenerContainerFactory 的情况下以编程方式设置我的值反序列化器？

【问题讨论】：

Is there a simple way to set my value deserializer programmatically without creating my own KafkaListenerContainerFactory 是的，你可以这样做，但我真的不明白你的要求，你能解释一下，或者上传你拥有的代码并显示你需要什么改变跨度>
“以编程方式设置值”是什么意思——比如，你想创建一个反序列化器 bean 吗？就像上面说的死侍一样，你的用例是什么还不太清楚。
我相信他想要反序列化器上的类型安全（而不是将类名表示为字符串）。看我的回答。
@GaryRussell，是的，我一直在寻找类型安全，这是一个合理的解决方案。我知道这些属性只在 ConsumerFactory 上，但不知道我可以只创建那个 bean。说得通。我喜欢使用 @KafkaListener 注释的简单性，并试图避免使用其他样板来设置一个属性。谢谢！

标签： java spring-boot spring-kafka

【解决方案1】：

反序列化器继续用于创建消费者工厂而不是容器工厂的属性，您可以按如下方式覆盖引导的消费者工厂：

@Bean
public ConsumerFactory<?, ?> kafkaConsumerFactory(KafkaProperties properties) {
    Map<String, Object> props = properties.buildConsumerProperties();
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MyDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

【讨论】：

【解决方案2】：

根据docs

您可以像这样设置其他属性：

@KafkaListener(topics = "test-transactional",
               properties={"foo:bar","isolation.level:read_committed"})
public void listen(ConsumerRecord<?, ?> cr) throws Exception {
    logger.info("got consumer record" + cr.toString());
}


@KafkaListener(topics = "test-transactional",        
                         properties={"isolation.level:read_uncommitted"})
public void listenDifferent(ConsumerRecord<?, ?> cr) throws Exception {
    logger.info("uncomitted: got consumer record" + cr.toString());
}

【讨论】：