如何实现Pulsar消费者消费消息后从主题队列移除消息？

Pulsar 本身不会在消息被消费后立刻从队列移除，而是通过消息确认（Ack）机制标记已消费状态，Broker 会基于消费进度和留存策略自动清理已确认的消息，以此避免消费者重启后重复拉取已处理的消息。以下是具体实现方式：

1. 严格执行手动消息确认 #

这是核心前提，只有当消息处理完成后再确认，Broker 才会将其标记为已消费，重启后不会再投递。

• 单条消息确认：

  Message<byte[]> msg = consumer.receive();
  try {
      // 执行消息业务处理逻辑
      handleMessage(msg);
      // 处理成功后确认消息
      consumer.acknowledge(msg);
  } catch (Exception e) {
      // 处理失败时发起负确认，让消息重新投递
      consumer.negativeAcknowledge(msg);
  }
  

• 批量累积确认：如果是顺序消费场景，可使用累积确认来提升性能，确认当前消息及之前所有未确认的消息：

  consumer.acknowledgeCumulative(msg);
  

  注意：累积确认不适合并发消费场景，可能导致未处理的消息被误标记为已消费。

2. 配置持久化订阅与正确的订阅类型 #

• 确保使用持久化订阅：Pulsar 默认就是持久化订阅（主题前缀为persistent://），Broker 会持久化保存消费进度，消费者重启后会从上次确认的位置继续消费。如果使用非持久化订阅（non-persistent://），消费进度不会被保存，重启后会从头拉取消息。
  客户端配置示例：

  Consumer<byte[]> consumer = client.newConsumer()
      .topic("persistent://my-tenant/my-namespace/my-topic")
      .subscriptionName("my-fixed-sub-name") // 固定订阅名，不要每次重启都变更
      .subscriptionType(SubscriptionType.Exclusive) // 根据业务选：Exclusive/Shared/Key_Shared
      .subscribe();
  

• 禁用自动确认：不要开启enableAutoAck(true)，自动确认会在收到消息后立刻标记为已消费，若业务处理失败会直接丢失消息，且无法回溯。

3. 配置Broker的消息留存策略 #

通过调整主题的留存规则，让Broker在消息被确认后自动清理：

• 设置留存时间/大小：当消息被确认且超过留存时间或大小阈值时，Broker会自动从队列中移除这些消息。
  命令行配置示例：

  pulsar-admin topics set-retention persistent://my-tenant/my-namespace/my-topic \
    --time 1440 \ # 留存1天，单位为分钟
    --size 1024   # 留存上限1GB，单位为MB
  

• 极端场景（消费后立即移除）：可将留存时间设为0，但此操作风险极高——若确认请求丢失，消息会直接丢失，无回溯可能，仅适合完全不需要消息回溯的业务。

4. 额外保障：实现业务幂等性 #

即使配置完善，极端情况下（如确认请求在网络中丢失）仍可能出现重复消费，因此业务逻辑必须实现幂等：

• 用消息的getMessageId()作为唯一标识，在处理前检查是否已处理过；
• 基于业务唯一键（如订单ID、用户ID+操作类型）做幂等校验，避免重复执行业务动作。

5. 配置死信队列（可选） #

将多次处理失败的消息转发到死信队列，避免重复投递影响正常消费，同时便于后续排查问题：

Consumer<byte[]> consumer = client.newConsumer()
    .topic("persistent://my-tenant/my-namespace/my-topic")
    .subscriptionName("my-subscription")
    .deadLetterPolicy(DeadLetterPolicy.builder()
        .maxRedeliverCount(3) // 最大重试3次
        .deadLetterTopic("persistent://my-tenant/my-namespace/my-dlq-topic")
        .build())
    .subscribe();

内容的提问来源于stack exchange，提问作者kavana kishore