示例工程【免费下载链接】rabbitmq-tutorialsTutorials 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.goreceive.gohello-go-streamTutorial Two: Offset Trackingoffset_tracking_send.gooffset_tracking_receive.gostream-offset-tracking-go所有示例均使用官方 Go 客户端库rabbitmq-stream-go-client中锁定版本为v1.4.6Go 版本要求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.goenv, 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/lz4LZ4 压缩编解码、github.com/golang/snappySnappy 压缩、github.com/spaolacci/murmur3哈希等库说明客户端在消息压缩与校验方面已有内置支持。安装完成后即可在go-stream目录下用go run直接执行任意示例示例均为package main独立程序无外部构建步骤。四、Tutorial OneHello World! —— 首个 Stream 应用文档给出 Tutorial One 的运行方式go run send.go go run receive.go建议先运行生产者再运行消费者或保持两个终端并行消费者从First()偏移量开始读取可读到生产者在它启动前已写入的消息。4.1 生产者 send.gosend.go 的完整流程如下创建环境Environmentstream.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.goreceive.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 TwoOffset Tracking —— 断点续读的偏移量跟踪Tutorial Two 解决一个核心问题如何让消费者从上次中断的位置继续读取而不是每次从头开始。运行方式文档原文go run offset_tracking_send.go go run offset_tracking_receive.go5.1 生产者 offset_tracking_send.gooffset_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()判断是否确认成功累计确认数达到messageCount100后向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.gooffset_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.goNewEnvironmentOptions().SetHost/SetPort/SetUser/SetPassword配置连接地址、Stream 端口5552、认证信息receive.goenv.DeclareStream(name, StreamOptions{...})声明流MaxLengthBytes控制保留上限send.goenv.NewProducer(name, options)创建生产者offset_tracking_send.goproducer.Send(amqp.NewMessage(body))发送消息body 为[]bytesend.goproducer.NotifyPublishConfirmation()获取发布确认通道实现可靠发送offset_tracking_send.goenv.NewConsumer(name, handler, options)创建消费者并注册消息回调receive.goNewConsumerOptions().SetOffset(spec)指定读取起点First()/Offset(n)receive.goNewConsumerOptions().SetManualCommit()开启手动偏移量提交offset_tracking_receive.goNewConsumerOptions().SetConsumerName(name)命名消费者偏移量跟踪的标识offset_tracking_receive.goenv.QueryOffset(name, stream)查询消费者已存储的偏移量offset_tracking_receive.goconsumer.StoreOffset()手动保存当前消费偏移量offset_tracking_receive.go七、小结与进一步学习本文从 go-stream/README.md 出发结合 go-stream 目录下的完整源码覆盖了 RabbitMQ Streams Go 开发的两大主题Hello World!Tutorial One环境创建 → 声明流 → 生产/消费的基本链路重点理解First()偏移量带来的可重读语义Offset TrackingTutorial TwoQueryOffsetStoreOffset 手动提交 命名消费者组合出的断点续读方案辅以发布确认保证发送可靠性。上手步骤回顾# 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 上的设计异同。赞分享示例工程【免费下载链接】rabbitmq-tutorialsTutorials for using RabbitMQ in various ways项目地址https://gitcode.com/gh_mirrors/ra/rabbitmq-tutorials点击查看免费下载相关推荐1009种编程语言Hello World终极指南从入门到精通的多语言学习宝典1009种编程语言Hello World终极指南从入门到精通的多语言学习宝典 你是否曾好奇不同编程语言如何输出Hello World 这个看似简单的示例工程如何快速掌握1000编程语言GitHub多语言Hello World项目终极指南如何快速掌握1000编程语言GitHub多语言Hello World项目终极指南 GitHub加速计划中的he/hello world项目是一个汇集了100示例工程hello-world 仓库解析1000 种语言实现 Hello World 的完整指南与贡献规范hello world 仓库解析1000 种语言实现 Hello World 的完整指南与贡献规范 hello world 是一个以“用每一种计算机语言输出示例工程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
