消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载导读在 Apache Pulsar 中broker 负责消息的持久化存储与消费流转。默认情况下broker 只为每个 topic 保留存在于**至少一个 backlog积压队列**中的消息这意味着消息一旦被所有订阅确认就会进入待删除状态。本文基于 Pulsar 官方 cookbook 文档site2/website-next/versioned_docs/version-2.2.1/cookbooks-retention-expiry.md系统讲解三大核心议题消息保留策略Retention Policies、Backlog 配额Backlog Quotas与消息 TTLTime to Live并结合仓库源码剖析其底层实现。读完本文你将能够通过pulsar-admin、REST API 与 Java Admin API 以 namespace 为粒度配置消息保留、积压上限与自动过期理解 Reader 接口场景下为何必须配置 Retention并掌握消息被真正删除的底层时机segment/entry log 滚动与垃圾回收周期。一、理解 Pulsar 默认的消息存储行为Pulsar broker 负责处理所有经过 Pulsar 的消息包括消息的持久化存储。默认情况下对于每个 topicbroker只保留至少存在于一个 backlog 中的消息。所谓 backlog是指某个特定订阅subscription上**未确认unacknowledged**消息的集合。由于一个 topic 可以拥有多个订阅因此一个 topic 也可以同时拥有多个 backlog。由此直接引出一个重要推论默认情况下从未创建过订阅的 topic 上不会保留任何消息——因为没有 backlog 承载它们。这里有一个容易被忽略的细节不再被存储的消息并不一定被立即物理删除它们实际上可能仍然可被访问直到下一次ledger 滚动rollover发生。由于客户端无法预测滚动何时发生因此不要依赖滚动恰好不会在不方便的时刻发生这种侥幸心理。Pulsar 提供了两种以namespace 为粒度修改上述默认行为的方式Retention 策略Retention Policies对不在 backlog 中的消息已被所有现有订阅确认、或根本没有订阅的消息进行持久化保留。TTLTime to Live对在指定时间内未被确认的消息执行自动确认acknowledge从而使其可以被清除。Pulsar 的管理接口admin-api-overview支持以 namespace 粒度管理这两类配置因此可以精确作用到某个 tenant 下的某个 namespace并落在一个具体集群或global集群上。要点Retention 与 TTL 解决的是两个不同的问题消息保留Retention将数据至少保留 X 小时即使已被确认。生存时间TTL在一段时间后丢弃数据通过自动确认实现。大多数应用只需要两者之一极少需要同时使用。二、Retention 策略让已被确认的消息继续保留2.1 默认行为与核心概念默认情况下当一条 Pulsar 消息到达 broker 后它会被存储直到它在所有订阅上都被确认此时该消息被标记为待删除。你可以通过为某个 namespace 下的所有 topic 设置retention policy来覆盖这一行为使那些已经在所有订阅上被确认的消息得以保留。Retention 策略同时基于**大小限制size limit与时间限制time limit**两个维度生效。Retention 策略的核心适用场景是 Reader 接口Reader 接口不使用确认机制消息不存在于任何 backlog 中。因此仅使用 Reader 的场景必须配置 Retention否则消息将无法被读取到。Retention 设置对两类消息生效没有任何订阅的 topic 上的所有消息以及已被所有订阅确认的消息。Retention 策略不影响有订阅 topic 上的未确认消息——未确认消息由 backlog 配额控制见本文第三节。当某个 topic 的保留限制被超过时最旧的消息会被标记删除直到保留的消息集合重新落在指定限制之内。2.2 实例级默认参数可以在实例级别instance level设置消息保留的默认值对应的两个参数是参数默认值说明defaultRetentionTimeInMinutes0默认保留时间分钟0表示不保留defaultRetentionSizeInMB0默认保留大小MB0表示不保留这两个参数位于 conf/broker.conf 中broker.conf更多说明可参考 reference-configuration。在 broker.conf 中还能找到与保留检查周期相关的参数retentionCheckIntervalInSeconds默认120秒conf/broker.confbroker 会按该周期定期检查并执行保留策略的清理动作。2.3 设置值组合矩阵设置 retention policy 时必须同时设置大小限制和时间限制二者缺一不可。下面这张表是官方给出的完整组合语义务必牢记时间限制大小限制消息保留效果-1-1无限保留Infinite retention-10仅按大小限制保留0-1仅按时间限制保留00禁用消息保留默认行为00非法组合00非法组合00已确认消息或无活跃订阅的消息在时间或大小任一达到限制后即不再保留从源码看RetentionPolicies 数据类pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/RetentionPolicies.java由两个字段构成retentionTimeInMinutesint与retentionSizeInMBlong其无参构造默认值为(0, 0)与文档所述默认行为一致。类的 Javadoc 也明确设置策略时两者必须同时给出不想限制某一维度时该值必须为-1任一方为0都会使保留策略实际失效。2.4 通过 pulsar-admin 设置 Retention使用set-retention子命令通过-s/--size指定大小限制、-t/--time指定时间限制。示例 1大小 10GB、时间 3 小时以下命令对my-tenant/my-nsnamespace 下的每个 topic 设置 10GB 大小限制与 3 小时时间限制若 3 小时内消息大小达到 10GB已确认消息将不再被保留若 3 小时后即使消息总大小不足 10GB已确认消息同样不再被保留。$ pulsar-admin namespaces set-retention my-tenant/my-ns \ --size 10G \ --time 3h示例 2不限时间、大小 1TB大小决定保留$ pulsar-admin namespaces set-retention my-tenant/my-ns \ --size 1T \ --time -1示例 3不限大小、时间 3 小时时间决定保留$ pulsar-admin namespaces set-retention my-tenant/my-ns \ --size -1 \ --time 3h示例 4无限保留两者都设为 -1$ pulsar-admin namespaces set-retention my-tenant/my-ns \ --size -1 \ --time -1示例 5禁用保留策略两者都设为 0$ pulsar-admin namespaces set-retention my-tenant/my-ns \ --size 0 \ --time 0注意-1与0的含义截然不同——-1表示该维度不受限0表示该维度立即失效。禁用保留必须同时将大小与时间都设为0只把其中一个设为0是非法的。2.5 通过 REST API 与 Java 设置 RetentionREST API使用 POST 请求访问/admin/v2/namespaces/:tenant/:namespace/retention端点设置保留策略。同样禁用保留策略需要将大小和时间限制都设为0只设一个为0无效。Java Admin APIint retentionTime 10; // 10 分钟 int retentionSize 500; // 500 兆字节 RetentionPolicies policies new RetentionPolicies(retentionTime, retentionSize); admin.namespaces().setRetention(namespace, policies);Java API 的同步与异步方法定义可见于 pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Namespaces.javasetRetention/setRetentionAsync、getRetention/getRetentionAsync。2.6 查询当前 Retention 配置查询某个 namespace 的 retention 策略返回结果是一个 JSON 对象包含两个键retentionTimeInMinutes和retentionSizeInMB。pulsar-admin使用get-retention子命令$ pulsar-admin namespaces get-retention my-tenant/my-ns { retentionTimeInMinutes: 10, retentionSizeInMB: 500 }REST APIGET/admin/v2/namespaces/:tenant/:namespace/retention。Javaadmin.namespaces().getRetention(namespace);三、Backlog 配额限制未确认消息的堆积3.1 什么是 backlog 配额Backlog是 topic 上由 bookies 存储的、尚未被确认的消息集合。Pulsar 会存储所有未确认消息直到它们被处理并确认。为了控制磁盘占用可以在 namespace 级别使用backlog 配额backlog quotas来控制 backlog 的允许大小。设置 backlog 配额需要指定每个 topic 在 namespace 中的允许大小阈值size threshold也可同时配置时间阈值当阈值被超过时 broker 采取何种动作的保留策略retention policy。可选的保留策略如下策略动作producer_request_holdbroker 将**暂存hold**生产请求的 payload不进行持久化producer_exceptionbroker 通过抛出异常断开与客户端的连接consumer_backlog_evictionbroker 开始丢弃积压消息警惕同名概念的混淆你可能已经注意到Pulsar 中存在两个retention policy定义一个应用于不在 backlog 中的消息的持久化保留本文第二节另一个应用于backlog 内的消息本节。使用时务必区分上下文。在源码层面这两个概念同样分离清晰Backlog 配额由 pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/BacklogQuota.java 定义其中RetentionPolicy枚举恰好包含上述三个取值且每个取值的 Javadoc 说明了其精确语义hold 生产请求、向生产者抛出 ResourceAllocationException、从最慢消费者的 backlog 中淘汰最旧消息。实现类 pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BacklogQuotaImpl.java 则承载了limitSize字节、limitTime秒与policy三个字段。值得一提的是BacklogQuota 还定义了配额类型枚举BacklogQuotaTypedestination_storage按字节大小限制 backlog与message_age按消息时间戳限制 backlog。新版 Pulsar 已支持按时间维度设置 backlog 配额这在本文 3.2 的命令示例中通过--type参数体现。3.2 设置大小/时间阈值与 backlog 保留策略可以为某个 namespace 下所有 topic 设置大小与/或时间阈值以及 backlog 保留策略set-backlog-quota命令定义于 pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java。pulsar-admin使用set-backlog-quota子命令通过-l/--limit指定大小限制、-lt/--limitTime指定时间限制单位秒非正数表示禁用时间限制、-p/--policy指定保留策略$ pulsar-admin namespaces set-backlog-quota my-tenant/my-ns \ --limit 2G \ --limitTime 36000 \ --policy producer_request_hold其中--limit 2G表示 2GB 大小阈值--limitTime 36000表示 36000 秒10 小时时间阈值。此外还可通过-t/--type指定配额类型destination_storage或message_age默认是destination_storage。REST APIPOST/admin/v2/namespaces/:tenant/:namespace/backlogQuota。Javalong sizeLimit 2147483648L; BacklogQuota.RetentionPolicy policy BacklogQuota.RetentionPolicy.producer_request_hold; BacklogQuota quota new BacklogQuota(sizeLimit, policy); admin.namespaces().setBacklogQuota(namespace, quota);Java API 中还提供了setBacklogQuota(namespace, backlogQuota, backlogQuotaType)重载形式便于针对destination_storage或message_age类型分别设置详见 Namespaces.java。3.3 broker 端如何处理配额超限从 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.java 的源码可以看到配额超限后的完整动作分支对于consumer_backlog_eviction策略broker 会执行dropBacklogForSizeLimit或dropBacklogForTimeLimit以配额限制的90% 作为目标值reductionFactor 0.9持续定位**最慢消费者slowest consumer**的游标并跳过/标记删除消息直到 backlog 回落到目标范围内。对于producer_exception与producer_request_hold策略broker 会调用disconnectProducers主动断开该 topic 上所有生产者的连接。同时在 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java 中checkBacklogQuotaExceeded会在生产者创建时进行检查对于producer_request_hold/producer_exception策略若isSizeBacklogExceeded()判定当前存储大小 ≥ 配额限制storageSize backlogQuotaLimitInBytes则会抛出TopicBacklogQuotaExceededException拒绝新生产者message_age类型则通过checkTimeBacklogExceeded按时间判断。3.4 实例级 backlog 配额默认值broker.conf 中与 backlog 配额相关的默认配置conf/broker.conf参数默认值说明backlogQuotaCheckEnabledtrue是否启用 backlog 配额检查backlogQuotaCheckIntervalInSeconds60配额检查的周期秒backlogQuotaDefaultLimitBytes-1默认大小上限字节-1表示不限制backlogQuotaDefaultLimitSecond-1默认时间上限秒-1表示不限制backlogQuotaDefaultRetentionPolicyproducer_request_hold默认超限动作另有历史遗留参数backlogQuotaDefaultLimitGB已标记 Deprecated建议改用backlogQuotaDefaultLimitBytes。3.5 查询与移除 backlog 配额查询当前配额使用get-backlog-quotas子命令$ pulsar-admin namespaces get-backlog-quotas my-tenant/my-ns { destination_storage: { limit : 2147483648, policy : producer_request_hold } }REST API 对应 GET/admin/v2/namespaces/:tenant/:namespace/backlogQuotaMapJava 对应MapBacklogQuota.BacklogQuotaType,BacklogQuota quotas admin.namespaces().getBacklogQuotas(namespace);移除配额使用remove-backlog-quota子命令$ pulsar-admin namespaces remove-backlog-quota my-tenant/my-nsREST API 对应 DELETE/admin/v2/namespaces/:tenant/:namespace/backlogQuotaJava 对应admin.namespaces().removeBacklogQuota(namespace)。3.6 手动清空 backlog使用clear-backlog子命令清空某个 namespace 的积压消息$ pulsar-admin namespaces clear-backlog my-tenant/my-ns默认情况下命令会提示你确认是否真的要为该 namespace 清空 backlog可以使用-f/--force标志跳过确认。四、TTL让未确认消息自动过期4.1 为什么需要 TTL默认情况下Pulsar 会永久存储所有未确认消息。当大量消息长期得不到确认时这会导致严重的磁盘占用。如果磁盘空间是关注点可以设置time to liveTTL来限定未确认消息的保留时长——超过 TTL 的消息会被自动确认从而可以被清理。4.2 为 namespace 设置 TTLpulsar-admin使用set-message-ttl子命令通过-ttl/--messageTTL指定 TTL单位秒$ pulsar-admin namespaces set-message-ttl my-tenant/my-ns \ --messageTTL 120 # TTL 为 2 分钟--messageTTL参数在 CmdNamespaces.java 中定义为Message TTL in seconds。REST APIPOST/admin/v2/namespaces/:tenant/:namespace/messageTTL。Javaadmin.namespaces().setNamespaceMessageTTL(namespace, ttlInSeconds);Java 接口定义位于 Namespaces.java返回CompletableFutureVoid的异步版本为setNamespaceMessageTTLAsync。4.3 查询与移除 TTL查询 TTL 配置使用get-message-ttl子命令$ pulsar-admin namespaces get-message-ttl my-tenant/my-ns 60输出为以秒为单位的整数。REST API 对应 GET/admin/v2/namespaces/:tenant/:namespace/messageTTLJava 对应admin.namespaces().getNamespaceMessageTTL(namespace)。移除 TTL 配置使用remove-message-ttl子命令$ pulsar-admin namespaces remove-message-ttl my-tenant/my-nsREST API 对应 DELETE/admin/v2/namespaces/:tenant/:namespace/messageTTLJava 对应admin.namespaces().removeNamespaceMessageTTL(namespace)。4.4 底层机制消息过期监控从源码结构看TTL 的落地由 broker 侧的过期监控器完成PersistentSubscription持有一个 pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.java它会根据游标的读取位置与消息发布/存储时间戳找出超过 TTL 阈值的消息并自动将其标记为已确认expireMessages从而推进游标的 mark-delete 位置。在 BacklogQuotaManager 的时间型清理dropBacklogForTimeLimit中同样会调用各订阅的 expiry monitor 按目标时间进行过期处理。因此TTL 与 message_age 型 backlog 配额在实现层面共享了同一条消息过期链路。五、消息真正被删除的时机滚动周期与垃圾回收如果 namespace 未配置任何 retention 期限、也从未产生大量 backlog那么已确认消息被保留的上限时长大致等于Pulsar segment 滚动周期 entry log 滚动周期 垃圾回收间隔 × 垃圾回收比例理解这个公式需要拆解三个概念1. Segment 滚动周期Segment rollover periodsegment 滚动周期本质上是指多久创建一个新 segment 的频率。一旦新 segment 被创建旧 segment 就会被删除。默认情况下触发条件是写入 50,000 条 entry消息或等待满 240 分钟二者先到先触发。你可以在 broker 中调优这两个触发条件。2. Entry log 滚动周期Entry log rollover periodBookKeeper 会将多个 ledger 交错写入同一个entry log文件中。为了删除某个已被删除的 ledger该 entry log 必须整体完成滚动。entry log 的滚动周期是可配置的但它纯粹基于 entry log 的文件大小触发而不是时间。一旦 entry log 完成滚动它就可以被垃圾回收。3. 垃圾回收间隔Garbage collection interval由于 entry log 中交错存放了多个 ledger为了释放空间需要重写entry log。垃圾回收间隔决定了 BookKeeper 执行垃圾回收的频率它与 entry log 的minor compaction小型压缩和major compaction大型压缩相关。简而言之配置了 retention / TTL 只是从逻辑层面让消息可被删除真正释放磁盘空间还依赖于底层 BookKeeper 的 ledger/entry log 滚动与垃圾回收机制。在生产环境中规划存储容量时必须把这部分物理删除延迟纳入考虑。六、实战决策总结场景推荐配置使用 Reader 接口消费历史消息必须配置 Retention如--size 10G --time 3h已确认消息希望继续保留一段固定时间Retention 时间限制--size -1 --time 3h已确认消息希望按容量上限保留Retention 大小限制--size 1T --time -1控制未确认消息的堆积总量Backlog 配额--limit 2G --policy producer_request_hold防止未确认消息永久占用磁盘TTL--messageTTL 120完全恢复 Pulsar 默认行为Retention 置0/0移除 backlog 配额与 TTL两条最重要的实战准则Retention 与 TTL 目标相反绝大多数应用二选一Retention 是多留TTL 是早删。同时配置时请先确认你的真实诉求避免策略互相打架。0与-1含义不同设置 retention 时-1 该维度无限0 该维度立即失效/禁用只设一个0是非法的组合。延伸阅读本文对应文档原文site2/website-next/versioned_docs/version-2.2.1/cookbooks-retention-expiry.md保留策略数据类pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/RetentionPolicies.javaBacklog 配额定义pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/BacklogQuota.javaBacklog 配额实现pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/BacklogQuotaImpl.javabroker 端配额处理pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BacklogQuotaManager.javatopic 端配额检查pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java消息过期监控pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentMessageExpiryMonitor.javaJava Admin 接口pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Namespaces.javaCLI 命令定义pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.javabroker 默认配置conf/broker.conf赞分享消息队列后端流处理【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pulsar28/pulsar点击查看免费下载相关推荐Apache Pulsar 消息保留与过期策略实战指南Retention、Backlog Quota 与 TTL 详解Apache Pulsar 消息保留与过期策略实战指南Retention、Backlog Quota 与 TTL 详解 Pulsar 的 Broker 默认只消息队列后端流处理Saliency框架高级技巧自定义显著性方法开发终极指南Saliency框架高级技巧自定义显著性方法开发终极指南 在深度学习可解释性领域 saliency框架 作为一款强大的显著性方法实现库为研究人员和开发者提消息队列后端流处理Apache Pulsar 消息保留与过期机制完全指南Retention、Backlog Quota 与 TTL 实战Apache Pulsar 消息保留与过期机制完全指南Retention、Backlog Quota 与 TTL 实战 本文基于 Apache Pulsar消息队列后端流处理上一篇ComfyUI超分辨率插件终极指南一键安装配置与高效使用技巧下一篇KeePassHttp终极指南如何实现浏览器与密码管理器的无缝集成创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考