NATS/JetStream如何记忆订阅者状态？重启消费如何重置？

核心实现机制 #

你观察到的"服务端记住消费进度"的能力，来源于NATS JetStream的**持久化消费者（Durable Consumer）**特性，和Stream全局游标无关：

• 你调用js.pull_subscribe("hello", "hello")时，传入的第二个参数就是持久化消费者名称（Durable Name）
• NATS服务端会将该名称对应的消费者的所有状态（包括已确认的消费位点、待重试的消息列表、重试次数等）持久化存储
• 只要你后续使用相同的Durable Name创建消费者，不管客户端重启多少次，服务端都会自动从上次记录的位点继续投递消息，不会从头开始
• 不同的Durable Name对应完全独立的消费位点，多个消费者使用不同名称消费同一个Stream时，进度完全互不干扰，不存在你担心的多客户端消费冲突问题

现象成因说明 #

首次运行消费者时：

• 服务端新建了名为hello的持久化消费者，默认从Stream最早的消息开始投递
• 你消费到序号为6的消息后，程序完成10次拉取退出，此时服务端记录该消费者的已确认位点到6之后
• 剩余未消费的消息为7、8、9，其中序号为9的消息触发了nack逻辑，会被标记为重试

第二次启动消费者时：

• 复用了名为hello的持久化消费者，服务端直接从已记录的位点开始投递剩下的7、8、9三条消息
• 序号为9的消息被nack后再次返回，你总共拉到4条消息后没有更多可投递的消息，就触发了fetch超时

从头开始消费的配置方案 #

方案1：使用临时消费者（推荐测试场景使用） #

创建消费者时不传入Durable Name参数，修改订阅代码为：

sub = await js.pull_subscribe("hello")

• 这种消费者称为临时（Ephemeral）消费者，服务端不会持久化其状态
• 每次客户端重启都会创建全新的消费者，默认从头开始消费
• 客户端断开连接后，服务端会自动清理该消费者的所有相关数据

方案2：重置持久化消费者的位点 #

如果需要保留持久化消费者，有两种重置方式：

1. 使用NATS CLI手动重置：

   # 语法：nats consumer seek <stream名称> <消费者名称> [位点参数]
   nats consumer seek hello hello --all
   

2. 代码创建消费者时指定投递策略：
   创建消费者时显式指定deliver_policy为从头开始，示例：

   from nats.js import api
   sub = await js.pull_subscribe(
       "hello", 
       "hello", 
       config=api.ConsumerConfig(
           deliver_policy=api.DeliverPolicy.ALL
       )
   )
   

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