Spring Cloud Stream 模式注册表

简介

当组织拥有基于消息传递的发布/订阅架构，并且多个生产者和消费者微服务相互通信时，通常需要所有这些微服务就基于架构的契约达成一致。当此类架构需要演进以适应新的业务需求时，现有组件仍然需要继续工作。Spring Cloud Stream 提供对独立架构注册表服务器的支持，应用程序可以使用该服务器注册上述架构。Spring Cloud Stream 架构注册表支持还提供对基于 avro 的架构注册表客户端的支持，这些客户端本质上提供消息转换器，用于与架构注册表通信以在消息转换期间协调架构。Spring Cloud Stream 提供的架构演进支持既适用于上述独立架构注册表，也适用于 Confluent 提供的架构注册表，后者专门适用于 Apache Kafka。

Spring Cloud Stream 架构注册表概述

Spring Cloud Stream Schema Registry 提供对模式演化的支持，以便数据可以随着时间的推移而演化，并且仍然可以与旧的或新的生产者和消费者一起工作，反之亦然。大多数序列化模型，特别是那些旨在跨不同平台和语言实现可移植性的模型，都依赖于描述数据如何在二进制有效负载中序列化的模式。为了序列化数据然后对其进行解释，发送方和接收方都必须能够访问描述二进制格式的模式。在某些情况下，模式可以从序列化时的有效负载类型或从反序列化时的目标类型推断出来。但是，许多应用程序受益于访问描述二进制数据格式的显式模式。模式注册表允许您以文本格式（通常为 JSON）存储模式信息，并使需要该信息来以二进制格式接收和发送数据的各种应用程序可以访问该信息。模式可以作为由以下内容组成的元组进行引用

模式的逻辑名称
模式版本
模式格式，描述数据的二进制格式

Spring Cloud Stream Schema Registry 提供以下组件

独立模式注册表服务器

By default, it is using an H2 database, but server can be used with PostgreSQL or MySQL by providing appropriate datasource configuration.

能够通过与模式注册表通信来执行消息编组的模式注册表客户端。

Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.

模式注册表客户端

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

public interface SchemaRegistryClient {

    SchemaRegistrationResponse register(String subject, String format, String schema);

    String fetch(SchemaReference schemaReference);

    String fetch(Integer id);

}

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

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

@SpringBootApplication
@EnableSchemaRegistryClient
public class ConsumerApplication {

}

默认转换器经过优化，不仅可以缓存来自远程服务器的模式，还可以缓存相当昂贵的 parse() 和 toString() 方法。因此，它使用不缓存响应的 DefaultSchemaRegistryClient。如果您打算更改默认行为，则可以在代码中直接使用客户端并将其覆盖到所需结果。为此，您必须将属性 spring.cloud.stream.schemaRegistryClient.cached=true 添加到您的应用程序属性中。

模式注册表客户端属性

模式注册表客户端支持以下属性

spring.cloud.stream.schemaRegistryClient.endpoint: 模式服务器的位置。设置此项时，请使用完整的 URL，包括协议（http 或 https）、端口和上下文路径。
默认: localhost:8990/
spring.cloud.stream.schemaRegistryClient.cached: 客户端是否应缓存架构服务器响应。通常设置为 false，因为缓存发生在消息转换器中。使用架构注册表客户端的客户端应将其设置为 true。
默认: false

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

对于在应用程序上下文中注册了 SchemaRegistryClient Bean 的应用程序，Spring Cloud Stream 会自动配置一个 Apache Avro 消息转换器以进行架构管理。这简化了架构演进，因为接收消息的应用程序可以轻松获取可与其自己的读取器架构协调的编写器架构。

