Pulsar key_shared订阅下按Key分组消息触发动作的实现咨询

API 探路者

2026-8-10

Pulsar分组消费与延迟确认的最佳实践咨询

问题场景

我向Pulsar主题发送的消息包含无限数量的Key集合{k0, k1, ..., kn}，消息负载里有有限的类别集合{c0, c1, c2}。核心需求是：当某个Key对应的所有类别消息都被消费后，触发应用特定动作（例如k0的三类消息全部到齐后，触发k0对应的动作）。

为保证应用弹性，我仅在该Key的所有类别消息消费完成后才确认（ack）消息；若同一类别收到重复消息，会确认旧消息，只保留最新的一条。

当前使用单个消费者连接key_shared类型的订阅，已消费(k0,c0)、(k0,c1)，在等待(k0,c2)时新增第二个消费者，发现新消费者必须等待现有消费者对未处理消息执行ack或nack操作后才能接收消息，这与相关问题描述的行为一致。

想咨询两个问题：

延迟确认这种方式是合理的，但需要注意几个关键边界问题：

控制未确认消息数量：若大量Key处于等待补全类别的状态，会堆积未确认消息，占用Pulsar内存资源，可能触发订阅的未确认消息阈值告警。建议设置合理的maxUnAckedMessages参数，避免过度堆积。
处理消息超时：如果某个Key的某类消息永远无法到达（例如生产端丢消息），对应的未确认消息会长期滞留。需要为这类消息设置超时逻辑，到时间后主动执行nack或丢弃操作，避免资源浪费。
保证重复消息处理的原子性：确认旧消息与保存新消息的操作要绑定，避免出现旧消息已确认但新消息未保存的异常情况。

当前新消费者无法立即接收消息的核心原因是key_shared订阅下，未确认消息会绑定在原消费者上。可以通过以下调整优化：

对每个Key的消息，收到后先暂存但不立即ack，定期检查该Key的类别是否补全。若长时间未补全，主动nack该Key的所有未确认消息，让Pulsar重新分配给其他消费者，避免阻塞新消费者。
配置合理的negativeAckRedeliveryDelay参数，平衡消息重发频率与重新分配的及时性。

将Key的类别完成状态从消费者内存剥离，存储到外部持久化介质（如Redis、Pulsar TableView）：

消费者收到消息后，立即将Key、类别及最新消息信息写入外部存储（以Key为主键，类别为字段）。
写入后检查该Key的所有类别是否已齐全：
- 若已全量存在，触发目标动作，然后批量ack该Key的所有消息。
- 若未全量存在，直接ack当前消息——状态已持久化，后续其他消费者收到该Key的其他类别消息时，可从存储读取已有状态继续判断。
  这种方式的优势：
消费者可及时ack消息，不会阻塞负载均衡，新消费者能立即接收其他Key的消息。
状态持久化，消费者重启后不会丢失已处理的类别信息。
天然支持多消费者并行处理不同Key的消息，弹性更强。

若业务允许，可通过Pulsar Functions先对同一Key的消息做聚合，集齐所有类别后再发送到新主题，下游消费者直接处理聚合后的消息：

函数订阅原主题，以Key为分组键，缓存每个Key的类别消息。
当某Key集齐所有类别后，输出聚合消息到目标主题，并清理缓存。
下游消费者只需消费目标主题的聚合消息，处理后直接ack即可。
该方案将分组逻辑从业务消费者中剥离，简化业务逻辑，同时利用Pulsar Functions的弹性扩缩容能力，避免业务消费者的状态管理负担。

内容的提问来源于stack exchange，提问作者David Farr