如何获取jetstream.Msg消息的当前处理尝试次数以实现渐进式超时？

架构老林

2026-7-3

获取Jetstream.Msg的当前处理尝试次数

Jetstream 的 jetstream.Msg 本身没有直接暴露获取处理尝试次数的方法，但可以通过消息的Headers获取NATS自动注入的投递计数，以此实现渐进式超时逻辑。

核心思路

NATS Jetstream会为每一条投递的消息自动添加 Nats-Delivered-Count 头字段，该字段的值从1开始递增，每次消息被重新投递（比如Nak后）都会加1，这个值就是当前的处理尝试次数。

代码实现示例

import (
    "strconv"
    "time"
    "github.com/nats-io/nats.go/jetstream"
)

func handler(msg jetstream.Msg) {
    // 从Headers中获取投递次数
    countStr := msg.Headers().Get("Nats-Delivered-Count")
    attemptCount, err := strconv.Atoi(countStr)
    if err != nil {
        // 解析失败时默认设为第一次尝试
        attemptCount = 1
    }

    // 根据尝试次数计算渐进式延迟（示例为线性递增：第N次尝试延迟N*10秒）
    delay := time.Duration(attemptCount) * time.Second * 10

    err = doWork()
    if err != nil {
        _ = msg.NakWithDelay(delay)
        return
    }

    _ = msg.Ack()
}