Spring Cloud Stream Schema Registry

DeepSeek V3 中英对照 Spring Cloud Stream Schema Registry

简介

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

Spring Cloud Stream Schema Registry 概述

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

• 主题，即模式的逻辑名称

• 模式版本

• 模式格式，用于描述数据的二进制格式

Spring Cloud Stream Schema Registry 提供以下组件

• 独立模式 Schema Registry 服务器

  默认情况下，它使用 H2 数据库，但通过提供适当的数据源配置，服务器也可以与 PostgreSQL 或 MySQL 一起使用。

• 能够通过与 Schema Registry 通信进行消息编组的 Schema Registry 客户端。

  目前，客户端可以与独立模式 Schema Registry 或 Confluent Schema Registry 进行通信。

Schema Registry Client

与模式注册表服务器交互的客户端抽象是 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 Registry 进行交互。

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

@SpringBootApplication
@EnableSchemaRegistryClient
public class ConsumerApplication {

}

备注

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

Schema Registry 客户端属性

Schema Registry 客户端支持以下属性：

spring.cloud.stream.schemaRegistryClient.endpoint

schema-server 的位置。设置此项时，请使用完整的 URL，包括协议（http 或 https）、端口和上下文路径。

默认

[localhost:8990/](http://localhost:8990/)

spring.cloud.stream.schemaRegistryClient.cached

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

默认

false

Avro Schema Registry 客户端消息转换器

对于在应用上下文中注册了 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 Schema Registry 消息转换器属性

如果你已通过设置 spring.cloud.stream.stream.bindings.<output-binding-name>.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

将此属性中列出的所有 .avsc 文件注册到 Schema Server。

默认值：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 序列化的消息转换器：

• 使用序列化或反序列化对象的类信息，或在启动时已知位置的模式（schema）的转换器。

• 使用模式注册表的转换器。它们在运行时定位模式，并在领域对象演进时动态注册新模式。

支持模式的转换器

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

Schema Registry Server

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 配置选项 来自定义模式存储。

Schema Registry Server API

Schema Registry Server API 包含以下操作：

• POST / — 参见 [注册新模式](#spring-cloud-stream-overview-registering-new-schema)

• GET /{subject}/{format}/{version} — 参见 [按主题、格式和版本检索现有模式](#spring-cloud-stream-overview-retrieve-schema-subject-format-version)

• GET /{subject}/{format} — 参见 [按主题和格式检索现有模式](#spring-cloud-stream-overview-retrieve-schema-subject-format)

• GET /schemas/{id} — 参见 [按 ID 检索现有模式](#spring-cloud-stream-overview-retrieve-schema-id)

• DELETE /{subject}/{format}/{version} — 参见 [按主题、格式和版本删除模式](#spring-cloud-stream-overview-deleting-schema-subject-format-version)

• DELETE /schemas/{id} — 参见 [按 ID 删除模式](#spring-cloud-stream-overview-deleting-schema-id)

• DELETE /{subject} — 参见 [按主题删除模式](#spring-cloud-stream-overview-deleting-schema-subject)

注册新Schema

要注册新的模式，请向 / 端点发送 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 的 Schema Registry

默认配置会创建一个 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 platform 版本 4.0.0 进行测试。

模式注册与解析

为了更好地理解 Spring Cloud Stream 如何注册和解析新 schema 及其对 Avro schema 比较功能的使用，我们提供了两个独立的子章节：

• [模式注册流程（序列化）](#spring-cloud-stream-overview-schema-registration-process)

• [模式解析流程（反序列化）](#spring-cloud-stream-overview-schema-resolution-process)

模式注册流程（序列化）

注册过程的第一步是从通过通道发送的有效负载中提取模式。Avro 类型（如 SpecificRecord 或 GenericRecord）已经包含模式，可以直接从实例中检索。对于 POJO，如果 spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled 属性设置为 true（默认值），则会推断出模式。

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

根据模式版本信息，转换器会设置消息的 contentType 头部来携带版本信息——例如：application/vnd.user.v1+avro。

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

当读取包含版本信息的消息时（即带有类似 [Schema Registration Process (Serialization)](#spring-cloud-stream-overview-schema-registration-process) 中描述方案的 contentType 头信息），转换器会查询 Schema 服务器以获取消息的写入器模式。一旦找到传入消息的正确模式，它将检索读取器模式，并利用 Avro 的模式解析支持，将其读取到读取器定义中（设置默认值和任何缺失的属性）。

备注

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