如何使用Java从NATS JetStream中检索存储的对象？

步骤说明 #

通过已有的JetStreamSubscription接收消息，提取消息中的字节数据后，用ObjectMapper反序列化为目标对象即可，以下是具体实现：

同步接收示例 #

假设你的目标对象类型为YourObject，代码如下：

// 已初始化的订阅实例和ObjectMapper
JetStreamSubscription subscription = ...;
ObjectMapper objectMapper = ...;

// 同步接收消息，设置超时时间
Message receivedMsg = subscription.nextMessage(Duration.ofSeconds(10));
if (receivedMsg != null) {
    // 手动确认消息（根据JetStream配置决定是否需要，避免重复投递）
    receivedMsg.ack();
    
    // 提取字节数据并反序列化
    byte[] msgData = receivedMsg.getData();
    YourObject retrievedObject = objectMapper.readValue(msgData, YourObject.class);
    
    // 后续处理对象逻辑
    // ...
}

异步接收示例 #

如果需要异步处理消息，可在订阅时指定回调：

JetStreamSubscription subscription = jetStream.subscribe("subject", (msg) -> {
    try {
        msg.ack();
        YourObject obj = objectMapper.readValue(msg.getData(), YourObject.class);
        // 异步处理对象逻辑
        // ...
    } catch (IOException e) {
        // 处理反序列化异常，告知JetStream处理失败
        msg.nak();
    }
});

关键注意点 #

• 消息确认：若订阅配置了手动确认模式，必须调用ack()，否则JetStream会重复投递未确认的消息。
• 异常处理：反序列化时可能抛出IOException，需捕获并处理，可调用nak()标记消息处理失败，触发重新投递。

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