消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载导读本文基于 Apache Pulsar 仓库中的设计文档 PIP-195: New bucket based delayed message tracker全面讲解 Pulsar 延迟消息Delayed Message从内存优先队列追踪演进为基于 Bucket 的持久化索引追踪的完整方案。文章覆盖新追踪器BucketDelayedDeliveryTracker的架构设计、Bucket 快照Snapshot机制、Bucket 合并与清理策略、跨订阅共享索引可选特性、全部新增/调整的配置参数与监控指标并结合仓库源码pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/给出实现级证据。读完本文你将掌握如何配置并启用 Bucket 延迟追踪器、如何理解其数据落盘与恢复流程、如何通过新指标调优参数以及滚动升级/降级时的兼容性注意事项。背景延迟消息与内存追踪器的两大痛点消息的定时/延迟投递Scheduled/Delayed Delivery是消息系统普遍支持的能力。Pulsar 自 2.4.0 起支持延迟消息其默认实现是内存版延迟消息追踪器InMemoryDelayedDeliveryTracker用一个内存优先队列维护所有延迟消息的索引消息 IDledgerId/entryId以及投递时间戳。这种实现存在两个主要问题正是 PIP-195 要解决的1. 优先队列的内存瓶颈Broker 的内存是有限的。当用户需要存储大量延迟消息时内存优先队列会占用大量 Broker 堆内存成为扩展延迟消息规模的瓶颈。虽然可以通过增加分区让延迟索引分布到多个 Broker 上但这并没有改变大量 Broker 内存被占用的事实一个 Topic 可能有多个订阅Subscription而内存版延迟索引无法跨订阅复用——每个订阅各自维护一份索引进一步加重了 Broker 的内存开销。2. 昂贵的延迟索引重建当订阅从故障中恢复、或 Broker 重启导致内存索引丢失时Broker 需要重新读取该 Topic 的所有延迟消息来重建延迟索引如果 Topic 上的延迟消息过多索引重建可能需要几分钟甚至几小时在索引重建期间消费者无法从该 Topic 消费消息带来了额外的消费者不可用时间。目标PIP-195 聚焦解决上述两个问题确立了两个明确目标支持延迟消息索引快照snapshot避免高成本的索引重建让延迟消息的规模不再受内存限制。总体方案引入基于 Bucket 的延迟消息追踪器核心思路是引入一种基于 Bucket桶的延迟消息追踪器把整个延迟消息索引按 Ledger 切分成多个 Bucket为每个 Bucket 生成多个不可变的 Segment 快照immutable segment snapshots追踪器把所有 Bucket 快照写入 BookKeeperBookie 存储节点追踪器只把每个 Bucket 当前会用到的 Segment 加载进内存从而使延迟消息规模不再受内存限制。该方案的源码实现位于仓库的pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/目录下核心类为bucket/BucketDelayedDeliveryTracker.java约 970 行配套的bucket/子包中包含ImmutableBucket.java、MutableBucket.java、BucketDelayedMessageIndex.java、BookkeeperBucketSnapshotStorage.java等实现类。延迟消息索引 Bucket 的组成一个延迟消息索引 Bucket包含若干个 Ledger 的延迟索引主要由两部分构成每个 Ledger 的 Bitset用于检查某个消息 ID 是否属于延迟消息即是否包含在延迟消息索引中优先队列用于获取到期可投递的定时消息。一个 Topic 可以拥有多个延迟消息索引 BucketBucket 的最大数量可配置delayedDeliveryMaxNumBuckets。Segment 加载与共享优先队列追踪器会将每个 Bucket 的第一个 Segment一个 Segment 对应 Bucket 快照中的一个 Entry加载到一个共享优先队列sharedBucketPriorityQueue源码中为TripleLongPriorityQueue中通过从共享优先队列轮询poll消息来获取 Topic 的到期消息。当一个 Bucket Segment 的所有消息都处理完后再加载该 Bucket 的下一个 Segment。这样任意时刻内存中只保留每个 Bucket 当前活跃 Segment的数据量。LastMutableBucket记录最新 Ledger 范围的延迟索引追踪器包含一个特殊的 Bucket——LastMutableBucket最后一个可变 Bucket它用额外的优先队列last mutable delayed message priority queue源码对应MutableBucket类记录当前最新 Ledger 范围的延迟消息索引当追踪器收到一个ledgerId LastMutableBucket.endLedgerId的消息 ID 时说明该 Bucket 的 Ledger 范围已经封口追踪器会创建一个不可变 BucketImmutableBucket并清空 LastMutableDelayedMessagePriorityQueue当常规定时任务tick触发、或从追踪器轮询消息 ID 时追踪器会把到期消息从 LastMutableDelayedMessagePriorityQueue搬移到共享延迟消息队列。从源码看BucketDelayedDeliveryTracker使用 Guava 的TreeRangeMapLong, ImmutableBucketimmutableBuckets字段维护按 ledgerId 范围Range索引的不可变 Bucket 集合实现 O(log n) 的按 Ledger 范围查找。游标如何过滤延迟消息读路径的改进订阅的分发器dispatcher通过游标cursor读取消息并派发给消费者。对于延迟消息游标需要基于延迟消息索引 Bucket 过滤掉它们例如 Topic 中有 10 条消息[0,9]其中[1,8]是延迟消息游标应该只从 Bookie 读取消息 0 和 9而当前内存版实现会读取全部 10 条消息、在 Broker 内再过滤掉[1,8]这正是需要改进之处——即让游标和 ManagedLedger 支持不连续读取discontinuous read entries。过滤判定规则如果消息不在延迟消息追踪器中且已到达投递时间Broker 直接把消息派发给消费者如果消息不在延迟追踪器中但未到达投递时间订阅只需要跳过它们——因为它们之后会被重新加入延迟消息追踪器。源码层面BucketDelayedDeliveryTracker实现了containsMessage(long ledgerId, long entryId)方法游标基于该方法过滤出所有延迟消息并在从 Bookie 读取时跳过它们同时向追踪器添加消息时也先用containsMessage避免重复记录消息索引。延迟消息索引 Bucket 快照SnapshotBucket 快照用于降低延迟消息索引重建的成本避免重放全部原始消息使用一个Ledger存储 Bucket 快照数据通过**游标属性cursor properties**维护 Bucket 快照列表即构建延迟消息索引的游标由此可以知道 Topic 有多少个延迟索引 Bucket并直接从持久化 Ledger 读取快照。源码中快照列表通过BucketDelayedDeliveryTracker.DELAYED_BUCKET_KEY_PREFIX值为CURSOR_INTERNAL_PROPERTY_PREFIX delayed.bucket即__internal_delayed.bucket前缀写入游标属性每个属性键对应bucketKey delimiter startLedgerId delimiter endLedgerId属性值即 BucketIdLedgerId。在recoverBucketSnapshot()方法中Broker 遍历游标属性、重建ImmutableBucket并并发加载各 Bucket 的快照数据到共享优先队列。Segment 划分与按需加载延迟索引 Bucket 快照数据按投递时间和索引数量上限切分成多个 SegmentSegment 划分由delayedDeliveryMaxTimeStepPerBucketSnapshotSegmentSeconds和delayedDeliveryMaxIndexesPerBucketSnapshotSegment两个参数控制只把第一个有效 Segment加载进内存当前 Segment 的延迟消息全部被调度完后再从下一个 Segment 加载延迟消息加载后续 Segment 时不会修改快照数据快照是只读的。Entry0 与元数据Bucket 快照数据从Entry1开始存储因为Entry0 记录了快照元数据metadata。maxScheduleTimestamps用于找到第一个仍有消息未到达投递时间的快照 Segment。在恢复延迟消息索引时如果某个快照 Segment 内所有消息都已到达投递时间Bucket 会跳过该 Segment因为 Broker 可以直接把这些消息派发给消费者无需重建延迟索引。delayedIndexBitMaps用于检查消息 ID 是否存在于该 Bucket 中它按快照 Segment 记录延迟消息索引的 BitSet 键值对。将某个快照 Segment 加载进内存时追踪器会把当前快照 Segment 与最后一个快照 Segment 的 BitSet 键值对合并。合并延迟消息索引 Bucket可以配置一个 Topic 的最大 Bucket 数量delayedDeliveryMaxNumBuckets。如果 Bucket 数量达到上限在封口seal新的不可变 Bucket 之前会触发 Bucket 合并追踪器找出延迟消息数量最少的两个相邻 Bucket进行合并例如存在 5 个 Bucket其消息数量为[5,3,2,4,3]则应合并第二个和第三个 Bucket合并后 Bucket 的startLedgerId更新为第二个 Bucket 的startLedgerIdendLedgerId更新为第三个 Bucket 的endLedgerId。删除延迟消息索引 Bucket 快照在以下场景需要删除 Bucket 快照Bucket 内快照的所有延迟消息都已被调度具体做法是为每个 Bucket 记录当前快照 Segment 的 entry当加载下一个快照 Segment 时如果snapshotEntryId lastSnapshotEntryId则触发删除该 Bucket 快照Bucket 合并之后追踪器删除被合并掉的旧 Bucket删除游标之前必须先删除该游标对应的所有 Bucket 快照。源码中BucketDelayedDeliveryTrackerFactory.cleanResidualSnapshots(ManagedCursor)提供了在未创建追踪器或追踪器已关闭时清理游标属性中残留快照数据的能力它遍历游标属性中所有以DELAYED_BUCKET_KEY_PREFIX开头的键逐个调用bucketSnapshotStorage.deleteBucketSnapshot(...)删除对应 Ledger 快照并移除游标属性。跨订阅共享延迟消息索引可选特性一个 Topic 可以有多个订阅。当前内存版实现为每个订阅分别构建延迟消息索引这会增加 Broker 内存开销也会多次重放日志构建索引。可选方案是使用一个独立的游标来构建共享的延迟消息索引使 Topic 下所有订阅复用同一份延迟消息索引任一订阅触发延迟消息检查时都会从延迟消息追踪器轮询消息 ID但与当前实现不同到期消息 ID 需要加入所有订阅的重放队列replay queue由各订阅的分发器负责处理新加入的消息 ID 并执行消息投递不同订阅有不同的 mark delete 位置已删除位置。如果调度消息位于 mark delete 位置之前游标读取操作会将其过滤掉风险点调度消息会从延迟消息追踪器中移除。如果 Broker 在把到期消息投递给消费者之前崩溃这些消息不会再次加入延迟追踪器Broker 也不会重新投递。但这并不是问题——因为在重放消息重建延迟索引时Broker 会跳过已过期的消息并直接把它们发送给消费者。配置变更启用 Bucket 延迟追踪器只需修改broker.conf或standalone.conf。PIP-195 给出的配置如下仓库conf/broker.conf第 742–789 行附近与pulsar-broker-common的ServiceConfiguration.java第 430–487 行附近中的默认值完全一致# Enable bucket based delayed message index tracker delayedDeliveryTrackerFactoryClassNameorg.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory # The delayed message index bucket min index count. When the index count of the current bucket is more than this value and all message indexes of current ledger have already been added to the tracker we will seal the bucket. delayedDeliveryMinIndexCountPerBucket50000 # The delayed message index bucket time step(in seconds) in per bucket snapshot segment, after reaching the max time step limitation, the snapshot segment will be cut off. delayedDeliveryMaxTimeStepPerBucketSnapshotSegmentSeconds300 # The max number of delayed message index in per bucket snapshot segment, -1 means no limitation # after reaching the max number limitation, the snapshot segment will be cut off. delayedDeliveryMaxIndexesPerBucketSnapshotSegment5000 # The max number of delayed message index bucket,after reaching the max buckets limitation, the adjacent buckets will be merged. (disable with value -1) delayedDeliveryMaxNumBuckets-1 # Enable share the delayed message index across subscriptions delayedDeliverySharedIndexEnabledfalse各参数说明与相关源码字段配置参数默认值说明delayedDeliveryTrackerFactoryClassNameorg.apache.pulsar.broker.delayed.InMemoryDelayedDeliveryTrackerFactory延迟投递追踪器的工厂类名设为org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory时启用基于 Bucket 的延迟消息索引追踪器。对应ServiceConfiguration.delayedDeliveryTrackerFactoryClassName第 433–439 行delayedDeliveryMinIndexCountPerBucket50000当前 Bucket 的索引数量超过该值、且当前 Ledger 的所有消息索引都已加入追踪器时封口该 Bucket触发不可变 Bucket 创建。对应BucketDelayedDeliveryTracker的minIndexCountPerBucket字段delayedDeliveryMaxTimeStepPerBucketSnapshotSegmentSeconds300每个 Bucket 快照 Segment 的时间步长秒达到最大时间步长后快照 Segment 被切断。工厂类读取后转换为毫秒传入追踪器TimeUnit.SECONDS.toMillis(...)delayedDeliveryMaxIndexesPerBucketSnapshotSegment5000每个 Bucket 快照 Segment 内延迟消息索引的最大数量-1表示无限制达到该数量后快照 Segment 被切断delayedDeliveryMaxNumBuckets-1Topic 的最大延迟消息索引 Bucket 数量达到上限后相邻 Bucket 会被合并-1表示禁用合并即不限制 Bucket 数量delayedDeliverySharedIndexEnabledfalse是否启用跨订阅共享延迟消息索引对应 PIP 中Share the delayed message index across subscriptions的可选特性需要注意ServiceConfiguration.java中还有几个与延迟投递相关的既有配置本文方案依赖或交互delayedDeliveryEnabled默认true是否启用延迟投递delayedDeliveryTickTimeMillis默认1000重试延迟投递的 tick 时间影响投递时间精度BucketDelayedDeliveryTrackerFactory用其构造HashedWheelTimer线程名pulsar-delayed-deliverydelayedDeliveryFixedDelayDetectionLookahead默认50000内存版追踪器检测所有消息是否固定延迟的 lookahead 窗口大小delayedDeliveryMaxDelayInMillis默认0允许的最大延迟时间毫秒超过则向生产者返回错误0表示不限制。从BucketDelayedDeliveryTrackerFactory.initialize(PulsarService)的源码pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/BucketDelayedDeliveryTrackerFactory.java可以看到工厂启动时创建BookkeeperBucketSnapshotStorage并start()随后把所有上述参数从ServiceConfiguration读入并用于构造BucketDelayedDeliveryTracker。此外当 Bucket 追踪器恢复失败抛出RecoverDelayedDeliveryTrackerException时工厂会记录告警日志并回退fallback到InMemoryDelayedDeliveryTracker保证延迟投递功能可用性。监控指标变更PIP 为延迟索引 Bucket 和快照新增了指标用于帮助用户调优配置。这些指标在源码bucket/BucketDelayedMessageIndexStats.java中实现指标名称与 PIP 完全一致名称类型说明pulsar_delayed_message_index_bucket_totalGauge延迟消息索引 Bucket 的数量不可变 Bucket LastMutableBucketpulsar_delayed_message_index_loadedGauge内存中加载的延迟消息索引总数pulsar_delayed_message_index_bucket_op_countCounter延迟消息索引 Bucket 快照操作的次数。state标签可为succeed、failed、allall表示所有状态的总数type标签可为create、load、delete、mergepulsar_delayed_message_index_bucket_snapshot_size_bytesGauge延迟消息索引 Bucket 快照的总大小字节pulsar_delayed_message_index_bucket_op_latency_msHistogram延迟消息索引 Bucket 快照操作在给定分位阈值下的延迟。type标签可为create、load、delete、mergequantile标签的含义如下quantile标签含义源码BucketDelayedMessageIndexStats中的分桶数组BUCKETS {50, 100, 500, 1000, 5000, 30000, 60000}与其一一对应quantile50操作延迟在(0ms, 50ms]区间quantile100操作延迟在(50ms, 100ms]区间quantile500操作延迟在(100ms, 500ms]区间quantile1000操作延迟在(500ms, 1s]区间quantile5000操作延迟在(1s, 5s]区间quantile30000操作延迟在(5s, 30s]区间quantile60000操作延迟在(30s, 60s]区间quantileoverflow操作延迟超过 1 分钟。注意如果启用了跨订阅共享延迟消息索引将无法获得精确的订阅级指标。实现细节PIP 的实现工作可归纳为以下几个方面仓库中均已落地1. 新增 Protobuf 定义快照元数据DelayedMessageIndexBucketMetadata.proto包名pulsar.delayJava 包org.apache.pulsar.broker.delayed.protosyntax proto2; package pulsar.delay; option java_package org.apache.pulsar.broker.delayed.proto; option optimize_for SPEED; option java_multiple_files true; message SnapshotSegmentMetadata { mapuint64, bytes delayed_index_bit_map 1; required uint64 max_schedule_timestamp 2; required uint64 min_schedule_timestamp 3; } message SnapshotMetadata { repeated SnapshotSegmentMetadata metadata_list 1; }快照 Segment 数据DelayedMessageIndexBucketSegment.protosyntax proto2; package pulsar.delay; option java_package org.apache.pulsar.broker.delayed.proto; option optimize_for SPEED; option java_multiple_files true; message DelayedIndex { required uint64 timestamp 1; required uint64 ledger_id 2; required uint64 entry_id 3; } message SnapshotSegment { repeated DelayedIndex indexes 1; }其中SnapshotSegmentMetadata的delayed_index_bit_map即上文提到的delayedIndexBitMaps按 Segment 记录 BitSet 键值对max_schedule_timestamp/min_schedule_timestamp用于快速定位第一个未到期的 SegmentDelayedIndex由timestamp投递时间戳、ledger_id、entry_id三元组构成。2. 新增BucketSnapshotStorage存储接口BucketSnapshotStorage接口用于存取延迟消息索引 Bucket 快照定义于pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketSnapshotStorage.java关键方法如下public interface BucketSnapshotStorage { /** * Create a delayed message index bucket snapshot with metadata and bucketSnapshotSegments. * * param snapshotMetadata the metadata of snapshot * param bucketSnapshotSegments the list of snapshot segments * param bucketKey the key of bucket is used to generate custom storage metadata * param topicName the name of topic is used to generate custom storage metadata * param cursorName the name of cursor is used to generate custom storage metadata * return the future with bucketId(ledgerId). */ CompletableFutureLong createBucketSnapshot(SnapshotMetadata snapshotMetadata, ListSnapshotSegment bucketSnapshotSegments, String bucketKey, String topicName, String cursorName); /** * Get delayed message index bucket snapshot metadata. * * param bucketId the bucketId of snapshot * return the future with snapshot expanded metadata */ CompletableFutureSnapshotMetadata getBucketSnapshotMetadata(long bucketId); /** * Get a sequence of delayed message index bucket snapshot segments. * param bucketId the bucketId of snapshot * param firstSegmentEntryId entryId of first segment of sequence * param lastSegmentEntryId entryId of last segment of sequence * return the future with snapshot segment */ CompletableFutureListSnapshotSegment getBucketSnapshotSegment(long bucketId, long firstSegmentEntryId, long lastSegmentEntryId); /** * Get total byte length of delayed message index bucket snapshot. * param bucketId the bucketId of snapshot * return the future with byte length of snapshot */ CompletableFutureLong getBucketSnapshotLength(long bucketId); /** * Delete delayed message index bucket snapshot by bucketId. * param bucketId the bucketId of snapshot */ CompletableFutureVoid deleteBucketSnapshot(long bucketId); /** * Start the bucket snapshot storage service. * throws Exception */ void start() throws Exception; /** * Close the bucket snapshot storage service. * throws Exception */ void close() throws Exception; }BookKeeper 后端实现为BookkeeperBucketSnapshotStoragepulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java负责把快照元数据写入 Entry0、把各 Segment 依次写入 Entry1 起的后续 Entry并提供按 Entry 范围读取、统计快照字节长度、删除快照Ledger等能力。3. 抽象重构与新追踪器从InMemoryDelayedDeliveryTracker中抽象出AbstractDelayedDeliveryTracker并实现新的追踪器BucketDelayedDeliveryTrackerbucket/BucketDelayedDeliveryTracker.java两者均位于org.apache.pulsar.broker.delayed包下新增BucketDelayedDeliveryTrackerFactoryBucketDelayedDeliveryTrackerFactory.java实现DelayedDeliveryTrackerFactory接口负责创建BucketDelayedDeliveryTracker并在创建失败时回退到内存版追踪器brokerService.initializeFallbackDelayedDeliveryTrackerFactory()BucketDelayedDeliveryTracker内部核心状态包括sharedBucketPriorityQueue共享优先队列TripleLongPriorityQueue、immutableBuckets按 Ledger 范围组织的TreeRangeMap、lastMutableBucketMutableBucket、snapshotSegmentLastIndexMap快照 Segment 最后索引到 Bucket 的映射用于判断何时删除快照等。4. 新增containsMessage过滤方法在BucketDelayedDeliveryTracker中新增containsMessage方法用于过滤出延迟消息public class BucketDelayedDeliveryTracker { //...... boolean containsMessage(long ledgerId, long entryId); }5. 游标与 ManagedLedger 支持不连续读取游标基于containsMessage过滤出所有延迟消息并在从 Bookie 读取消息时跳过它们该改动包括让游标和 ManagedLedger 支持不连续读取条目discontinuous read entries。6. 避免重复索引使用containsMessage在向延迟消息追踪器添加消息时避免记录重复的消息索引。7. 共享索引可选使用独立游标构建延迟消息追踪器并在任一订阅触发延迟消息检查时把到期消息加入所有订阅的重放队列。仓库测试覆盖方面pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java提供了端到端的 Bucket 延迟投递测试pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java与BucketDelayedDeliveryTrackerThreadSafetyTest.java覆盖了追踪器核心逻辑与线程安全性DelayedDeliveryTrackerFactoryTest.java验证了工厂与回退逻辑。兼容性升级与降级升级Upgrade可以通过滚动升级rolling upgradeBroker 节点来启用基于 Bucket 的延迟消息追踪器因为旧追踪器中的延迟消息索引只存在于内存中无需在节点间迁移持久化状态。步骤在所有 Broker 上启用基于 Bucket 的延迟消息追踪器特性修改delayedDeliveryTrackerFactoryClassName并滚动重启等待所有 Broker 节点升级完成。同样也可以通过滚动升级来启用共享延迟消息索引特性但此时延迟消息 Bucket 索引会被重建。降级Downgrade可以通过滚动降级rolling downgradeBroker 节点来禁用基于 Bucket 的延迟消息追踪器、并禁用共享延迟消息索引特性因为之前的内存追踪器可以重新构建延迟消息索引。步骤在所有 Broker 上禁用共享延迟消息索引特性、并禁用基于 Bucket 的延迟消息追踪器特性恢复InMemoryDelayedDeliveryTrackerFactory后滚动重启等待所有 Broker 节点降级完成。总结PIP-195 通过按 Ledger 切分 Bucket 快照 Segment 持久化到 BookKeeper 按需加载的设计解决了 Pulsar 内存版延迟消息追踪器在内存占用和索引重建成本上的两个根本性瓶颈同时通过containsMessage让游标支持不连续读取把延迟消息过滤从读取后过滤改进为读取前跳过。配合新增的 5 类监控指标与 6 个可调配置参数运维人员可以按 Topic 规模精细控制内存占用、快照粒度与 Bucket 数量实现延迟消息规模不受 Broker 内存限制的持久化方案。对于需要深入源码的读者建议从 BucketDelayedDeliveryTracker.java 的恢复流程recoverBucketSnapshot()入手并结合 BucketDelayedDeliveryTrackerFactory.java 与 BucketSnapshotStorage.java 理解完整链路。赞分享消息队列流处理后端微服务消息路由【免费下载链接】pulsarApache Pulsar - distributed pub-sub messaging system项目地址https://gitcode.com/gh_mirrors/pu/pulsar点击查看免费下载相关推荐Apache Pulsar PIP-315 深度解析为延迟消息投递引入可配置的最大延迟上限maxDeliveryDelayInMillisApache Pulsar PIP 315 深度解析为延迟消息投递引入可配置的最大延迟上限maxDeliveryDelayInMillis 导读 Apac消息队列后端Apache Pulsar PIP-315 深度解析为延迟投递配置最大延迟上限maxDeliveryDelayInMillisApache Pulsar PIP 315 深度解析为延迟投递配置最大延迟上限maxDeliveryDelayInMillis 本文围绕 pip 315.消息队列流处理后端微服务消息路由Apache Pulsar 延迟消息投递PIP-26 设计原理与 Broker 端实现解析Apache Pulsar 延迟消息投递PIP 26 设计原理与 Broker 端实现解析 本文基于 PIP 26Delayed Message Deliv消息队列流处理后端微服务消息路由上一篇华硕笔记本性能控制终极指南如何用G-Helper替代Armoury Crate实现极致轻量化下一篇G-Helper深度解析如何用开源工具彻底解放华硕ROG笔记本性能潜力创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
