消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本指南以 Apache Pulsar 官方 Cookbook《BookKeeper Ledger Metadata》为主体深入讲解 Pulsar 如何通过 BookKeeper ledger 自定义元数据metadata标记每一份数据卷的用途与归属。读完本文你将掌握 ledger 元数据在 ZooKeeper 中的存储形态、每一类元数据字段的语义与取值以及如何通过 BookKeeper API 读取这些元数据来排查与运维 Pulsar 的存储层。一、背景为什么需要查看 Ledger 元数据Pulsar 的所有消息数据最终都落盘在 BookKeeper 的 ledger 上。一个运行中的 Pulsar 集群会同时维护大量 ledger它们各自承担不同职责有的是主题topic数据的 managed ledger有的是消费游标cursor持久化状态有的是 topic 压缩compaction产生的专用 ledger还有的是 Schema 存储 ledger。仅凭一个数值型的 ledger id运维人员很难判断这个 ledger 到底存的是什么。为此Pulsar 在创建每一个 ledger 时都会附加一组自定义元数据custom metadata将 ledger 的应用归属、组件类型、对应实体名称等信息写入其中。这些元数据保存在 ZooKeeper 上并且可以通过 BookKeeper 的标准 API 读取出来——这正是排查数据、分析存储分布、诊断压缩或游标问题时的重要入口。说明本指南对应的官方文档版本为 2.3.0见 cookbooks-bookkeepermetadata.md而当前仓库主干版本为 2.10.6-SNAPSHOT见 pom.xml。文中会同时给出两版信息并以当前源码为准标注差异。二、元数据存在哪里ZooKeeper BookKeeper API原文档明确指出Pulsar 将数据存储在 BookKeeper ledgers 上你可以通过检查 ledger 附加的元数据来理解该 ledger 的内容这些元数据存储在 ZooKeeper 上它们可以使用 BookKeeper API 读取。在 BookKeeper 的存储模型中每个 ledger 的元数据包括创建时间、ensemble、写入 quorum、ack quorum、digest 类型以及自定义属性都会持久化到 ZooKeeper 的元数据节点中。Pulsar 在调用asyncCreateLedger时会通过metadata参数把自定义属性一并写入随后这些属性便与 ledger 本体绑定可通过LedgerHandle.getLedgerMetadata()获取。从当前源码看Pulsar 暴露了一条便捷的查询链路ManagedLedgerImpl.getLedgerMetadata(long ledgerId)见 ManagedLedgerImpl.java——对于当前活跃的 ledger 直接返回currentLedger.getLedgerMetadata().toSafeString()对于已滚动的旧 ledger 则通过getLedgerHandle(ledgerId)打开后读取其元数据。该查询结果也会出现在getManagedLedgerInternalStats(boolean includeLedgerMetadata)的管理统计中见 ManagedLedgerImpl.java便于管理员通过 Pulsar Admin 接口直接审视每个 ledger 的元数据内容。三、当前元数据字段总览原文档给出了当时2.3.0全部元数据字段的权威说明下表完整继承并补充了取值说明作用域Scope元数据名Metadata name元数据值Metadata value所有 ledgerapplicationpulsar所有 ledgercomponentmanaged-ledger、schema、compacted-topicManaged ledgerspulsar/managed-ledgerledger 的名称name of the ledgerCursorpulsar/cursor游标名称name of the cursorCompacted topicpulsar/compactedTopic原始主题名称name of the original topicCompacted topicpulsar/compactedTo最后一条已压缩消息的 idid of the last compacted message其中application与component是每个 Pulsar 创建的 ledger 都会携带的基础字段用于标识这是 Pulsar 写的数据、属于哪个组件pulsar/managed-ledger标记该 ledger 属于哪一个 managed ledger即哪个 topicpulsar/cursor标记游标持久化 ledger 对应的游标名pulsar/compactedTopic与pulsar/compactedTo仅出现在压缩compaction产生的 ledger 上分别记录被压缩的原始主题和压缩完成后最后一条消息的位置。四、源码级实现LedgerMetadataUtils当前仓库中所有这些元数据键名与构造逻辑都集中在 LedgerMetadataUtils.java 内。该类是Utilities for managing BookKeeper Ledgers custom metadata定义了全部键常量常量键名用途METADATA_PROPERTY_APPLICATIONapplication应用标识固定为pulsarMETADATA_PROPERTY_COMPONENTcomponent组件标识METADATA_PROPERTY_MANAGED_LEDGER_NAMEpulsar/managed-ledgermanaged ledger 名称METADATA_PROPERTY_CURSOR_NAMEpulsar/cursor游标名称METADATA_PROPERTY_COMPACTEDTOPICpulsar/compactedTopic被压缩的原始主题METADATA_PROPERTY_COMPACTEDTOpulsar/compactedTo最后一条已压缩消息 idMETADATA_PROPERTY_SCHEMAIDpulsar/schemaIdSchema id新增于 2.3.0 之后的版本版本差异提示原文档将component的取值列举为managed-ledger、schema、compacted-topic而在当前源码2.10.6-SNAPSHOT中压缩 ledger 的组件值实际写作compacted-ledger见 LedgerMetadataUtils.java并且新增了pulsar/schemaId这一键见 LedgerMetadataUtils.java。排查时请以实际部署版本的取值为准。4.1 各类 ledger 的元数据构造LedgerMetadataUtils提供了四个核心构造方法分别对应原文档表格中的各个作用域1. Managed ledger 基础元数据static MapString, byte[] buildBaseManagedLedgerMetadata(String name) { return ImmutableMap.of( METADATA_PROPERTY_APPLICATION, METADATA_PROPERTY_APPLICATION_PULSAR, // applicationpulsar METADATA_PROPERTY_COMPONENT, METADATA_PROPERTY_COMPONENT_MANAGED_LEDGER, // componentmanaged-ledger METADATA_PROPERTY_MANAGED_LEDGER_NAME, name.getBytes(StandardCharsets.UTF_8)); // pulsar/managed-ledgername }该方法在ManagedLedgerImpl构造时即被调用成为该 managed ledger 一切 ledger 的默认元数据见 ManagedLedgerImpl.java因此每个 managed ledger 下的数据 ledger 都天然携带applicationpulsar、componentmanaged-ledger、pulsar/managed-ledgerledger名三组键值。2. Cursor 附加元数据static MapString, byte[] buildAdditionalMetadataForCursor(String name) { return ImmutableMap.of(METADATA_PROPERTY_CURSOR_NAME, name.getBytes(StandardCharsets.UTF_8)); }游标cursor在创建自己的持久化 ledger 时调用该方法将游标名写入pulsar/cursor见 ManagedCursorImpl.java。这样游标 ledger 既带有 managed ledger 的基础属性又额外标明自己属于哪个游标。3. 压缩 ledger 元数据public static MapString, byte[] buildMetadataForCompactedLedger(String compactedTopic, byte[] compactedToMessageId) { return ImmutableMap.of( METADATA_PROPERTY_APPLICATION, METADATA_PROPERTY_APPLICATION_PULSAR, METADATA_PROPERTY_COMPONENT, METADATA_PROPERTY_COMPONENT_COMPACTED_LEDGER, // componentcompacted-ledger METADATA_PROPERTY_COMPACTEDTOPIC, compactedTopic.getBytes(StandardCharsets.UTF_8), METADATA_PROPERTY_COMPACTEDTO, compactedToMessageId ); }topic 压缩分两阶段执行第二阶段phaseTwo在创建压缩 ledger 时调用此方法传入原始 topic 名与to压缩后的最后消息 id序列化后的字节数组见 TwoPhaseCompactor.java。这正是原文档表格中pulsar/compactedTopic与pulsar/compactedTo两行数据的真实来源。4. Schema ledger 元数据public static MapString, byte[] buildMetadataForSchema(String schemaId) { return ImmutableMap.of( METADATA_PROPERTY_APPLICATION, METADATA_PROPERTY_APPLICATION_PULSAR, METADATA_PROPERTY_COMPONENT, METADATA_PROPERTY_COMPONENT_SCHEMA, // componentschema METADATA_PROPERTY_SCHEMAID, schemaId.getBytes(StandardCharsets.UTF_8) ); }Broker 侧的 Schema 存储BookkeeperSchemaStorage在createLedger(String schemaId)中调用该方法构造元数据并随bookKeeper.asyncCreateLedger(...)的metadata参数一并写入 ZooKeeper见 BookkeeperSchemaStorage.java。5. 放置策略配置元数据此外当前版本还支持通过buildMetadataForPlacementPolicyConfig将EnsemblePlacementPolicyConfig编码进 ledger 元数据键名为EnsemblePlacementPolicyConfig见 LedgerMetadataUtils.java 与 EnsemblePlacementPolicyConfig.java用于在 topic 级别定制 ledger 的放置策略。五、如何实际读取元数据由于元数据随 ledger 一起持久化在 ZooKeeper 中读取方式与普通 BookKeeper 客户端一致打开目标 ledger 后通过LedgerHandle.getLedgerMetadata()拿到LedgerMetadata对象再读取其自定义属性。Pulsar 内部也复用了这一机制——例如ManagedLedgerImpl.getLedgerMetadata(ledgerId)返回rh.getLedgerMetadata().toSafeString()的 JSON 文本见 ManagedLedgerImpl.java其中便包含上文所有的application、component、pulsar/...键值对。以压缩场景为例CompactedTopicTest中的测试直接展示了打开压缩 ledger 并校验其元数据属性的流程先用bk.createLedger(...)创建 ledger再通过bk.openLedger(ledgerId, ...)打开并读取见 CompactedTopicTest.java。实际运维排查时可参照同样的思路通过pulsar-admin topics stats-internal拿到 topic 内部统计中的 ledger id 列表getManagedLedgerInternalStats支持includeLedgerMetadatatrue直接返回元数据文本见 ManagedLedgerImpl.java用 BookKeeper 客户端或bookkeeper shell ledgermetadata ledgerId打开对应 ledger读取LedgerMetadata.getCustomMetadata()/toSafeString()输出按本文表格中的键名对照解读。六、元数据的运维实践价值理解这些元数据后你可以获得以下实际的排查与运维能力快速识别 ledger 归属component字段直接告诉你一个陌生 ledger 是 managed-ledger、schema 还是压缩产物无需猜测定位主题数据pulsar/managed-ledger把 ledger 与具体 topicledger 名精确对应方便做存储分布统计与数据迁移核对追踪游标状态pulsar/cursor让游标持久化 ledger 与其消费游标一一对应可用于排查游标堆积、回溯消费位置审计压缩结果pulsar/compactedTopic与pulsar/compactedTo记录了压缩覆盖的原始主题与最后压缩位置可验证压缩任务是否按预期完成识别 Schema 存储较新版本中pulsar/schemaId帮助区分 Schema 专用 ledger。需要注意的是ledger 元数据在创建时一次性写入 ZooKeeper属于静态描述信息它与消息条数、字节大小等运行时统计不同适合作为这是什么的定性依据而不适合作为实时监控指标。结合getManagedLedgerInternalStats输出的 ensemble、quorum 等存储布局信息一起分析可以获得对 Pulsar 存储层更完整的认知。七、小结本文完整覆盖了官方 Cookbook《BookKeeper Ledger Metadata》的全部内容元数据存放于 ZooKeeper、可通过 BookKeeper API 读取并详细列出application、component、pulsar/managed-ledger、pulsar/cursor、pulsar/compactedTopic、pulsar/compactedTo六类键的语义。在此基础上我们从当前仓库源码 LedgerMetadataUtils.java 出发还原了每一类元数据的构造时机与调用链ManagedLedger 创建、游标持久化、两阶段压缩、Schema 存储并给出了可落地的读取与排查方法。掌握这套元数据体系你就能在 Pulsar 存储层排查中多一把精准的放大镜。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar BookKeeper Ledger 元数据完全指南如何通过 ZooKeeper 与 BookKeeper API 解读数据存储结构Apache Pulsar BookKeeper Ledger 元数据完全指南如何通过 ZooKeeper 与 BookKeeper API 解读数据存储结构消息队列后端流处理Apache Pulsar BookKeeper Ledger 元数据全解析如何从 ZooKeeper 中读懂数据存储结构Apache Pulsar BookKeeper Ledger 元数据全解析如何从 ZooKeeper 中读懂数据存储结构 Apache Pulsar 的所有消息队列后端流处理Apache Pulsar 配置完全指南从 BookKeeper 到 ZooKeeper 的 conf 参数深度解析Apache Pulsar 配置完全指南从 BookKeeper 到 ZooKeeper 的 conf 参数深度解析 Apache Pulsar 是一个分布式消息队列后端流处理创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考