Schema进化支持

Spring Cloud Stream通过其spring-cloud-stream-schema模块为基于模式的消息转换器提供支持。目前，基于模式的消息转换器开箱即用的唯一序列化格式是Apache Avro，在将来的版本中可以添加更多的格式。

Apache Avro讯息转换器

spring-cloud-stream-schema模块包含可用于Apache Avro序列化的两种类型的消息转换器：

• 使用序列化/反序列化对象的类信息的转换器，或者启动时已知位置的模式;
• 转换器使用模式注册表 - 他们在运行时定位模式，以及随着域对象的发展动态注册新模式。

具有模式支持的转换器

AvroSchemaMessageConverter支持使用预定义模式或使用类中可用的模式信息（反射或包含在SpecificRecord）中的序列化和反序列化消息。如果转换的目标类型是GenericRecord，则必须设置模式。

对于使用它，您可以简单地将其添加到应用程序上下文中，可选地指定一个或多个MimeTypes将其关联。默认MimeType为application/avro。

以下是在注册Apache Avro MessageConverter的宿应用程序中进行配置的示例，而不需要预定义的模式：

@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {
  ...
  @Bean
  public MessageConverter userMessageConverter() {
   return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
  }
}

相反，这里是一个应用程序，注册一个具有预定义模式的转换器，可以在类路径中找到：

@EnableBinding(Sink.class)
@SpringBootApplication
public static class SinkApplication {
  ...
  @Bean
  public MessageConverter userMessageConverter() {
   AvroSchemaMessageConverter converter = new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
   converter.setSchemaLocation(new ClassPathResource("schemas/User.avro"));
   return converter;
  }
}

为了了解模式注册表客户端转换器，我们将首先描述模式注册表支持。

Schema注册表支持

大多数序列化模型，特别是旨在跨不同平台和语言进行可移植性的序列化模型，依赖于描述数据如何在二进制有效载荷中被序列化的模式。为了序列化数据然后解释它，发送方和接收方都必须访问描述二进制格式的模式。在某些情况下，可以从序列化的有效载荷类型或从反序列化时的目标类型中推断出模式，但是在许多情况下，应用程序可以从访问描述二进制数据格式的显式模式中受益。模式注册表允许您以文本格式（通常为JSON）存储模式信息，并使该信息可访问需要它的各种应用程序以二进制格式接收和发送数据。一个模式可以作为一个元组引用，它由

• 作为模式的逻辑名称的主题 ;
• 模式版本 ;
• 描述数据 的二进制格式的模式格式。

Schema注册服务器

Spring Cloud Stream提供了模式注册表服务器实现。为了使用它，您可以简单地将spring-cloud-stream-schema-server工件添加到项目中，并使用@EnableSchemaRegistryServer注释，将模式注册表服务器REST控制器添加到应用程序中。此注释旨在与Spring Boot Web应用程序一起使用，服务器的监听端口由server.port设置控制。spring.cloud.stream.schema.server.path设置可用于控制模式服务器的根路径（特别是嵌入其他应用程序时）。spring.cloud.stream.schema.server.allowSchemaDeletion布尔设置可以删除模式。默认情况下，这是禁用的。

模式注册表服务器使用关系数据库来存储模式。默认情况下，它使用一个嵌入式数据库。您可以使用Spring Boot SQL数据库和JDBC配置选项自定义模式存储。

启用模式注册表的Spring Boot应用程序如下所示：

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

Schema注册服务器API

Schema注册服务器API由以下操作组成：

POST /

注册一个新的架构

接受具有以下字段的JSON有效载荷：

• subject模式主题;
• format模式格式;
• definition模式定义。

响应是JSON格式的模式对象，包含以下字段：

• id模式标识;
• subject模式主题;
• format模式格式;
• version模式版本;
• definition模式定义。

GET /{subject}/{format}/{version}

根据其主题，格式和版本检索现有模式。

响应是JSON格式的模式对象，包含以下字段：

• id模式标识;
• subject模式主题;
• format模式格式;
• version模式版本;
• definition模式定义。

GET /{subject}/{format}

根据其主题和格式检索现有模式的列表。

响应是JSON格式的每个模式对象的模式列表，包含以下字段：

• id模式标识;
• subject模式主题;
• format模式格式;
• version模式版本;
• definition模式定义。

GET /schemas/{id}

通过其id来检索现有的模式。

响应是JSON格式的模式对象，包含以下字段：

• id模式标识;
• subject模式主题;
• format模式格式;
• version模式版本;
• definition模式定义。

DELETE /{subject}/{format}/{version}

按其主题，格式和版本删除现有模式。

DELETE /schemas/{id}

按其ID删除现有模式。

