Go JetStream Go 客户端怎样分批拉取消费消息

Go JetStream 里想“每次取一小批消息再处理”，优先使用 jetstream.Consumer.Fetch；它按消息数返回一个 MessageBatch，通道关闭后还要检查 batch.Error()。如果是常驻 worker，则用 Messages() 或 Consume()，让客户端复用拉取缓冲。这样既能控制单批工作量，也能避免把一次性 Fetch 误当成高吞吐循环。

官方地址：https://pkg.go.dev/github.com/nats-io/nats.go/jetstream

本文只讨论 nats.go 的 JetStream pull consumer：按消息数或字节数取批次、显式确认、批次错误和生命周期清理。示例是接口用法说明，不把未执行代码描述成运行结果。

先分清三种拉取方式

Fetch、Messages 和 Consume 都建立在 pull consumer 上，但控制点不同。一次性的导入任务、定时批处理适合 Fetch；长期运行的 worker 更适合 Messages 或 Consume，因为它们会复用客户端缓冲并预取后续请求。

因此，“分批拉取”不一定意味着每轮都重新调用 Fetch。如果批次只是业务处理粒度，而进程需要持续工作，可以使用 Messages 的缓冲，再由业务代码决定何时凑够一批。

先创建显式确认的 pull consumer

消费者配置决定了后面的拉取行为。下面的示例创建一个名为 batch-worker 的 durable pull consumer，并把确认策略设为 AckExplicitPolicy。显式确认的好处是：只有业务处理成功后才确认，处理失败的消息不会被代码误标记为完成。

package main

import (
	"context"
	"fmt"
	"time"

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

func createConsumer(ctx context.Context) (jetstream.Consumer, *nats.Conn, error) {
	// 连接关闭后，JetStream 客户端和消费者句柄都不应继续使用。
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		return nil, nil, fmt.Errorf("连接 NATS 失败: %w", err)
	}

	js, err := jetstream.New(nc)
	if err != nil {
		nc.Close() // 初始化 JetStream 失败时及时回收连接。
		return nil, nil, fmt.Errorf("创建 JetStream 客户端失败: %w", err)
	}

	consumer, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{
		Durable:       "batch-worker",
		AckPolicy:     jetstream.AckExplicitPolicy,
		FilterSubject: "events.order",
	})
	if err != nil {
		nc.Close()
		return nil, nil, fmt.Errorf("创建消费者失败: %w", err)
	}
	return consumer, nc, nil
}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel() // 管理请求也要有明确的结束时间。

	consumer, nc, err := createConsumer(ctx)
	if err != nil {
		panic(err)
	}
	defer nc.Close() // 所有消费工作结束后再关闭连接。
	_ = consumer
}

生产代码通常会在启动阶段创建或取得 stream，再把这个 consumer 句柄交给消费循环。FilterSubject 是可选的；如果不需要按 subject 筛选，可以省略它。不要在每次批处理循环里重复创建 consumer，durable 名称应由应用生命周期负责。

用 Fetch 按消息数处理一批

Fetch(20) 表示“最多取 20 条”，不是保证一定有 20 条。默认等待时间是 30 秒，也可以使用 FetchMaxWait 调整。读取 MessageBatch.Messages() 得到的通道后，必须继续检查 MessageBatch.Error()，否则心跳丢失等传输问题可能被当成正常的少量消息。

func consumeBatch(ctx context.Context, consumer jetstream.Consumer) error {
	// 单次最多取 20 条，避免空闲时无限等待；实际数量可能少于 20。
	batch, err := consumer.Fetch(20, jetstream.FetchMaxWait(2*time.Second))
	if err != nil {
		return fmt.Errorf("发起批量拉取失败: %w", err)
	}

	for msg := range batch.Messages() {
		// 先完成业务处理，再确认，避免失败消息被提前标记为完成。
		if err := handleOrder(msg.Data()); err != nil {
			return fmt.Errorf("处理消息失败: %w", err)
		}
		if err := msg.Ack(); err != nil {
			return fmt.Errorf("确认消息失败: %w", err)
		}
	}

	// 通道结束只代表本次投递结束，批次本身仍可能携带传输错误。
	if err := batch.Error(); err != nil {
		return fmt.Errorf("批次投递失败: %w", err)
	}
	return nil
}

