使用Apache Pulsar共享订阅构建分布式任务队列，如何获取当前运行任务？

你的需求完全可行，以下是几种具体实现方式：

一、通过Pulsar Admin API获取统计数据（推荐用于数量统计） #

Pulsar提供的Admin API可直接查询订阅的未确认消息状态，适合快速获取当前运行任务的数量：

• 调用getSubscriptionStats(topic, subscription)接口，返回的SubscriptionStats对象包含核心统计字段：
  • unackedMessages：该订阅下所有消费者的未确认消息总数
  • consumers数组：每个消费者的详细统计，其中ConsumerStats的unackedMessages字段代表单个消费者的未确认消息数
• Java客户端示例代码：

  PulsarAdmin admin = PulsarAdmin.builder().serviceUrl("pulsar://localhost:6650").build();
  SubscriptionStats stats = admin.topics().getSubscriptionStats("persistent://my-tenant/my-namespace/my-topic", "my-shared-subscription");
  long totalUnacked = stats.getUnackedMessages();
  System.out.println("当前运行任务总数：" + totalUnacked);
  

• 注意：Admin API默认仅返回未确认消息的数量及元数据，无法直接获取消息内容。

二、消费者本地跟踪+分布式汇总（支持获取具体任务列表） #

如果需要获取具体的当前运行任务内容，可在消费者端维护本地状态，再通过分布式存储实现全局汇总：

• 每个消费者在接收消息后（receive()方法或监听器的received()回调），将消息ID与内容存入本地线程安全集合（如ConcurrentHashMap<MessageId, T>）
• 在调用acknowledge()确认消息前，从集合中移除对应记录
• 为实现全局汇总，将每个消费者的本地集合同步至共享存储（如Redis、ZooKeeper）：
  • 接收消息时向共享存储添加记录
  • 确认消息时从共享存储删除记录
  • 查询时直接读取共享存储的有效记录，即可得到全局未确认任务的列表与数量

三、使用Cursor API获取未确认消息范围（进阶用法） #

通过Pulsar的低级Cursor API，可获取订阅中未确认消息的范围，进而遍历获取消息内容：

• 先通过Admin API获取订阅的所有消费者名称
• 针对每个消费者，获取对应的Cursor实例，调用getLastMessageId()和getMarkDeletePosition()计算未确认消息的范围
• 使用readMessagesFromCursor()方法遍历该范围内的消息，即可获取具体的未确认消息
• 注意：该方式实现复杂度较高，需处理分区主题的多分区场景，适合熟悉Pulsar底层机制的开发者

内容的提问来源于stack exchange，提问作者Marco Stahl