DELETE /{subject}

按其主题删除现有模式。

Schema注册表客户端

与模式注册表服务器交互的客户端抽象是SchemaRegistryClient接口，具有以下结构：

public interface SchemaRegistryClient {
	SchemaRegistrationResponse register(String subject, String format, String schema);
	String fetch(SchemaReference schemaReference);
	String fetch(Integer id);
}

Spring Cloud Stream提供了开箱即用的实现，用于与其自己的模式服务器交互，以及与Confluent Schema注册表进行交互。

可以使用@EnableSchemaRegistryClient配置Spring Cloud Stream模式注册表的客户端，如下所示：

  @EnableBinding(Sink.class)
  @SpringBootApplication
  @EnableSchemaRegistryClient
  public static class AvroSinkApplication {
  ...
  }

Schema注册表客户端属性

Schema注册表客户端支持以下属性：

spring.cloud.stream.schemaRegistryClient.endpoint

模式服务器的位置。在设置时使用完整的URL，包括协议（http或https），端口和上下文路径。

默认

http://localhost:8990/

spring.cloud.stream.schemaRegistryClient.cached

客户端是否应缓存模式服务器响应。通常设置为false，因为缓存发生在消息转换器中。使用模式注册表客户端的客户端应将其设置为true。

默认

true

Avro Schema注册表客户端消息转换器

对于在应用程序上下文中注册了SchemaRegistryClient bean的Spring Boot应用程序，Spring Cloud Stream将自动配置使用模式注册表客户端进行模式管理的Apache Avro消息转换器。这简化了模式演进，因为接收消息的应用程序可以轻松访问可与自己的读取器模式进行协调的写入器模式。

对于出站邮件，如果频道的内容类型设置为application/*+avro，MessageConverter将被激活，例如：

spring.cloud.stream.bindings.output.contentType=application/*+avro

在出站转换期间，消息转换器将尝试基于其类型推断出站消息的模式，并使用SchemaRegistryClient根据有效载荷类型将其注册到主题。如果已经找到相同的模式，那么将会检索对它的引用。如果没有，则将注册模式并提供新的版本号。该消息将使用application/[prefix].[subject].v[version]+avro的方案contentType头发送，其中prefix是可配置的，并且从有效载荷类型推导出subject。

例如，类型为User的消息可以作为内容类型为application/vnd.user.v2+avro的二进制有效载荷发送，其中user是主题，2是版本号。

当接收到消息时，转换器将从传入消息的头部推断出模式引用，并尝试检索它。该模式将在反序列化过程中用作写入器模式。

Avro Schema注册表消息转换器属性

如果您已通过设置spring.cloud.stream.bindings.output.contentType=application/*+avro启用基于Avro的模式注册表客户端，则可以使用以下属性自定义注册的行为。

spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled

如果您希望转换器使用反射从POJO推断Schema，则启用。

默认

false

spring.cloud.stream.schema.avro.readerSchema

Avro通过查看编写器模式（源有效载荷）和读取器模式（应用程序有效负载）来比较模式版本，查看Avro文档以获取更多信息。如果设置，这将覆盖模式服务器上的任何查找，并将本地模式用作读取器模式。

默认

null

spring.cloud.stream.schema.avro.schemaLocations

使用Schema服务器注册此属性中列出的任何.avsc文件。

默认

empty

spring.cloud.stream.schema.avro.prefix

要在Content-Type头上使用的前缀。

默认

vnd

Schema注册和解决

为了更好地了解Spring Cloud Stream注册和解决新模式以及其使用Avro模式比较功能，我们将提供两个单独的子部分：一个用于注册，一个用于解析模式。

Schema注册流程（序列化）

注册过程的第一部分是从通过信道发送的有效载荷中提取模式。Avro类型，如SpecificRecord或GenericRecord已经包含一个模式，可以从实例中立即检索。在POJO的情况下，如果属性spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled设置为true（默认），则会推断出一个模式。

一旦获得了架构，转换器就会从远程服务器加载其元数据（版本）。首先，它查询本地缓存，如果没有找到它，则将数据提交到将使用版本控制信息回复的服务器。转换器将始终缓存结果，以避免为每个需要序列化的新消息查询Schema服务器的开销。

使用模式版本信息，转换器设置消息的contentType头，以携带版本信息，如application/vnd.user.v1+avro

Schema解析过程（反序列化）

当读取包含版本信息的消息（即，具有上述方案的contentType标头）时，转换器将查询Schema服务器以获取消息的写入器架构。一旦找到传入消息的正确架构，它就会检索读取器架构，并使用Avro的架构解析支持将其读入读取器定义（设置默认值和缺少的属性）。