对于出站消息，如果绑定的内容类型设置为 application/*+avro，则会激活 MessageConverter，如下例所示

spring.cloud.stream.stream.bindings.<output-binding-name>.contentType=application/*+avro

在出站转换期间，消息转换器会尝试推断每个出站消息的架构（基于其类型），并使用 SchemaRegistryClient 将其注册到主题（基于有效负载类型）。如果已找到相同的架构，则会检索对其的引用。如果没有，则会注册架构，并提供一个新版本号。消息会使用以下方案通过 contentType 标头发送：application/[prefix].[subject].v[version]+avro，其中 prefix 是可配置的，subject 从有效负载类型中推断出来。

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

在接收消息时，转换器会从传入消息的标头中推断架构引用并尝试检索它。架构在反序列化过程中用作编写器架构。

Avro 架构注册表消息转换器属性

如果您通过设置 spring.cloud.stream.stream.bindings.<output-binding-name>.contentType=application/*+avro 启用了基于 Avro 的架构注册表客户端，则可以通过设置以下属性来自定义注册行为。

spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled: 启用此选项，如果您希望转换器使用反射从 POJO 推断架构。

默认值：false
spring.cloud.stream.schema.avro.readerSchema: Avro 通过查看编写器架构（原始有效负载）和读取器架构（您的应用程序有效负载）来比较架构版本。有关更多信息，请参见 Avro 文档。如果设置，这将覆盖架构服务器上的任何查找，并将本地架构用作读取器架构。默认值：null
spring.cloud.stream.schema.avro.schemaLocations: 使用架构服务器注册此属性中列出的任何 .avsc 文件。

默认值：empty
spring.cloud.stream.schema.avro.prefix: 用于 Content-Type 标头的前缀。

默认值：vnd
spring.cloud.stream.schema.avro.subjectNamingStrategy: 确定用于在模式注册表中注册 Avro 模式的主题名称。提供两种实现，org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy（主题为模式名称）和 org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy（使用 Avro 模式命名空间和名称返回完全限定的主题）。可以通过实现 org.springframework.cloud.stream.schema.avro.SubjectNamingStrategy 创建自定义策略。

默认值：org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy
spring.cloud.stream.schema.avro.ignoreSchemaRegistryServer: 忽略任何模式注册表通信。这对于测试目的很有用，这样在运行单元测试时，它不会不必要地尝试连接到模式注册表服务器。

默认值：false

Apache Avro 消息转换器

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

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

使用已序列化或已反序列化的对象的类信息或在启动时已知位置的模式的转换器。
使用模式注册表的转换器。它们在运行时查找模式，并在域对象演化时动态注册新模式。

支持模式的转换器

AvroSchemaMessageConverter 支持通过使用预定义模式或使用类中可用的模式信息（通过反射或包含在 SpecificRecord 中）来序列化和反序列化消息。如果您提供自定义转换器，则不会创建默认的 AvroSchemaMessageConverter Bean。以下示例显示了一个自定义转换器

要使用自定义转换器，您可以简单地将其添加到应用程序上下文中，可以选择指定一个或多个要与其关联的 MimeTypes。默认 MimeType 为 application/avro。

如果转换的目标类型是 GenericRecord，则必须设置模式。

以下示例展示了如何在接收应用程序中配置转换器，方法是注册不带预定义模式的 Apache Avro MessageConverter。在此示例中，请注意 MIME 类型值为 avro/bytes，而不是默认的 application/avro。

@SpringBootApplication
public static class SinkApplication {

  //...

  @Bean
  public MessageConverter userMessageConverter() {
      return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
  }
}

相反，以下应用程序使用预定义架构（在类路径中找到）注册转换器

@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;
  }
}

架构注册表服务器

Spring Cloud Stream 提供架构注册表服务器实现。要使用它，您可以下载最新的 spring-cloud-stream-schema-registry-server 版本并将其作为独立应用程序运行

wget https://repo1.maven.org/maven2/org/springframework/cloud/spring-cloud-stream-schema-registry-server/4.0.3/spring-cloud-stream-schema-registry-server-4.0.3.jar
java -jar ./spring-cloud-stream-schema-registry-server-4.0.3.jar

您可以将架构注册表嵌入到现有的 Spring Boot Web 应用程序中。为此，将 spring-cloud-stream-schema-registry-core 工件添加到您的项目中，并使用 @EnableSchemaRegistryServer 注释，它会将架构注册表服务器 REST 控制器添加到您的应用程序中。以下示例显示了启用架构注册表的 Spring Boot 应用程序

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

spring.cloud.stream.schema.server.path 属性可用于控制架构服务器的根路径（尤其是在将其嵌入到其他应用程序中时）。spring.cloud.stream.schema.server.allowSchemaDeletion 布尔属性启用架构的删除。默认情况下，这是禁用的。

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

架构注册表服务器 API

架构注册表服务器 API 包含以下操作

POST / — 请参阅 注册新架构
GET /{subject}/{format}/{version} — 请参阅 按主题、格式和版本检索现有架构
GET /{subject}/{format} — 请参阅 按主题和格式检索现有架构
GET /schemas/{id} — 请参阅 按 ID 检索现有架构
DELETE /{subject}/{format}/{version} — 请参阅 按主题、格式和版本删除架构
DELETE /schemas/{id} — 请参阅 按 ID 删除架构
DELETE /{subject} — 请参阅 按主题删除架构

注册新架构

要注册新架构，请向 / 端点发送 POST 请求。

/ 接受具有以下字段的 JSON 有效负载

subject：模式主题
format：模式格式
definition：模式定义

其响应是 JSON 中的模式对象，具有以下字段

id：模式 ID
subject：模式主题
format：模式格式
version：模式版本
definition：模式定义

按主题、格式和版本检索现有模式

要按主题、格式和版本检索现有模式，请向 {subject}/{format}/{version} 端点发送 GET 请求。

其响应是 JSON 中的模式对象，具有以下字段

id：模式 ID
subject：模式主题
format：模式格式
version：模式版本
definition：模式定义

按主题和格式检索现有模式

要按主题和格式检索现有模式，请向 /subject/format 端点发送 GET 请求。

其响应是具有每个 JSON 模式对象的模式列表，具有以下字段

id：模式 ID
subject：模式主题
format：模式格式
version：模式版本
definition：模式定义

按 ID 检索现有模式

要按其 ID 检索模式，请向 /schemas/{id} 端点发送 GET 请求。

其响应是 JSON 中的模式对象，具有以下字段

id：模式 ID
subject：模式主题
format：模式格式
version：模式版本
definition：模式定义

按主题、格式和版本删除模式

要删除由其主题、格式和版本标识的模式，请向 {subject}/{format}/{version} 端点发送 DELETE 请求。

按 ID 删除模式

要按其 ID 删除模式，请向 /schemas/{id} 端点发送 DELETE 请求。

按主题删除模式

DELETE /{subject}

按其主题删除现有模式。

此说明仅适用于 Spring Cloud Stream 1.1.0.RELEASE 用户。Spring Cloud Stream 1.1.0.RELEASE 使用表名 schema 来存储 Schema 对象。Schema 是许多数据库实现中的关键字。为了避免将来出现任何冲突，从 1.1.1.RELEASE 开始，我们为存储表选择了名称 SCHEMA_REPOSITORY。任何升级的 Spring Cloud Stream 1.1.0.RELEASE 用户都应在升级前将其现有模式迁移到新表。

使用 Confluent 的模式注册表

默认配置创建一个 DefaultSchemaRegistryClient bean。如果您想使用 Confluent 模式注册表，则需要创建一个类型为 ConfluentSchemaRegistryClient 的 bean，它将取代框架默认配置的 bean。以下示例显示了如何创建这样的 bean

@Bean
public SchemaRegistryClient schemaRegistryClient(@Value("${spring.cloud.stream.schemaRegistryClient.endpoint}") String endpoint){
  ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient();
  client.setEndpoint(endpoint);
  return client;
}

ConfluentSchemaRegistryClient 已针对 Confluent 平台版本 4.0.0 进行了测试。

模式注册和解析

为了更好地理解 Spring Cloud Stream 如何注册和解析新模式及其如何使用 Avro 模式比较功能，我们提供了两个单独的小节

模式注册过程（序列化）
模式解析过程（反序列化）

模式注册过程（序列化）

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

一旦获得模式，转换器就会从远程服务器加载其元数据（版本）。首先，它会查询本地缓存。如果未找到结果，它会将数据提交到服务器，服务器会回复版本信息。转换器始终缓存结果，以避免为需要序列化的每条新消息查询模式服务器所带来的开销。

使用模式版本信息，转换器会将消息的 contentType 标头设置为承载版本信息，例如：application/vnd.user.v1+avro。

模式解析过程（反序列化）

在读取包含版本信息（即，contentType 标头，其中包含类似于 模式注册过程（序列化） 中所述方案的方案）的消息时，转换器会查询模式服务器以获取消息的写入模式。一旦找到传入消息的正确模式，它就会检索读取模式，并通过使用 Avro 的模式解析支持，将其读入读取器定义（设置默认值和任何缺失的属性）。

你应该理解写入模式（编写消息的应用程序）和读取模式（接收应用程序）之间的区别。我们建议花点时间阅读 Avro 术语并了解该过程。Spring Cloud Stream 始终获取写入模式以确定如何读取消息。如果你希望让 Avro 的模式演化支持正常工作，则需要确保为你的应用程序正确设置了 readerSchema。