Go 语言 RabbitMQ Stream 实战指南：Hello World 与 Offset Tracking 教程深度解析-开发者社区

示例工程

【免费下载链接】rabbitmq-tutorials

Tutorials for using RabbitMQ in various ways

项目地址：https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials

导读

本文以 go-stream/README.md 为核心，系统讲解如何在 Go 语言中通过官方客户端库rabbitmq-stream-go-client接入 RabbitMQ Streams：从环境准备、依赖安装，到两套官方教程（Tutorial One "Hello World!" 与 Tutorial Two Offset Tracking）的源码逐行剖析与运行方法。读完本文，你将掌握 RabbitMQ Stream 的生产/消费基本流程、基于发布确认的可靠发送，以及通过StoreOffset/QueryOffset实现断点续读的偏移量跟踪机制。

一、背景：RabbitMQ Streams 与 Go 客户端

RabbitMQ Streams 是 RabbitMQ 中一类特殊队列：消息一旦写入即可被反复读取（多消费者同时读同一数据），天然适合日志聚合、事件溯源、指标采集等场景。与经典 AMQP 队列"消费即删除"的模型不同，Stream 的数据保留策略由长度与大小上限控制，消费者通过**偏移量（offset）**定位读取位置。

仓库中 go-stream 目录提供了全部 4 个 Go 示例程序，覆盖两套教程：

教程	生产者	消费者	流名称
Tutorial One: Hello World!	send.go	receive.go	`hello-go-stream`
Tutorial Two: Offset Tracking	offset_tracking_send.go	offset_tracking_receive.go	`stream-offset-tracking-go`

所有示例均使用官方 Go 客户端库rabbitmq-stream-go-client中锁定版本为v1.4.6，Go 版本要求1.22.0及以上。

二、环境准备：启用 Stream 插件

成功运行示例需要满足两个前提条件：

一台运行中的 RabbitMQ 服务器；
已启用 Stream 插件（rabbitmq_stream）。

文档明确指出：To successfully use the examples you will need a running RabbitMQ server with the stream plugin enabled。Stream 插件默认监听TCP 5552 端口，这一点与源码中的连接配置完全吻合——所有示例均通过stream.NewEnvironmentOptions().SetPort(5552)建立连接，例如 send.go：

env, err := stream.NewEnvironment( stream.NewEnvironmentOptions(). SetHost("localhost"). SetPort(5552). SetUser("guest"). SetPassword("guest"))

注意：5552 是 Stream 二进制协议端口，与经典 AMQP 的 5672 不同。默认连接凭据为guest/guest（仅限 localhost 访问，生产环境请务必替换）。

启用插件的方式有两种：

运行时动态启用：rabbitmq-plugins enable rabbitmq_stream
通过插件配置文件（enabled_plugins file）预配置，使插件在服务启动时自动加载

三、获取依赖：安装 Go 客户端库

在仓库go-stream目录下执行文档给出的命令拉取客户端依赖：

go get -u github.com/rabbitmq/rabbitmq-stream-go-client

该命令会将依赖写入go.mod并更新go.sum。仓库中 go.mod 记录了完整依赖树：除了核心的rabbitmq-stream-go-client v1.4.6，还间接依赖github.com/klauspost/compress（压缩）、github.com/pierrec/lz4（LZ4 压缩编解码）、github.com/golang/snappy（Snappy 压缩）、github.com/spaolacci/murmur3（哈希）等库，说明客户端在消息压缩与校验方面已有内置支持。

安装完成后，即可在go-stream目录下用go run直接执行任意示例（示例均为package main独立程序，无外部构建步骤）。

四、Tutorial One：Hello World! —— 首个 Stream 应用

文档给出 Tutorial One 的运行方式：

go run send.go go run receive.go

建议先运行生产者再运行消费者，或保持两个终端并行（消费者从First()偏移量开始读取，可读到生产者在它启动前已写入的消息）。

4.1 生产者 send.go

send.go 的完整流程如下：

创建环境（Environment）：stream.NewEnvironment是一切操作的入口，相当于 AMQP 中的 Connection/Channel 组合，负责管理连接、流声明与生产者/消费者生命周期；
声明流（DeclareStream）：若流hello-go-stream不存在则创建，并设置MaxLengthBytes: stream.ByteCapacity{}.GB(2)，即该流最大保留2GB数据（超过后按规则淘汰最旧消息）；
创建生产者：env.NewProducer(streamName, stream.NewProducerOptions())使用默认生产者选项；
发送消息：producer.Send(amqp.NewMessage([]byte("Hello world")))发送一条Hello world消息；
优雅关闭：程序等待用户按下回车后调用producer.Close()关闭生产者。

streamName := "hello-go-stream" err = env.DeclareStream(streamName, &stream.StreamOptions{ MaxLengthBytes: stream.ByteCapacity{}.GB(2), }, ) CheckErrSend(err) producer, err := env.NewProducer(streamName, stream.NewProducerOptions()) CheckErrSend(err) // Send a message err = producer.Send(amqp.NewMessage([]byte("Hello world"))) CheckErrSend(err) fmt.Printf(" [x] 'Hello world' Message sent\n")

