消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读本文聚焦 Apache Pulsar 内置的Schema RegistrySchema 注册中心系统讲解它在 Pulsar 消息系统中解决“类型安全Type Safety”问题的核心机制Schema 如何在 topic 级别上传、存储、校验与版本化客户端如何基于 Schema 进行序列化/反序列化以及如何通过 REST API 与pulsar-admin命令行工具管理 Schema。读完本文你将掌握 Pulsar Schema Registry 的架构与工作流程、受支持的全部 Schema 格式、Schema 版本演进规则以及从 Java 客户端到管理命令的完整实战用法。适用版本说明本文以当前仓库site2/website-next/versioned_docs/version-2.2.0/下的 2.2.0 版本文档为主体并结合仓库源码中 Schema Registry 相关实现进行深化文中涉及的 API 与命令行为以仓库当前实际实现为准。为什么需要 Schema Registry消息系统中的类型安全在围绕 Pulsar 这类消息总线构建的任何应用里类型安全都极其重要。Producer 与 Consumer 需要在 topic 级别协调数据类型否则会引发一系列潜在问题——最典型的就是序列化/反序列化错误。消息在 Pulsar 中本质上是原始字节流如果 Producer 在topic-1上发送温度传感器数据而该 topic 的 Consumer 却试图把数据解析成湿度传感器读数就会立刻出错。围绕消息的类型安全应用通常采用两种基本思路“客户端侧client-side”方案Producer 和 Consumer 不仅要负责消息原始字节的序列化与反序列化还要自行“知晓”哪个 topic 传输哪种类型。这种方案把所有类型安全的维护负担都交给了应用层即所谓的 out-of-band带外管理。“服务端侧server-side”方案Producer 和 Consumer 主动告知系统某个 topic 可以传输哪些数据类型。消息系统负责强制类型安全确保 Producer 与 Consumer 始终保持同步。Pulsar 对两种方案都支持你可以自由选择其中一种也可以在逐 topic 粒度上混用采用“客户端侧”方案时Producer 与 Consumer 可以发送/接收由原始字节数组构成的消息把全部类型安全交给应用在带外处理采用“服务端侧”方案时Pulsar 内置的Schema Registry允许客户端按 topic 上传数据 Schema这些 Schema 决定了该 topic 上哪些数据类型被视为合法。注意在 2.2.0 版本中Pulsar Schema Registry 仅对 Java 客户端、CGo 客户端、Python 客户端 和 C 客户端 可用。基本架构Schema 如何进入注册中心从架构上看Pulsar Schema Registry 的写入与读取遵循以下路径当你使用 Schema 创建带类型的 Producer时Schema 会被自动上传你也可以通过 Pulsar 的 REST API/admin/v2/schemas系列端点手动上传、获取和更新Schema。在存储层面Pulsar 开箱即用地使用Apache BookKeeper日志存储系统作为 Schema 的持久化后端对应 Pulsar 架构中的 persistent storage 层。如果你愿意也可以接入其他存储后端——2.2.0 文档说明自定义 Schema 存储逻辑的文档“即将推出”因此本文以默认的 BookKeeper 后端为准。从源码看这一架构在 broker 侧由SchemaRegistryService接口统一抽象其实现通过工厂方法创建SchemaRegistryService.create(SchemaStorage, SetString checkerClasses)会先构建SchemaType - SchemaCompatibilityCheck的兼容性检查器映射并为KEY_VALUE类型装配KeyValueSchemaCompatibilityCheck随后用SchemaRegistryServiceWithSchemaDataValidator包装真实的SchemaRegistryServiceImpl当 Schema 存储创建失败时则退化为DefaultSchemaRegistryService空实现相关逻辑见 SchemaRegistryService.java。Schema 如何工作数据模型与应用范围Pulsar Schema 是相当简单的数据结构由以下部分组成组成部分说明Name名称在 Pulsar 中Schema 的名称就是它被应用的那个 topicPayload负载Schema 的二进制表示Type类型Schema 的格式类型见下文“支持的 Schema 格式”Properties属性用户自定义的 string/string 映射。其用法完全取决于具体应用常见的例子包括关联的 Git hash、环境标识如dev、prod等两个关键约束必须牢记Schema 只作用于 topic 级别不能应用到 namespace 或 tenant 级别Producer 和 Consumer 都是把 Schema上传给 Pulsar broker由 broker 统一存储与校验。这个“topic 级”数据模型在 broker 实现中体现得非常直接SchemaRegistryServiceImpl的所有读写方法都以schemaId为键而 schemaId 正是 topic 名putSchemaIfAbsent、getSchema、getAllSchemas、deleteSchema等操作全部围绕schemaId展开参见 SchemaRegistryServiceImpl.java。Schema 版本机制三种连接场景逐项剖析Schema 版本化是 Pulsar Schema Registry 最核心的机制。为了说明其工作原理先看一个 Java 客户端创建带 Schema 的 Producer 的示例PulsarClient client PulsarClient.builder() .serviceUrl(pulsar://localhost:6650) .build(); ProducerSensorReading producer client.newProducer(JSONSchema.of(SensorReading.class)) .topic(sensor-data) .sendTimeout(3, TimeUnit.SECONDS) .create();当这个 Producer 尝试连接 broker 时可能出现以下三种场景及其对应处理场景会发生什么topic 上不存在任何 SchemaProducer 以给定 Schema 创建。Schema 被传送到 broker 并存储因为没有已有 Schema 与SensorReading兼容。任何使用相同 Schema/topic 创建的 Consumer 都可以消费sensor-datatopic 上的消息已存在SchemaProducer 使用相同Schema 连接Schema 被传送到 broker。broker 判定该 Schema 是兼容的尝试将其存储到 BookKeeper但随后发现它已经存储过于是该 Schema 被用来为生产的消息打上标签version已存在SchemaProducer 使用新的、兼容的Schema 连接Producer 将 Schema 传送给 broker。broker 判定其兼容将新 Schema 存储为当前版本获得新的版本号版本分配规则Schema 按先后顺序进行版本化。Schema 的存储发生在处理对应 topic 的 broker 上以便分配版本号。一旦某个 Schema 被分配/获取了版本该 Producer 后续生产的所有消息都会被标记上相应的版本号。源码层面印证了这套流程putSchemaIfAbsent会先查询已存在的 Schema 列表若发现相同 Schema通过 SHA-256 哈希比对见SchemaRegistryServiceImpl中的hashFunction Hashing.sha256()与SchemaHash.of(...)则直接返回已有版本否则按兼容性策略校验通过后把新 Schema 连同类型、用户、时间戳、属性等写入存储并生成新版本。兼容性校验失败的路径会抛出IncompatibleSchemaException例如“已存在 schema 类型 X新 schema 类型 Y”参见 SchemaRegistryServiceImpl.java。支持的 Schema 格式Pulsar Schema Registry 支持以下格式Schema 类型说明None若 topic 未指定 SchemaProducer 和 Consumer 直接处理原始字节String用于 UTF-8 编码的字符串JSONJSON 对象编码与校验ProtobufProtocol Buffers 消息编码与解码Avro通过 Avro 进行序列化/反序列化对应地客户端 API 的SchemaType枚举定义了这些类型的序号NONE0, STRING1, JSON2, PROTOBUF3, AVRO4并注释说明新增需要记录进 Schema Registry 的类型时应同步修改pulsar-common/src/main/proto/PulsarApi.proto与pulsar-broker/src/main/proto/SchemaRegistryFormat.proto两个 proto 文件见 SchemaType.java。2.2.0 文档同时指出其他 Schema 格式的支持将在未来版本中陆续加入。从当前仓库的SchemaType枚举看后续版本已扩展了BOOLEAN、INT8/16/32/64、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、BYTES等更多内建类型标记为since 2.3.0等这印证了文档的“未来扩展”方向。实战示例用 RecordSchemaBuilder 构建 Avro Schema 并消费 GenericRecord下面的例子演示如何用RecordSchemaBuilder定义 Avro Schema、用GenericRecordBuilder生成通用 Avro 记录并把消息消费为GenericRecord。第 1 步用RecordSchemaBuilder构建 SchemaRecordSchemaBuilder recordSchemaBuilder SchemaBuilder.record(schemaName); recordSchemaBuilder.field(intField).type(SchemaType.INT32); SchemaInfo schemaInfo recordSchemaBuilder.build(SchemaType.AVRO); ProducerGenericRecord producer client.newProducer(Schema.generic(schemaInfo)).create();第 2 步用GenericRecordBuilder构建并发送通用记录producer.newMessage().value(schema.newRecordBuilder() .set(intField, 32) .build()).send();这段代码中涉及的核心 API 都定义在 pulsar-client-api 模块下RecordSchemaBuilder用于按字段描述 Schema支持field(name).type(SchemaType)链式声明SchemaBuilder.record(name)创建记录型 Schema 构建器Schema.generic(schemaInfo)生成可处理GenericRecord的通用 SchemaGenericRecordBuilder则用于按字段名设置值并产出GenericRecord。这是“带类型 Producer 服务端 Schema 校验”组合的典型用法Producer 端定义并注册 Schemabroker 端依据该 Schema 校验后续消息。管理 SchemaREST API 与 pulsar-admin 命令你可以使用 Pulsar 管理工具pulsar-admin或 REST API 管理 topic 的 Schema。REST API 端点Broker 侧的v2管理端点定义在 SchemasResource.java基类实现见 SchemasResourceBase.java路径前缀为/schemas方法路径功能GET/schemas/{tenant}/{namespace}/{topic}/schema获取 topic 的最新 SchemaGET/schemas/{tenant}/{namespace}/{topic}/schema/{version}获取指定版本的 SchemaGET/schemas/{tenant}/{namespace}/{topic}/schemas获取 topic 的全部各版本SchemaPOST/schemas/{tenant}/{namespace}/{topic}/schema上传/更新 Schema请求体为PostSchemaPayloadDELETE/schemas/{tenant}/{namespace}/{topic}/schema删除 topic 的最新 Schemapulsar-admin schemas 子命令命令行入口位于 CmdSchemas.java命令组为pulsar-admin schemas支持以下四个子命令1.get获取 topic 的 Schemapulsar-admin schemas get persistent://tenant/namespace/topic参数persistent://tenant/namespace/topic必填可选参数-v, --version指定版本号必须大于 0-a, --all-version列出全部版本--version与--all-version不能同时指定。不带任何可选参数时默认输出最新 Schema 及其版本信息。2.delete删除 topic 的最新 Schemapulsar-admin schemas delete persistent://tenant/namespace/topic对应getAdmin().schemas().deleteSchema(topic)删除操作在 broker 端会写入一条标记deletedtrue的 Schema 记录因此删除后该 topic 的旧版本 Schema 无法再被获取参见SchemaRegistryServiceImpl.getSchema对isDeleted()的过滤逻辑。3.upload为 topic 上传/更新 Schemapulsar-admin schemas upload persistent://tenant/namespace/topic -f /path/to/schema.json-f, --filename必填包含PostSchemaPayload即 type、schema、properties 等字段的 JSON 文件路径内部实现是把文件内容解析为PostSchemaPayload后调用createSchema(topic, input)。4.extract从 JAR 中提取 POJO 的 Schema 并上传pulsar-admin schemas extract persistent://tenant/namespace/topic \ -j /path/to/pojo.jar -t avro -c com.example.SensorReading-j, --jar必填包含 POJO 类的 JAR 文件路径-t, --type必填提取类型仅支持avro或json-c, --classname必填POJO 类全限定名--always-allow-null设置 Schema 是否始终允许 null默认true-n, --dry-run只打印将上传的PostSchemaPayloadJSON 形式不真正应用到 Schema Registry。该命令通过URLClassLoader加载 JAR 中的 POJO再借助SchemaExtractor生成 Avro/JSON Schema是“从现有 Java 类快速注册 Schema”的实用工具。与消息生产/消费流程的联动理解 Schema Registry 后还需要把它放回消息生命周期中看待。从 broker 侧源码看Schema 的注册与校验深度嵌入在消息处理链路中生产者Producer侧ServerCnx在处理 producer 创建请求时调用 SchemaRegistryService 完成 Schema 注册与兼容性检查消费者Consumer侧checkConsumerCompatibility依据兼容性策略校验消费者携带的 Schema 与已存 Schema 是否匹配ALWAYS_COMPATIBLE策略直接放行其余策略按 BACKWARD/FORWARD/FULL 等规则与最新版或全部历史版本比对见 SchemaRegistryServiceImpl.java。这样Producer 上传并版本化 Schema、消息被打上 Schema 版本标签、Consumer 按兼容策略校验三者构成闭环类型安全由 broker 强制执行Producer 与 Consumer 无需在带外自行约定类型——这正是本文开头“服务端侧方案”的完整落地。小结Pulsar Schema Registry 为消息总线场景下的类型安全提供了一条服务端强制的路径Schema 在 topic 级别注册由 broker 统一存储于 BookKeeper 并按顺序版本化Producer 上传 Schema 后其生产的消息被标记上对应版本Consumer 通过兼容性策略与已存 Schema 对齐。配合pulsar-admin schemas命令与 REST API你可以完整地管理 Schema 的生命周期查询、上传、更新、删除。在实际项目中可结合本文示例从 Java 客户端侧开启 Schema如JSONSchema.of(Class)或RecordSchemaBuilderSchema.generic(...)并在后续演进 Schema 时充分理解版本兼容规则避免不兼容变更导致生产链路中断。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar Schema Registry 完整指南主题级类型安全、Schema 版本管理与多格式支持Apache Pulsar Schema Registry 完整指南主题级类型安全、Schema 版本管理与多格式支持 Apache Pulsar 内置的 S消息队列后端流处理Apache Pulsar Schema 入门理解 Schema Registry、类型安全与生产消费实战Apache Pulsar Schema 入门理解 Schema Registry、类型安全与生产消费实战 本指南以 Apache Pulsar 的 Sche消息队列后端流处理Apache Pulsar Schema 入门指南Schema Registry、类型安全与首个带 Schema 的 Java 客户端Apache Pulsar Schema 入门指南Schema Registry、类型安全与首个带 Schema 的 Java 客户端 Apache Puls消息队列后端流处理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考