消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载本文基于 Apache Pulsar 官方文档《Pulsar geo-replication》仓库路径 site2/website-next/versioned_docs/version-2.4.0/administration-geo.md系统讲解 Pulsar 地理复制的核心原理、租户/命名空间级配置流程、选择性复制与复制订阅Replicated Subscriptions等高级能力。读完本文你将掌握如何用pulsar-admin在多集群间搭建复制通道、如何在客户端精确控制消息的复制范围以及如何在故障切换场景下让消费者从失败点无缝续读。适用说明本文命令与配置以当前仓库Apache Pulsar 2.4.0 版本线为准bin/pulsar-admin等命令行工具基于发行版目录结构与源码仓库中的 pulsar-client-tools 模块对应。地理复制是什么地理复制Geo-replication是指在一个 Pulsar 实例Pulsar instance由多个相互通信的集群组成内将持久化存储的消息数据在多个集群之间进行复制。它是一种跨集群、异步、面向租户/命名空间的复制机制常用于多地域部署、容灾备份、就近读写等场景。核心工作流程在上图所示的三集群场景中P1、P2、P3三个生产者分别在Cluster-A、Cluster-B、Cluster-C上向T1主题发布消息消息发布后会被立即复制到其他集群C1、C2消费者可以从各自所在集群消费到所有生产者包括 P3发布的消息。如果没有地理复制C1 和 C2 将无法消费 P3 在 Cluster-C 上发布的消息——这正是地理复制要解决的核心问题让任意集群的写入对全局可见。从实现层面看Pulsar 在每个 broker 内部维护了一个面向远程集群的复制器Replicator。仓库源码中该机制由 AbstractReplicator抽象基类及其两个子类实现PersistentReplicator用于持久化主题persistent topic是地理复制的默认实现NonPersistentReplicator用于非持久化主题。从 AbstractReplicator 的构造函数可以看到每个复制器对应一条「本地集群 → 远程集群」方向的消息流它会基于本地主题创建复制生产者replicator producer并以本地集群 -- 远程集群的格式命名源码中常量REPL_PRODUCER_NAME_DELIMITER --。复制生产者的队列深度由 broker 配置项replicationProducerQueueSize决定代码中通过getReplicationProducerQueueSize()读取。在 PersistentReplicator 中复制器通过一个独立的 ManagedCursor 从本地 ManagedLedger 读取消息readEntries中先执行cursor.rewind()确保重启后不丢消息随后激活 cursor 并持续读取条目getReplicatorReadPosition()返回cursor.getMarkDeletedPosition()即本地复制光标已确认删除的位置——这个位置正是「已复制到远程集群」的进度标记。本地持久化与异步转发消息在 Pulsar 主题上被生产时首先持久化到本地集群然后异步转发到远程集群正常网络状况下消息在派发给本地消费者的同时即被复制端到端投递延迟主要由跨地域的网络往返时间RTT决定即使在远程集群不可达例如网络分区时应用依然可以在任意集群创建生产者和消费者——复制通道会按需重连期间消息继续在本地持久化不会丢失。复制粒度命名空间而非主题地理复制必须按租户tenant启用只有当某个租户被允许访问两个集群时才可以在它们之间启用复制。虽然启用动作发生在两个集群之间但复制的实际管理粒度是命名空间namespace。要为一个命名空间启用地理复制需要完成两步启用地理复制命名空间将该命名空间配置为在两个或更多已开通的集群之间复制。该命名空间下任何主题上发布的消息都会被复制到指定集群集合中的所有集群。配置地理复制如前面所述地理复制由 Pulsar 的租户级别管理。下面按操作顺序展开。为租户授予集群权限要复制到某个集群租户必须拥有使用该集群的权限。你可以在创建租户时授予也可以事后更新。创建租户并指定允许的集群$ bin/pulsar-admin tenants create my-tenant \ --admin-roles my-admin-role \ --allowed-clusters us-west,us-east,us-cent参数说明my-tenant租户名称--admin-roles my-admin-role授予该租户的管理员角色--allowed-clusters us-west,us-east,us-cent该租户允许使用的集群列表以逗号分隔必须包含所有计划参与复制的集群。若要更新已有租户的权限把create换成update即可bin/pulsar-admin tenants update ...。启用地理复制命名空间创建命名空间$ bin/pulsar-admin namespaces create my-tenant/my-namespace刚创建时命名空间未绑定到任何集群。使用set-clusters子命令将其指派给多个集群$ bin/pulsar-admin namespaces set-clusters my-tenant/my-namespace \ --clusters us-west,us-east,us-cent命名空间的复制集群集合可以随时更改且不中断正在进行的流量——配置变更后所有集群中的复制通道会立即建立或停止。这一点很关键扩缩容复制范围是动态热操作无需重启 broker也无需重新创建主题。在地理复制命名空间中使用主题创建好地理复制命名空间后只要生产者和消费者在该命名空间内创建主题主题就会自动跨集群复制。典型用法是每个应用使用本地集群的serviceUrl连接读写就近完成数据由后台复制机制同步到远端。选择性复制Selective replication默认情况下消息会复制到命名空间配置的所有集群。你可以通过为单条消息指定复制列表replication list来限制复制范围——该消息将只复制到列表中的集群子集。以下为 Java API 示例注意构造Message对象时使用setReplicationClusters方法ListString restrictReplicationTo Arrays.asList( us-west, us-east ); Producer producer client.newProducer() .topic(some-topic) .create(); producer.newMessage() .value(my-payload.getBytes()) .setReplicationClusters(restrictReplicationTo) .send();该能力位于客户端 API 的消息构建链上producer.newMessage()返回MessageBuilder其中提供了setReplicationClusters(ListString)用于设置这条消息的复制范围。注意复制列表中的集群名称必须与租户的allowed-clusters及命名空间的复制集群配置一致否则复制目标不会生效。查看主题统计Topic stats地理复制主题的统计信息可通过pulsar-admin工具及 REST API 获取$ bin/pulsar-admin persistent stats persistent://my-tenant/my-namespace/my-topic每个集群各自上报本地统计包括入站复制速率与积压incoming replication rates and backlogs远程集群复制到本集群的速度与未消费积压出站复制速率与积压outgoing replication rates and backlogs本集群复制到远程集群的速度与积压。在实现上这些统计由复制器内部的ReplicatorStatsImpl采集见 PersistentReplicator 中的private final ReplicatorStatsImpl statsbroker 配置项replicationMetricsEnabled默认true控制是否导出复制相关指标。删除地理复制主题由于地理复制主题同时存在于多个地域无法直接删除某个地理复制主题而应依赖自动主题垃圾回收automatic topic garbage collection。Pulsar 中一个主题在满足以下三个条件时会被自动删除没有生产者或消费者连接它没有任何订阅没有更多消息需要保留retention。对地理复制主题而言每个地域都会使用一种容错机制来决定何时可以在本地安全删除该主题。你可以通过将 broker 配置中的brokerDeleteInactiveTopicsEnabled设为false来显式禁用主题垃圾回收完整 broker 配置项见 reference-configuration.md。要删除一个地理复制主题请执行关闭该主题上的所有生产者和消费者在每个复制集群中删除该主题的所有本地订阅当 Pulsar 判定整个系统中该主题不再存在有效订阅时会自动对其进行垃圾回收。复制订阅Replicated subscriptionsPulsar 支持复制订阅对于一个正在跨多个地理区域异步复制的主题可以将订阅状态在亚秒级时间内保持同步。这样故障切换failover时消费者可以在不同集群从失败点继续消费。启用复制订阅复制订阅默认关闭需要在创建消费者时显式开启ConsumerString consumer client.newConsumer(Schema.STRING) .topic(my-topic) .subscriptionName(my-subscription) .replicateSubscriptionState(true) .subscribe();该方法定义在客户端 API 的 ConsumerBuilder 上ConsumerBuilderT replicateSubscriptionState(boolean)任何语言客户端Java、C、Python、Go 等在使用该主题时均需设置此选项订阅状态同步才会生效。底层原理快照机制从源码看复制订阅的核心实现在 ReplicatedSubscriptionsController每个持久化主题一个实例。它通过 broker 执行器定时触发startNewSnapshot按replicatedSubscriptionsSnapshotFrequencyMillis配置的周期默认 1 秒生成分布式一致性快照建立各集群消息 ID 之间的关联映射。同步过程基于 Pulsar 的**标记消息marker message**机制涉及四类标记源码中对应MarkerType枚举REPLICATED_SUBSCRIPTION_SNAPSHOT_REQUEST发起快照请求REPLICATED_SUBSCRIPTION_SNAPSHOT_RESPONSE返回快照响应携带本集群最新写入位置lastMsgIdREPLICATED_SUBSCRIPTION_UPDATE广播订阅更新localSubscriptionUpdated中构造ReplicatedSubscriptionsUpdate标记并写入快照完成后各集群的订阅游标位置被统一到同一基线。每份快照中记录各集群对应的消息 IDClusterMessageId消费者故障切换后即可依据这份映射在另一集群定位到上次消费位置。优势逻辑实现简单快照与标记机制由 broker 内部完成业务代码无需额外逻辑可按需开关你可以选择启用或禁用复制订阅启用时开销低、易配置仅需在创建消费者时加一个布尔参数禁用时零开销默认关闭不影响常规消费路径。限制存在亚秒级重复消费窗口启用复制订阅时系统会定期创建一致性分布式快照将不同集群的消息 ID 关联起来。快照默认每1 秒取一次因此消费者故障切换到另一集群后可能重复收到最多约 1 秒的消息。快照频率可在broker.conf中配置。只同步基线游标位置复制订阅只同步基线游标位置base line cursor position不同步单条确认individual acknowledgments。这意味着如果消息是乱序确认的集群故障切换时这些已确认的消息可能被重新投递。上述两个参数均可在 broker 配置中调整以当前仓库 conf/broker.conf 为准复制订阅快照频率配置项位于 ServiceConfiguration 的replicatedSubscriptionsSnapshotFrequencyMillis默认1_000毫秒即 1 秒。调小可缩小重复窗口但会增加快照开销。与复制相关的 broker 配置除了复制订阅快照频率conf/broker.conf中还提供了若干与地理复制强相关的配置项理解它们有助于调优复制通道配置项默认值作用replicationMetricsEnabledtrue是否启用复制指标Replication metricsreplicationConnectionsPerBroker16每个 broker 到远程集群最多打开的连接数更多连接在高延迟链路上可带来更好的吞吐replicationProducerQueueSize1000复制生产者replicator producer的队列大小即复制器在内存中可堆积的待发送消息条数其读取深度还受调度器最大读批大小约束见 PersistentReplicator 中readBatchSize的计算replicatorPrefixpulsar.repl复制生产者名称与游标名称使用的前缀replicationPolicyCheckDurationSeconds600检查复制策略的间隔避免因缺失 ZooKeeper watch 导致复制器状态不一致设为0可禁用replicationTlsEnabledfalse复制通道是否启用 TLS这些配置项定义于 ServiceConfiguration如replicationConnectionsPerBroker 16、replicationProducerQueueSize 1000均可在源码中找到默认值完整说明见 reference-configuration.md。典型使用场景与注意事项场景一多地域就近读写。在每个地域部署一个 Pulsar 集群客户端连接本地serviceUrl写入与消费均本地完成地理复制保证数据全局一致可见跨地域读延迟由 RTT 决定。场景二跨地域容灾/故障切换。配合复制订阅当一个地域整体故障时消费者可在另一集群从快照基线位置继续消费注意存在约 1 秒的重复投递窗口。生产端无需特殊处理消息在本地持久化后异步转发即使复制通道中断也不丢消息。使用时的关键注意事项租户的--allowed-clusters必须覆盖所有参与复制的集群复制范围以命名空间为准set-clusters可动态调整、不中断流量单条消息可用setReplicationClusters做更细粒度的选择性复制复制订阅需要每个客户端在创建消费者时设置replicateSubscriptionState(true)且要接受亚秒级重复与乱序确认可能导致的重复投递地理复制主题不能直接删除需清理全部生产/消费者与订阅后依赖自动垃圾回收。至此你已经掌握了 Pulsar 地理复制从租户授权、命名空间指派到复制订阅调优的完整链路。更多相关概念可继续阅读 reference-terminology.md、client-libraries-java.md 与 reference-pulsar-admin.md。赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 跨集群地理复制Geo-Replication实战指南配置、原理与复制订阅Apache Pulsar 跨集群地理复制Geo Replication实战指南配置、原理与复制订阅 本篇技术指南以 Apache Pulsar本仓库为消息队列后端流处理Apache Pulsar 跨地域复制Geo-replication实战指南原理、配置与复制订阅Apache Pulsar 跨地域复制Geo replication实战指南原理、配置与复制订阅 导读 跨地域复制Geo replication是 A消息队列后端流处理Apache Pulsar 跨地域复制Geo-replication完整实战指南原理、配置与复制订阅机制Apache Pulsar 跨地域复制Geo replication完整实战指南原理、配置与复制订阅机制 本文以 Apache Pulsar 的 geo消息队列后端流处理上一篇OpenCore Legacy Patcher技术深度解析揭秘老Mac升级的底层原理与硬件兼容性革命下一篇gh stack sync完全指南一条命令完成fetch、rebase、push与同步创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考