如何分析Pulsar 的消息保留和过期策略

# 如何分析Pulsar的消息保留和过期策略 ## 目录 1. [Pulsar消息保留机制概述](#1-pulsar消息保留机制概述) 2. [消息保留策略的配置方法](#2-消息保留策略的配置方法) 3. [消息过期(TTL)机制详解](#3-消息过期ttl机制详解) 4. [保留策略与过期策略的协同工作](#4-保留策略与过期策略的协同工作) 5. [性能影响与最佳实践](#5-性能影响与最佳实践) 6. [监控与故障排查](#6-监控与故障排查) 7. [实际案例研究](#7-实际案例研究) --- ## 1. Pulsar消息保留机制概述 ### 1.1 基本概念 Apache Pulsar采用多层存储架构，其消息保留策略包含两个维度： - **时间保留（Retention Time）**：消息在Topic中的最小保存时长 - **空间保留（Retention Size）**：消息在Topic中占用的最大存储空间 ### 1.2 保留策略的作用 - 防止消费者滞后时的数据丢失 - 控制存储成本 - 满足合规性要求 - 为消息重放(Replay)提供基础 ### 1.3 架构层面的实现 ```mermaid graph TD A[Producer] -->|Publish| B[Broker] B -->|Persist| C[BookKeeper Ledger] C -->|Offload| D[Long-Term Storage] E[Consumer] -->|Subscribe| B 

---

2. 消息保留策略的配置方法

2.1 命名空间级别配置

# 设置时间保留策略（默认-1表示不限制） bin/pulsar-admin namespaces set-retention my-tenant/my-ns \ --time 7d \ --size 10G # 查看当前配置 bin/pulsar-admin namespaces get-retention my-tenant/my-ns 

2.2 Topic级别覆盖配置

# 覆盖特定Topic的保留策略 bin/pulsar-admin topics set-retention persistent://tenant/ns/topic1 \ --time 30d \ --size 50G 

2.3 配置项详解

参数	类型	默认值	说明
retentionTimeInMinutes	int	-1	分钟为单位的时间保留
retentionSizeInMB	long	-1	MB为单位的空间保留

---

3. 消息过期(TTL)机制详解

3.1 TTL工作原理

// 生产者端设置消息TTL Producer<byte[]> producer = client.newProducer() .topic("my-topic") .sendTimeout(30, TimeUnit.SECONDS) .create(); // 单个消息设置TTL producer.newMessage() .value("content".getBytes()) .property("TTL", "3600000") .send(); 

3.2 过期检查流程

1. Broker启动后台清理线程
2. 定期扫描Ledger中的消息时间戳
3. 对比当前时间与publishTimestamp + TTL
4. 物理删除过期消息

3.3 配置建议

• 全局默认TTL在broker.conf中设置：

  # 默认消息TTL（秒） ttlDurationDefaultInSeconds=3600 # TTL检查间隔 brokerDeleteInactiveTopicsFrequencySeconds=60 

---

4. 保留策略与过期策略的协同工作

4.1 策略优先级矩阵

场景	生效策略
消息未过期 + 在保留期内	保留
消息已过期 + 在保留期内	删除
消息未过期 + 超出保留期	删除
消息已过期 + 超出保留期	删除

4.2 数据清理流程

sequenceDiagram participant C as Cleanup Thread participant L as Ledger participant S as Storage C->>L: 获取消息元数据 loop 每条消息 alt 消息已过期 C->>L: 标记删除 else 超出保留大小 C->>S: 触发卸载 end end L->>S: 执行物理清理 

---

5. 性能影响与最佳实践

5.1 性能考量因素

• IO压力：频繁的过期检查会增加磁盘IO
• 内存消耗：大保留策略需要更多缓存
• 网络带宽：跨机房复制时的数据传输

5.2 推荐配置方案

# 生产环境推荐配置 retention_policy: time_based: default: 72h important_topics: 30d size_based: default: 50GB high_volume: 200GB ttl: default: 24h sensitive_data: 1h 

---

6. 监控与故障排查

6.1 关键监控指标

# 消息积压量 pulsar_storage_size{cluster="prod",topic="persistent://tenant/ns/topic1"} # 过期消息数 pulsar_expired_messages_count{namespace="my-tenant/my-ns"} # 清理操作耗时 pulsar_broker_storage_operation_latency{op="delete"} 

6.2 常见问题处理

问题1：消息过早被删除 - 检查顺序： 1. 确认TTL设置 2. 验证系统时钟同步 3. 检查保留策略覆盖

问题2：存储空间不释放 - 排查步骤：

 # 检查待卸载数据 pulsar-admin topics stats-internal persistent://tenant/ns/topic1 # 手动触发清理 pulsar-admin topics cleanup persistent://tenant/ns/topic1 

---

7. 实际案例研究

7.1 金融交易场景

需求： - 交易数据保留7天 - 敏感信息1小时后过期 - 最大保留100GB

实现：

# 命名空间配置 pulsar-admin namespaces set-retention finance/trades \ --time 7d \ --size 100G # 敏感Topic设置TTL pulsar-admin topics set-message-ttl finance/trades/sensitive \ --messageTTL 3600 

7.2 IoT数据处理

特殊挑战： - 设备可能长期离线 - 需要支持历史数据重放 - 海量小消息存储

解决方案： - 分层保留策略：

 -- Hot层: 保留1天 -- Warm层: 保留30天（卸载到对象存储） -- Cold层: 保留1年（归档到HDFS） 

---

附录：相关配置参考

完整参数列表参见Pulsar官方文档 “`