代码中的CheckErrSend是贯穿所有示例的简易错误处理：出错即打印并os.Exit(1)退出，便于教学演示；生产代码应替换为更健壮的错误处理与重试策略。

4.2 消费者 receive.go

receive.go 展示了 Stream 消费的核心——消息处理器回调：

messagesHandler := func(consumerContext stream.ConsumerContext, message *amqp.Message) { fmt.Printf("Stream: %s - Received message: %s\n", consumerContext.Consumer.GetStreamName(), message.Data) } consumer, err := env.NewConsumer(streamName, messagesHandler, stream.NewConsumerOptions().SetOffset(stream.OffsetSpecification{}.First())) CheckErrReceive(err)

要点解析：

消息处理器：messagesHandler在每条消息到达时被回调，通过consumerContext.Consumer.GetStreamName()拿到所属流名，message.Data取消息体；
偏移量起点：SetOffset(stream.OffsetSpecification{}.First())明确指定从流的第一条消息开始读取。这是 Stream 与传统队列的本质差异——消费者可以自由选择起点（First()、Last()、Offset(n)、Timestamp(t)等），而不是被动等待新消息；
阻塞等待：主协程在bufio.NewReader(os.Stdin).ReadString('\n')处挂起，直到用户回车才consumer.Close()退出，保证消息回调有充足时间执行。

4.3 运行效果验证

依次运行后，生产者终端输出[x] 'Hello world' Message sent；消费者终端输出Stream: hello-go-stream - Received message: Hello world。由于设置了First()偏移量，即使先启动消费者、后运行生产者，消费者也能在生产者发送后立即收到该消息——这正是 Stream"数据持久可重读"特性的直观体现。

五、Tutorial Two：Offset Tracking —— 断点续读的偏移量跟踪

Tutorial Two 解决一个核心问题：如何让消费者从上次中断的位置继续读取，而不是每次从头开始。运行方式（文档原文）：

go run offset_tracking_send.go go run offset_tracking_receive.go

5.1 生产者 offset_tracking_send.go

offset_tracking_send.go 在 Hello World 的基础上引入了**发布确认（Publish Confirmation）**机制，实现"确认送达后再继续"的可靠发送：

messageCount := 100 chPublishConfirm := producer.NotifyPublishConfirmation() ch := make(chan bool) handlePublishConfirm(chPublishConfirm, messageCount, ch) fmt.Printf("Publishing %d messages...\n", messageCount) for i := 0; i < messageCount; i++ { var body string if i == messageCount-1 { body = "marker" } else { body = "hello" } err = producer.Send(amqp.NewMessage([]byte(body))) CheckErrSend(err) } _ = <-ch fmt.Println("Messages confirmed.")

核心逻辑：

发布确认通道：producer.NotifyPublishConfirmation()返回一个确认通道，异步接收每条消息的确认状态；
确认计数：handlePublishConfirm在独立 goroutine 中遍历确认事件，用msg.IsConfirmed()判断是否确认成功，累计确认数达到messageCount（100）后向ch发送信号；
同步等待：主流程_ = <-ch阻塞直到 100 条消息全部确认，随后关闭生产者。这样程序退出时不会丢弃未确认的消息；
marker 消息：第 100 条（最后一条）消息内容为marker，作为"数据流结束"的哨兵标记，供消费者识别终止条件。

func handlePublishConfirm(confirms stream.ChannelPublishConfirm, messageCount int, ch chan bool) { go func() { confirmedCount := 0 for confirmed := range confirms { for _, msg := range confirmed { if msg.IsConfirmed() { confirmedCount++ if confirmedCount == messageCount { ch <- true } } } } }() }

5.2 消费者 offset_tracking_receive.go

offset_tracking_receive.go 是 Offset Tracking 教程的核心，完整实现了"查询已存偏移量 → 从断点继续消费 → 周期性保存偏移量"的闭环：

第一步：定义消费者名称与偏移量查询

var offsetSpecification stream.OffsetSpecification consumerName := "offset-tracking-tutorial" storedOffset, err := env.QueryOffset(consumerName, streamName) if errors.Is(err, stream.OffsetNotFoundError) { offsetSpecification = stream.OffsetSpecification{}.First() } else { offsetSpecification = stream.OffsetSpecification{}.Offset(storedOffset + 1) }

env.QueryOffset(consumerName, streamName)查询该消费者在此流上最近一次存储的偏移量；
若返回stream.OffsetNotFoundError（首次消费、尚无记录），则从First()开始；
否则从storedOffset + 1继续——storedOffset是最后已处理消息的偏移量，+1正好跳过已消费的消息，实现精确断点续读。

第二步：以手动提交模式创建消费者

_, err = env.NewConsumer(streamName, messagesHandler, stream.NewConsumerOptions(). SetManualCommit(). SetConsumerName(consumerName). SetOffset(offsetSpecification))

SetManualCommit()：开启手动提交模式，偏移量的保存时机完全由应用代码控制（区别于自动提交）；
SetConsumerName(consumerName)：为消费者命名，该名称是QueryOffset的索引键，也是偏移量跟踪的标识；
SetOffset(offsetSpecification)：应用第一步算出的起点。

