Go JetStream 异步发布的 Future 怎样区分排队与确认

Go JetStream 的 PublishAsync 返回 PubAckFuture 且即时 error 为 nil，只说明异步发布调用已交出待追踪结果，不代表已经取得服务端确认。要判断一条消息是否被流接受，需要同时等待 Future 的 Ok() 或 Err()；批量发布还要把即时失败和未决消息分别记录。

官方地址：https://nats.io/

我在整理这类发布代码时，会先把“排队数”和“成功数”拆开。异步接口容易给人已经完成的感觉，但如果计数器加在 PublishAsync 返回之后，页面显示的成功其实只是提交计数。下面用一个收据事件小批次说明我的取舍；示例未实际运行，不涉及性能实测。

迁移范围：换成异步接口，还要换成功口径

本文使用 github.com/nats-io/nats.go/jetstream，由 jetstream.New(nc) 创建对象。旧代码若使用 nc.JetStream()，不要把两个包的选项和类型拼在一起。这里讨论的是同步发布改为异步发布时的应用语义调整，不宣称某个版本刚刚改变了这些规则。

标题中的“排队”是应用侧对待确认工作的简称，不是“已经进入持久化队列”的承诺。异步发布不等待确认才返回，也不等于永不阻塞：待确认上限和背压可能使调用等待或失败。因此我更愿意把中间态命名为“待确认”，而不是“成功排队”。

一个 Future 有两种结果入口，不要只等 Ok

单条调用包含 PublishAsync、即时错误和 PubAckFuture；结果对象包含 Ok()、Err() 和 PubAck。PublishAsync 与即时错误、Future 相关，Future 分别提供 Ok 与 Err，Ok 承载 PubAck。这些是接口与对象之间的静态关系。

我会先处理函数直接返回的 error，只在它为 nil 时保存 Future。之后使用 select 同时等 Ok 和 Err，并加入等待期限。若先无条件读 Ok 再读 Err，一条失败消息可能使第一处永久等待；成功后再无条件读 Err 也没有必要。

一个 Future 安排一个结果收集者最容易维护。不要让日志协程与统计协程各自读取同一结果通道，也不要把它当成可以无限广播或反复读取的事件源。需要多方使用结果时，由收集者写入自己的结果表或事件总线。

我采用的小批量写法：先保存句柄，再归集结果

下面的函数属于独立示例包，调用方提供连接和专用 JetStream 对象。业务 ID 必须稳定且唯一，主题必须已绑定一个可写的流。函数将即时错误、异步错误、确认成功、未提交和未决分别返回，不把部分失败藏在一个成功布尔值里。

package publishdemo

import (
    "context"
    "fmt"

    "github.com/nats-io/nats.go/jetstream"
)

type Item struct {
    ID   string
    Data []byte
}

type Result struct {
    ID       string
    State    string
    Stream   string
    Sequence uint64
    Err      error
}

type pendingItem struct {
    id string
    future jetstream.PubAckFuture
}

