如何在独立请求中手动确认NATS JetStream消息序列？

通常你会在确定的上下文里从消费者接收消息、处理并确认：

const messages = await consumer.consume({ max_messages: maxMessages });

for await (const m of messages) {
    // 一些处理逻辑
    await m.ack();
}

我的使用场景是把HTTP API和NATS消息流做桥接，定义了两个HTTP端点：

• GET /events：从拉取式消费者中拉取消息；
• POST /events/ack?lastReceivedEventSeqId=10：向NATS服务器确认序列号为10及之前的消息已处理完成。

问题在于，当丢失了最初从NATS服务器接收消息的上下文后，没法对消息执行ack操作：

// 在POST /events/ack端点的实现里
const s : Stream | undefined = await jetstream.streams.get('mystream');
const message = await s.getMessage({seq: 10});
// message没有ack方法

解决方案 #

你不需要通过从流中获取的消息对象来确认，而是要通过持久化的消费者实例直接执行确认操作，核心是保留消费者的上下文而非单独的消息上下文。

关键步骤 #

1. 初始化一个持久化的拉取式消费者：设置durable_name让NATS服务器保存消费者状态，跨请求复用该实例。
2. 在确认端点中，复用同一个消费者实例，调用其ack方法传入要确认的最大序列号，即可批量确认该序列号及之前的所有消息。

示例代码 #

// 全局维护持久化消费者实例（服务启动时初始化）
let durableConsumer;

async function initConsumer() {
    const stream = await jetstream.streams.get('mystream');
    durableConsumer = await stream.consume({
        durable_name: 'http_bridge_consumer', // 必须设置，确保消费者状态持久化
        ack_policy: 'explicit', // 显式确认策略，必填
        // 拉取模式相关配置，比如批量拉取大小等
    });
}

// POST /events/ack 端点实现
app.post('/events/ack', async (req, res) => {
    const lastSeqId = parseInt(req.query.lastReceivedEventSeqId);
    if (isNaN(lastSeqId)) {
        return res.status(400).send('无效的序列号参数');
    }

    try {
        // 通过消费者直接确认指定序列号及之前的所有消息
        await durableConsumer.ack(lastSeqId);
        res.status(200).send(`已成功确认序列号 ${lastSeqId} 及之前的消息`);
    } catch (err) {
        res.status(500).send(`确认失败：${err.message}`);
    }
});

注意事项 #

• 必须复用同一个消费者实例：不能每次请求都创建新消费者，否则无法关联之前的消费状态。
• 持久化名称不可省略：durable_name是让NATS服务器保存消费者进度的关键，重启服务或跨请求都能恢复状态。
• 同步确认可选：如果需要确保确认操作已被NATS服务器处理，可以使用durableConsumer.ackSync(lastSeqId)替代ack，它会等待服务器的确认响应。

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