func handleOrder(data []byte) error {
	// 这里替换为幂等的订单处理逻辑；不要在示例中吞掉业务错误。
	if len(data) == 0 {
		return fmt.Errorf("消息体为空")
	}
	return nil
}

批次内某条消息处理失败时，示例直接返回而不确认该消息。真实项目可以把失败消息转入重试队列、记录业务错误或等待消费者重新投递，但应先明确幂等键和重复处理策略。

消息大小差异大时按字节控量

只按条数控制并不能限制网络和内存开销：一条 JSON 可能只有几百字节，也可能接近消息大小上限。此时可以使用 FetchBytes，或者在 Fetch 上组合 FetchMaxBytes。结果仍是“达到任一限制或等待时间到期就结束”，读取方式和错误处理不变。

func consumeByBytes(consumer jetstream.Consumer) error {
	// 同时限制条数和总字节数，先达到哪个条件就结束本次批次。
	batch, err := consumer.Fetch(100,
		jetstream.FetchMaxBytes(4

字节上限应覆盖可能出现的最大单条消息，并留意官方文档对 subject、reply subject 和 headers 等数据也计入限制的说明。若把限制设得过小，可能出现拉取停滞或批次经常提前结束；这不是把批量参数越小越安全。

持续消费时改用 Messages 或 Consume

如果 worker 需要持续接收消息，不建议在无限循环里反复调用 Fetch。官方文档指出，Messages 与 Consume 会做持续消费优化；前者更适合需要自己掌握逐条处理节奏的 work queue，后者适合把处理逻辑放进回调。

func consumeContinuously(ctx context.Context, consumer jetstream.Consumer) error {
	// PullMaxMessages 控制客户端缓冲，避免预取过多消息占用内存。
	iter, err := consumer.Messages(
		jetstream.PullMaxMessages(20),
		jetstream.PullExpiry(2*time.Second),
	)
	if err != nil {
		return fmt.Errorf("创建消息迭代器失败: %w", err)
	}
	defer iter.Stop() // 退出循环时释放订阅和迭代器资源。

	for {
		select {
		case 

需要回调模型时可以使用 Consume，并保留返回的 ConsumeContext：

func startCallbackConsumer(consumer jetstream.Consumer) (jetstream.ConsumeContext, error) {
	// 回调不会自动确认，处理完成后仍要显式调用 Ack。
	return consumer.Consume(func(msg jetstream.Msg) {
		if err := handleOrder(msg.Data()); err != nil {
			return // 失败消息不确认，交给重投策略处理。
		}
		_ = msg.Ack()
	}, jetstream.PullMaxMessages(20))
}

应用关闭时调用 Stop 可以立即停止接收；希望缓冲中的消息先处理完，则根据生命周期选择 Drain。不要丢弃消费上下文，否则后台 goroutine 和订阅可能继续存活。

分批消费的四个检查点

常见问题

Fetch(20) 一定会返回 20 条吗？不会，它最多返回 20 条，等待时间先到、当前可用消息不足或发生错误时都可能提前结束。

什么时候用 FetchBytes？当单条消息大小差别很大，业务更关心单批传输量或内存上限时使用；同时要确保上限不会小于最大可能消息。

为什么持续循环不推荐一直 Fetch？Fetch 是一次请求并为每次执行建立投递路径，Messages 和 Consume 更适合持续消费，会复用缓冲和拉取优化。

Messages 和 Consume 怎么选？需要自己控制每条消息的处理节奏、并发和退出条件时选 Messages；已有回调式处理模型时选 Consume，但要保存并停止消费上下文。