// PublishBatch 只供小批量演示；js 应由调用方专用于当前批次。
// ID 需非空且在业务中稳定、唯一；调用方负责限制批量和正文大小。
func PublishBatch(ctx context.Context, js jetstream.JetStream,
    subject string, items []Item) []Result {
    out := make([]Result, 0, len(items))
    pending := make([]pendingItem, 0, len(items))

    for _, item := range items {
        if err := ctx.Err(); err != nil {
            out = append(out, Result{ID: item.ID, State: "not-submitted", Err: err})
            continue
        }
        // 独立复制正文，异步结果回来前不复用或修改这份数据。
        data := append([]byte(nil), item.Data...)
        future, err := js.PublishAsync(subject, data, jetstream.WithMsgID(item.ID))
        if err != nil {
            out = append(out, Result{ID: item.ID, State: "immediate-error", Err: err})
            continue
        }
        pending = append(pending, pendingItem{id: item.ID, future: future})
    }

    // 每个 Future 只由这里消费一次，两个通道用 select 二选一。
    for _, item := range pending {
        r := Result{ID: item.id}
        select {
        case ack := 

在同一包的调用函数中，可以按下面的方式初始化。该片段需要调用文件导入 context、time 和 jetstream 包，并已有可用 nc；它不是单独完整的 main 文件。

// nc 是调用方已建立的 NATS 连接；服务端启用 JetStream，主题已绑定流。
// 使用新 jetstream 包的选项，不与旧 nats.JetStreamContext 选项混用。
js, err := jetstream.New(nc,
    jetstream.WithPublishAsyncMaxPending(16),
    jetstream.WithPublishAsyncTimeout(3*time.Second),
)
if err != nil { return err }
ctx, cancel := context.WithTimeout(context.Background(), 6*time.Second)
defer cancel()

results := PublishBatch(ctx, js, "demo.receipt.created", []Item{
    {ID: "receipt-batch-a-001", Data: []byte("receipt-001")},
    {ID: "receipt-batch-a-002", Data: []byte("receipt-002")},
})
// 按 State 统计结果并保存稳定业务 ID，再决定是否需要补偿。
_ = results

这里有两个不同期限：WithPublishAsyncTimeout 为异步发布确认设置超时，ctx 控制应用收集结果的等待。PublishAsync 没有 ctx 参数，函数前面的 ctx 检查无法取消已经进入的背压等待。需要严格调用耗时预算时，还应依据所用版本设置提交等待策略、控制生产速度，而不是假设 context 自动传到所有异步操作。

我把小批量上限作为调用合同，避免把几十万条 Future 一口气存进切片。逐条等待不会把已经提交的消息重新变成逐条同步发送，但结果列表不保留输入顺序：即时错误先加入，Future 结果随后加入，应按 ID 对照。若需要输入顺序，可以为结果增加原始索引。

按逻辑推导，两条消息全部得到确认时，会有两条 acknowledged，且带有服务端返回的 Stream 和 Sequence；具体序号由流的现有状态决定，不能固定写成 1 和 2。出现异步错误就保留对应 error；ctx 先结束则产生 unresolved。本文没有把这些预期描述成实测日志。

批量完成：Complete 是范围信号，不是结果清单

PublishAsyncPending() 描述该 JetStream 对象当前仍待处理的异步发布数量，不是失败累计，也不是整个应用的队列长度。PublishAsyncComplete() 返回等待该对象 outstanding 集合完成的通道；终结中可能包含错误，清理发布器也可能让它关闭。因此我只把它用于收尾协调，不用它给每条消息补写“成功”。

此处我分为“批次归属”和“状态观察”两组：生产者、批次结果表、业务ID属于批次归属；Pending、Complete、逐条Future属于状态观察。生产者与业务ID关联，批次结果表按业务ID归档，表中的提交记录关联逐条Future，Pending与Complete描述相同发布器的在途集合，Complete还关联这些Future的终结状态。它们不构成消费者完成的证据。

要给一个有限批次收尾，先阻止新提交，并等待所有生产者完成提交调用，再观察批量信号与结果。若生产者还在提交、或调用仍等待空闲名额，Complete 即使曾经关闭，也不能覆盖后续才加入的消息。专用发布器对象比与其他业务共享同一套 Pending 统计更容易解释。

超时、重试和退出：保留未知，而不是急着改成失败

我最在意的是“停止等待”与“未写入”之间的差别。客户端没收到确认，不足以证明服务端没有存储；对异步错误也要按具体错误分类，不能统统转成可安全重试。示例的 unresolved 是保守记录，可能包含稍后才到达、但本次不再读取的结果。

需要补偿时保存业务ID、消息内容和原始错误，重试同一逻辑事件时继续使用同一个 WithMsgID。服务端去重依赖流的去重窗口等配置，不能把它理解为永久唯一约束；窗口外及消费者业务侧仍需自己的幂等机制。示例固定的两个 ID 仅为演示，生产中不能让不同收据复用它们。

关闭前先停止生产者，再收集已提交消息的结果并持久化未决记录，最后释放发布资源和连接。CleanupPublisher 是清理手段，会使仍在等待的发布失败，不是替代确认的捷径。示例函数不拥有连接，也不负责清理专用发布器；这些由调用方在退出策略中安排。

迁移前我会保留的清单与问答

• 确认导入的是新 jetstream 包，选项属于同一套 API。
• 提交计数、确认计数、即时错误、异步错误和未决分开。
• 逐条 Future 用 select 收集一次结果，保留业务 ID 和 PubAck。
• 限制批量、待确认数与正文大小，明确背压和超时策略。
• 收尾先停止提交，Complete 仅做集合协调，不能抹去逐条错误。

拿到 PubAck 后，消费者是否已经完成？没有。发布确认与消费者处理及其 Ack 属于不同方向，收据入库是否成功仍需业务自己的结果合同。

只有两三条消息也一定要用异步吗？我的取舍是先保留清楚的错误处理。若并发发布没有明确收益，使用同步 Publish 更省去 Future 的生命周期管理；这里没有吞吐测试，不能据此承诺异步一定更快。

能否只用错误回调而不保存 Future？可以围绕回调另做完整的结果归集，但不能既丢 Future 又漏回调。本文选择 Future 让消息与结果直接对应，不同时维护两套成功统计。

技术依据：官方 Publisher API、官方异步发布实现、发布器选项定义。收据事件、状态命名与收集函数为本文原创示例设计。