第三步：消息处理与偏移量存储

messagesHandler := func(consumerContext stream.ConsumerContext, message *amqp.Message) { if atomic.CompareAndSwapInt64(&firstOffset, -1, consumerContext.Consumer.GetOffset()) { fmt.Println("First message received.") } if atomic.AddInt64(&messageCount, 1)%10 == 0 { _ = consumerContext.Consumer.StoreOffset() } if string(message.GetData()) == "marker" { lastOffset.Store(consumerContext.Consumer.GetOffset()) _ = consumerContext.Consumer.StoreOffset() _ = consumerContext.Consumer.Close() ch <- true } }

处理器中的关键设计：

记录首个偏移量：用atomic.CompareAndSwapInt64(&firstOffset, -1, consumerContext.Consumer.GetOffset())保证只记录第一条消息的偏移量（并发安全的 CAS 操作，不依赖处理顺序）；
周期性存储：每处理 10 条消息调用一次consumerContext.Consumer.StoreOffset()，把当前消费进度落盘——这就是"跟踪"的具体动作；
终止条件：遇到marker消息时，记录lastOffset、再次StoreOffset()保存最终进度、Close()关闭消费者并向ch发信号结束程序；
并发安全：firstOffset、messageCount、lastOffset全部使用sync/atomic原子操作，因为消息回调可能被客户端在多个 goroutine 中并发调用。

5.3 两次运行的断点续读验证

Offset Tracking 最有说服力的验证方式是运行两次消费者：

第一次运行：QueryOffset返回OffsetNotFoundError→ 从First()消费全部 100 条消息，期间每 10 条StoreOffset()一次，遇到marker后保存最终偏移量并退出；终端输出Started consuming...、First message received.、Done consuming, first offset 0, last offset 99.；
第二次运行：QueryOffset返回上次保存的偏移量（如 99）→ 从Offset(100)开始消费。由于流中已无新消息（数据仍保留），消费者会等待新的写入；此时再次运行offset_tracking_send.go发送 100 条新消息，消费者将只收到新增的 100 条，而不会重复读取第一次已消费的数据。

这一机制正是 Stream 在"多消费者进度独立维护"场景下的核心价值：每个消费者名称拥有独立的偏移量记录，互不干扰，天然支持水平扩展与故障恢复。

六、从源码看 rabbitmq-stream-go-client 的常用 API

结合本文 4 个示例，可将客户端常用 API 归纳如下（均来自源码中实际调用）：

API	作用	出处示例
`stream.NewEnvironment(options)`	创建客户端运行环境（连接入口）	send.go
`NewEnvironmentOptions().SetHost/SetPort/SetUser/SetPassword`	配置连接地址、Stream 端口（5552）、认证信息	receive.go
`env.DeclareStream(name, &StreamOptions{...})`	声明流；`MaxLengthBytes`控制保留上限	send.go
`env.NewProducer(name, options)`	创建生产者	offset_tracking_send.go
`producer.Send(amqp.NewMessage(body))`	发送消息（body 为`[]byte`）	send.go
`producer.NotifyPublishConfirmation()`	获取发布确认通道，实现可靠发送	offset_tracking_send.go
`env.NewConsumer(name, handler, options)`	创建消费者并注册消息回调	receive.go
`NewConsumerOptions().SetOffset(spec)`	指定读取起点（`First()`/`Offset(n)`）	receive.go
`NewConsumerOptions().SetManualCommit()`	开启手动偏移量提交	offset_tracking_receive.go
`NewConsumerOptions().SetConsumerName(name)`	命名消费者（偏移量跟踪的标识）	offset_tracking_receive.go
`env.QueryOffset(name, stream)`	查询消费者已存储的偏移量	offset_tracking_receive.go
`consumer.StoreOffset()`	手动保存当前消费偏移量	offset_tracking_receive.go

七、小结与进一步学习

本文从 go-stream/README.md 出发，结合 go-stream 目录下的完整源码，覆盖了 RabbitMQ Streams Go 开发的两大主题：

Hello World!（Tutorial One）：环境创建 → 声明流 → 生产/消费的基本链路，重点理解First()偏移量带来的"可重读"语义；
Offset Tracking（Tutorial Two）：QueryOffset+StoreOffset+ 手动提交 + 命名消费者组合出的断点续读方案，辅以发布确认保证发送可靠性。

上手步骤回顾：

# 1. 启用 Stream 插件并确认 RabbitMQ 运行 # 2. 在 go-stream 目录拉取依赖 go get -u github.com/rabbitmq/rabbitmq-stream-go-client # 3. 运行 Hello World 教程 go run send.go go run receive.go # 4. 运行 Offset Tracking 教程（可重复运行 receive 验证断点续读） go run offset_tracking_send.go go run offset_tracking_receive.go

如需深入，可继续研读仓库中其余语言的 Stream 实现（如 python-stream、javascript-nodejs-stream、java-stream-mvn、dotnet-stream），以及 elixir-stream 中基于同一套教程协议的实现，横向对比不同语言客户端在偏移量跟踪 API 上的设计异同。