此版本仍在开发中,尚未被视为稳定版本。对于最新的稳定版本,请使用 spring-cloud-stream 4.2.0! |
Spring Cloud Stream Schema 注册表
介绍
当组织拥有基于消息传递的发布/订阅架构,并且多个生产者和使用者微服务相互通信时,所有这些微服务通常需要就基于架构的合同达成一致。 当此类架构需要发展以适应新的业务需求时,现有组件仍需要继续工作。 Spring Cloud Stream 提供了对独立模式注册表服务器的支持,应用程序可以使用该服务器注册和使用上述模式。 Spring Cloud Stream 模式注册表支持还提供对基于 avro 的模式注册表客户端的支持,这些客户端本质上提供了与模式注册表通信的消息转换器,以便在消息转换期间协调模式。 Spring Cloud Stream 提供的模式演变支持既适用于上述独立模式注册表,也适用于 Confluent 提供的专门用于 Apache Kafka 的模式注册表。
Spring Cloud Stream Schema Registry 概述
Spring Cloud Stream Schema Registry 提供了对模式演变的支持,以便数据可以随着时间的推移而演变,并且仍然可以与较旧或较新的生产者和使用者一起使用,反之亦然。 大多数序列化模型,尤其是那些旨在在不同平台和语言之间实现可移植性的序列化模型,都依赖于描述如何在二进制有效负载中序列化数据的架构。 为了序列化数据,然后解释数据,发送方和接收方都必须能够访问描述二进制格式的架构。 在某些情况下,可以从序列化时的有效负载类型或反序列化时的目标类型推断架构。 但是,许多应用程序受益于访问描述二进制数据格式的显式架构。 Schema 注册表允许您以文本格式(通常为 JSON)存储 Schema 信息,并使需要它以二进制格式接收和发送数据的各种应用程序可以访问该信息。 架构可作为元组进行引用,该元组由以下部分组成:
-
作为架构的逻辑名称的主题
-
架构版本
-
架构格式,用于描述数据的二进制格式
Spring Cloud Stream Schema Registry 提供以下组件
-
独立 Schema Registry 服务器
By default, it is using an H2 database, but server can be used with PostgreSQL or MySQL by providing appropriate datasource configuration.
-
Schema Registry 客户端能够通过与 Schema Registry 通信来进行消息封送。
Currently, the client can communicate to the standalone schema registry or the Confluent Schema Registry.
Schema Registry 客户端
用于与 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 Registry 进行交互。
Spring Cloud Stream 模式注册表的客户端可以使用@EnableSchemaRegistryClient
如下:
@SpringBootApplication
@EnableSchemaRegistryClient
public class ConsumerApplication {
}
默认转换器经过优化,不仅可以缓存来自远程服务器的架构,还可以缓存parse() 和toString() 方法,这些方法非常昂贵。
因此,它使用DefaultSchemaRegistryClient ,则不会缓存响应。
如果您打算更改默认行为,则可以直接在代码上使用 Client 端并将其覆盖为所需的结果。
为此,您必须添加属性spring.cloud.stream.schemaRegistryClient.cached=true 添加到您的应用程序属性中。 |
Schema Registry 客户端属性
Schema Registry Client 支持以下属性:
spring.cloud.stream.schemaRegistryClient.endpoint
-
架构服务器的位置。 设置此项时,请使用完整的 URL,包括协议 (
http
或https
)、端口和上下文路径。 - 违约
spring.cloud.stream.schemaRegistryClient.cached
-
客户端是否应缓存架构服务器响应。 通常设置为
false
,因为缓存发生在消息转换器中。 使用 Schema 注册表客户端的客户端应将此设置为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
是从 payload 类型推导出来的。
例如,类型为User
可以作为内容类型为application/vnd.user.v2+avro
哪里user
是主题,而2
是版本号。
接收消息时,转换器从传入消息的标头推断架构引用,并尝试检索它。该架构在反序列化过程中用作写入器架构。
Avro Schema Registry 消息转换器属性
如果您通过设置spring.cloud.stream.stream.bindings.<output-binding-name>.contentType=application/*+avro
中,您可以通过设置以下属性来自定义注册的行为。
- spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled
-
如果您希望转换器使用反射从 POJO 推断 Schema,请启用 Enable。
违约:
false
- spring.cloud.stream.schema.avro.readerSchema
-
Avro 通过查看写入器架构(源负载)和读取器架构(您的应用程序负载)来比较架构版本。有关更多信息,请参阅 Avro 文档。 如果设置,这将覆盖架构服务器上的任何查找,并使用本地架构作为读取器架构。 违约:
null
- spring.cloud.stream.schema.avro.schema位置
-
注册任何
.avsc
文件。违约:
empty
- spring.cloud.stream.schema.avro.prefix
-
要在 Content-Type 标头上使用的前缀。
违约:
vnd
- spring.cloud.stream.schema.avro.subject命名Strategy
-
确定用于在架构注册表中注册 Avro 架构的主题名称。有两种实现可用:
org.springframework.cloud.stream.schema.avro.DefaultSubjectNamingStrategy
,其中 subject 是架构名称,而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
-
忽略任何 Schema 注册表通信。用于测试目的,以便在运行单元测试时,它不会不必要地尝试连接到 Schema Registry 服务器。
违约:
false
Apache Avro 消息转换器
Spring Cloud Stream 通过其spring-cloud-stream-schema-registry-client
模块。
目前,基于架构的消息转换器唯一支持的开箱即用序列化格式是 Apache Avro,未来版本将添加更多格式。
这spring-cloud-stream-schema-registry-client
module 包含两种类型的消息转换器,可用于 Apache Avro 序列化:
-
使用序列化或反序列化对象的类信息或启动时位置已知的架构的转换器。
-
使用 Schema 注册表的转换器。它们在运行时定位架构,并随着域对象的发展动态注册新架构。
支持 Schema 的转换器
这AvroSchemaMessageConverter
支持使用预定义的架构或使用类中可用的架构信息(反射式或包含在SpecificRecord
).
如果提供自定义转换器,则不会创建默认的AvroSchemaMessageConverter Bean。
以下示例显示了一个自定义转换器:
要使用自定义转换器,您只需将其添加到应用程序上下文中,可以选择指定一个或多个MimeTypes
与它相关联。
默认的MimeType
是application/avro
.
如果转化的目标类型是GenericRecord
,则必须设置 Schema。
以下示例演示如何通过注册 Apache Avro 在 sink 应用程序中配置转换器MessageConverter
没有预定义的架构。
在此示例中,请注意 mime type 值为avro/bytes
,而不是默认application/avro
.
@SpringBootApplication
public static class SinkApplication {
//...
@Bean
public MessageConverter userMessageConverter() {
return new AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes"));
}
}
相反,以下应用程序使用预定义的模式(在 Classpath 上找到)注册一个转换器:
@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 服务器
Spring Cloud Stream 提供了一个 schema registry 服务器实现。
要使用它,您可以下载最新的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
您可以将 Schema 注册表嵌入到现有的 Spring Boot Web 应用程序中。
为此,请添加
|
这spring.cloud.stream.schema.server.path
属性可用于控制架构服务器的根路径(尤其是当它嵌入到其他应用程序中时)。
这spring.cloud.stream.schema.server.allowSchemaDeletion
boolean 属性允许删除架构。默认情况下,此选项处于禁用状态。
Schema 注册表服务器使用关系数据库来存储 Schema。 默认情况下,它使用嵌入式数据库。 您可以使用 Spring Boot SQL 数据库和 JDBC 配置选项自定义架构存储。
Schema Registry 服务器 API
Schema Registry Server API 由以下作组成:
-
POST /
— 参见Registering a New Schema
-
GET /{subject}/{format}/{version}
— 参见Retrieving an Existing Schema by Subject, Format, and Version
-
GET /{subject}/{format}
— 参见Retrieving an Existing Schema by Subject and Format
-
GET /schemas/{id}
— 参见Retrieving an Existing Schema by ID
-
DELETE /{subject}/{format}/{version}
— 参见Deleting a Schema by Subject, Format, and Version
-
DELETE /schemas/{id}
— 参见Deleting a Schema by ID
-
DELETE /{subject}
— 参见Deleting a Schema by Subject
注册新架构
要注册新 Schema,请发送POST
请求发送到终端节点。/
接受包含以下字段的 JSON 有效负载:/
-
subject
:架构主题 -
format
:架构格式 -
definition
:架构定义
其响应是 JSON 格式的架构对象,具有以下字段:
-
id
:架构 ID -
subject
:架构主题 -
format
:架构格式 -
version
:架构版本 -
definition
:架构定义
按主题、格式和版本检索现有架构
要按主题、格式和版本检索现有架构,请发送GET
请求发送到{subject}/{format}/{version}
端点。
其响应是 JSON 格式的架构对象,具有以下字段:
-
id
:架构 ID -
subject
:架构主题 -
format
:架构格式 -
version
:架构版本 -
definition
:架构定义
按主题和格式检索现有架构
要按主题和格式检索现有架构,请发送GET
请求发送到/subject/format
端点。
其响应是一个架构列表,其中包含 JSON 中的每个架构对象,其中包含以下字段:
-
id
:架构 ID -
subject
:架构主题 -
format
:架构格式 -
version
:架构版本 -
definition
:架构定义
按 ID 检索现有架构
要按 ID 检索 Schema,请发送GET
请求发送到/schemas/{id}
端点。
其响应是 JSON 格式的架构对象,具有以下字段:
-
id
:架构 ID -
subject
:架构主题 -
format
:架构格式 -
version
:架构版本 -
definition
:架构定义
按主题删除 Schema
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 对于 storage 表。
任何升级的 Spring Cloud Stream 1.1.0.RELEASE 用户都应在升级之前将其现有 schema 迁移到新表。 |
使用 Confluent 的 Schema Registry
默认配置会创建一个DefaultSchemaRegistryClient
豆。
如果要使用 Confluent 模式注册表,则需要创建一个ConfluentSchemaRegistryClient
,它将取代框架默认配置的 URL。下面的示例展示了如何创建这样的 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 进行了测试。 |
Schema 注册过程 (序列化)
注册过程的第一部分是从通过通道发送的有效负载中提取架构。
Avro 类型,例如SpecificRecord
或GenericRecord
已包含架构,可以立即从实例中检索该架构。
对于 POJO,如果spring.cloud.stream.schema.avro.dynamicSchemaGenerationEnabled
属性设置为true
(默认值)。
获取 schema 后,转换器将从远程服务器加载其元数据(版本)。 首先,它查询本地缓存。如果未找到结果,它将数据提交到服务器,服务器会回复版本控制信息。 转换器始终缓存结果,以避免在 Schema Server 中查询需要序列化的每条新消息的开销。
使用架构版本信息,转换器将contentType
标头来携带版本信息——例如:application/vnd.user.v1+avro
.
Schema 解析过程 (反序列化)
读取包含版本信息的消息(即contentType
标头,其方案如下所述Schema Registration Process (Serialization)
中,转换器将查询 Schema 服务器以获取消息的写入器 Schema。
找到传入消息的正确架构后,它会检索读取器架构,并使用 Avro 的架构解析支持将其读取到读取器定义中(设置默认值和任何缺少的属性)。
您应该了解编写器架构(编写消息的应用程序)和读取器架构(接收应用程序)之间的区别。
我们建议花点时间阅读 Avro 术语并了解该过程。
Spring Cloud Stream 始终获取写入器架构以确定如何读取消息。
如果你想让 Avro 的架构演变支持正常工作,你需要确保readerSchema 已为您的应用程序正确设置。 |