Spring Cloud Stream Kafka 自定义 Serde Bean答案

【问题标题】：Spring Cloud Stream Kafka Custom Serde BeanSpring Cloud Stream Kafka 自定义 Serde Bean
【发布时间】：2020-12-22 21:01:34
【问题描述】：

我正在使用 Spring Cloud Stream 开发一个新的 Kafka 应用程序。我关注了最近推出的新功能绑定。

从文档中它说活页夹将尝试匹配来自 Kafka Streams 的一些常见类型，例如 Integer、String、byte[] 等。

我有一个自定义类型，根据文档我需要创建一个自定义 Serde。

我的自定义 Serde bean 定义如下所示：

@Bean
public Serde<Foo> fooSerde() {...}

此 Stream 绑定按预期工作。

@Bean
public Function<KStream<String, byte[]>, KStream<String, Foo[]>> process() {...}

但是，如果我更改为以下内容，则活页夹无法找到我的自定义 Serde bean，它默认为 Json Serde。

@Bean
public Function<KStream<String, byte[]>, KStream<String, ?[]>> process() {...}

请注意“？”作为 KStream 的值类型。这是预期的行为还是我在这里遗漏了什么？

我绝对可以回到application.yml 中的绑定，例如spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.valueSerde=Foo.class，但是由于依赖注入，我无法创建这样的无参数 Serde。我需要依靠SchemaRegistryClient 注入我的Serde，因为我使用的是AVRO，因此我不能在Kafka 请求的自定义Serde 中使用无参数构造函数，除非我使用反模式来制作@987654330 @静态。

任何帮助将不胜感激！

【问题讨论】：

标签： apache-kafka spring-cloud-stream

【解决方案1】：

是时候回答我自己的问题了。

问题在于通配符“？”。由于这个通配符是 KStream 的值，spring 无法挂钩 serde bean。因此解决方案是在参数中具有强类型。然后一切正常。

【讨论】：