带有Avro的Spring Cloud Stream无法正确转换字符串消息

Question

我有一个问题，源发送GenericMessage[payload=xxxxx，...]而接收器接收消息作为10,120,120,120,120,120。

这个问题发生在我设置Avro消息转换器之后。如果我删除Avro消息转换器并使用StreamListener来处理消息转换，它会正常工作。

源应用程序。属性

spring.cloud.stream.bindings.toGreeting.destination=greeting
spring.cloud.stream.bindings.toGreeting.contentType=application/*+avro
spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled=true

水槽应用

server.port=8990 
spring.cloud.stream.bindings.greeting.destination=greeting

消息转换器

@Configuration
@EnableSchemaRegistryClient
public class MessageConverterConfig {
    @Bean
    public MessageConverter topic1MessageConverter() throws IOException {
        return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
    }
}

应用程序类

@SpringBootApplication
@EnableSchemaRegistryClient
public class SourceApplication {
    public static void main(String[] args) {
        SpringApplication.run(SourceApplication.class, args);
    }
}

@EnableSchemaRegistryServer
@EnableSchemaRegistryClient
@SpringBootApplication
public class SinkApplication {
    public static void main(String[] args) {
        SpringApplication.run(SinkApplication.class, args);
    }
}

我是否缺少配置？谢谢

Answer 1

如果您正在设置spring.cloud.stream.bindings.toGreeting.contentType=Application/*avro，则需要使用AvroSchemaRegstryClientMessageConverter（由SCSt配置），并且您不必为MimeTypeavro/bytes设置显式转换器topic1MessageConverter。

如果要使用此转换器，则需要设置spring。云流动绑定。toGreeting。contentType=avro/字节。

Answer 2

这里有一条简单的规则：

如果您只想拥有一个可以从avro序列化/反序列化的消息转换器，并且您可以在配置GenericRecords期间提供架构位置，或者您的StreamListener方法具有SpecificRecord类型的签名。然后选择AvroSchemaMessageConverter，像设置一样进行设置，但使用avro/bytes。我们保留应用程序/*avro以支持模式演化。

因此，如果您设置了EnableSchemaRegistryClient，那么您将委托给外部注册表来拥有您的模式。在这种情况下，不仅需要注册表，还需要在那里注册的模式。

默认情况下，如果是spring，生产者将自动注册SpecificRecord/GenericRecord或POJO类型的任何有效负载。云流动架构。阿夫罗。已启用dynamicSchemaGenerationEnabled。

在这种情况下，生产者实际上会将头设置为类似于应用程序/vnd的内容。使用者v1 avro假设你的主题是用户，并且是第一个版本。

下游，如果您的消费者还配置了应用程序/*avro，他们将能够读取此contentType并推断其主题/版本，以查询架构服务器并检索适当的架构。