跳至主内容

RabbitMQ Stream 教程 - 偏移量跟踪

简介​

信息

先决条件​

本教程假定 RabbitMQ 已安装，正在localhost上运行，并且stream 插件已启用。标准的 stream 端口是 5552。如果您使用不同的主机、端口或凭据，则需要调整连接设置。

使用 Docker​

如果您没有安装 RabbitMQ，可以在 Docker 容器中运行它

docker run -it --rm --name rabbitmq -p 5552:5552 -p 15672:15672 -p 5672:5672  \
    -e RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_stream advertised_host localhost' \
    rabbitmq:4-management

等待服务器启动，然后启用 stream 和 stream management 插件

docker exec rabbitmq rabbitmq-plugins enable rabbitmq_stream rabbitmq_stream_management

哪里寻求帮助​

如果您在学习本教程时遇到困难，可以通过邮件列表或Discord 社区服务器与我们联系。

RabbitMQ Streams 在 RabbitMQ 3.9 中引入。更多信息请参见此处。

偏移量跟踪​

设置​

本教程的这一部分将编写两个 Go 程序：一个生产者，它发送一批带有标记消息的消息，以及一个消费者，它接收消息并在收到标记消息时停止。它演示了消费者如何浏览流，甚至可以从之前的执行中断处继续。

本教程使用 stream Go 客户端。请确保按照第一个教程中的 设置步骤 进行操作。

本教程的可执行版本可以在 RabbitMQ 教程仓库 中找到。发送程序名为 offset_tracking_send.go，接收程序名为 offset_tracking_receive.go。本教程侧重于客户端库的使用，因此应使用仓库中的最终代码来创建文件的脚手架（例如，导入、主函数等）。

发送​

发送程序首先创建环境并声明流。

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

streamName := "stream-offset-tracking-go"
env.DeclareStream(streamName,
    &stream.StreamOptions{
        MaxLengthBytes: stream.ByteCapacity{}.GB(2),
    },
)

请注意，为了简洁起见，已省略错误处理代码。

然后，程序创建一个生产者并发布 100 条消息。最后一条消息的正文值设置为 marker；这是消费者停止消费的标记。

该程序使用 handlePublishConfirm 函数和一个通道来确保所有消息在退出前都已发送到代理。我们暂时跳过这部分，先看看发送的主要内容。

producer, _ := env.NewProducer(streamName, stream.NewProducerOptions())

messageCount := 100
ch := make(chan bool)
chPublishConfirm := producer.NotifyPublishConfirmation()
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"
    }
    producer.Send(amqp.NewMessage([]byte(body)))
}
_ = <-ch
fmt.Println("Messages confirmed")

producer.Close()

发送程序使用 2 个通道和一个 Go 例程来继续执行，并在代理确认消息后退出。让我们关注这部分。

生产者 NotifyPublishConfirmation 函数返回第一个通道。客户端库在此通道上发送代理确认，并在 handlePublishConfirm 中声明的例程接收这些确认。

messageCount := 100
chPublishConfirm := producer.NotifyPublishConfirmation()
ch := make(chan bool)
handlePublishConfirm(chPublishConfirm, messageCount, ch)

该例程处理消息，并在达到预期确认数时向第二个通道发送 true。

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
                    }
                }
            }
        }
    }()
}

主程序在发送循环结束后等待第二个通道，因此一旦有东西进入通道（例程发送的 true 值），它就会继续执行。

得益于这种同步机制，程序不会在所有消息都发送到代理之前停止。

现在，让我们创建接收程序。

接收​

接收程序也创建环境并声明流。这部分代码与发送程序相同，因此为了简洁起见，在接下来的代码片段中已省略。

接收程序启动一个消费者，该消费者附加到流的开头（stream.OffsetSpecification{}.First()）。它使用变量来输出程序结束时收到的第一条和最后一条消息的偏移量。

消费者在收到标记消息时停止：它将偏移量分配给一个变量，关闭消费者，并将 true 发送到一个通道。与发送方一样，当消费者完成其工作时，该通道会通知程序继续执行。

var firstOffset int64 = -1
var lastOffset atomic.Int64
ch := make(chan bool)
messagesHandler := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
    if atomic.CompareAndSwapInt64(&firstOffset, -1, consumerContext.Consumer.GetOffset()) {
        fmt.Println("First message received.")
    }
    if string(message.GetData()) == "marker" {
        lastOffset.Store(consumerContext.Consumer.GetOffset())
        _ = consumerContext.Consumer.Close()
        ch <- true
    }
}

offsetSpecification := stream.OffsetSpecification{}.First()
_, _ = env.NewConsumer(streamName, messagesHandler,
    stream.NewConsumerOptions().
        SetOffset(offsetSpecification))

fmt.Println("Started consuming...")
_ = <-ch

fmt.Printf("Done consuming, first offset %d, last offset %d.\n", firstOffset, lastOffset.Load())

探索流​

要运行这两个示例，请打开两个终端（shell）标签页。

在第一个选项卡中，运行发送程序以发布一系列消息。

go run offset_tracking_send.go

输出如下：

Publishing 100 messages
Messages confirmed.

现在让我们运行接收程序。打开一个新标签页。请记住，由于 first 偏移量规范，它应该从流的开头开始。

go run offset_tracking_receive.go

这是输出：

Started consuming...
First message received.
Done consuming, first offset 0, last offset 99.
什么是偏移量？

流可以看作是一个包含消息的数组。偏移量是数组中给定消息的索引。

流与队列不同：消费者可以读取和重读相同的消息，并且消息会保留在流中。

让我们尝试使用 offset 规范附加到给定偏移量来测试此功能。将 offsetSpecification 变量从 stream.OffsetSpecification{}.First() 更改为 stream.OffsetSpecification{}.Offset(42)。

offsetSpecification := stream.OffsetSpecification{}.Offset(42)

偏移量 42 是任意的，它可以是 0 到 99 之间的任何数字。再次运行接收程序。

go run offset_tracking_receive.go

输出如下：

Started consuming...
First message received.
Done consuming, first offset 42, last offset 99.

还有一种方法可以附加到流的末尾，以便在创建消费者时只查看新消息。这就是 next 偏移量规范。让我们试试。

offsetSpecification := stream.OffsetSpecification{}.Next()

运行接收程序。

go run offset_tracking_receive.go

这次消费者没有收到任何消息。

Started consuming...

它正在等待流中的新消息。通过再次运行发送程序来发布一些消息。回到第一个选项卡。

go run offset_tracking_send.go

等待程序退出，然后切换回接收程序选项卡。消费者收到了新消息。

Started consuming...
First message received.
Done consuming, first offset 100, last offset 199.

接收程序因为发送程序将其放在流末尾的新标记消息而停止。

本节展示了如何“浏览”流：从开头、从任何偏移量，甚至对于新消息。下一节将介绍如何利用服务器端偏移量跟踪，以便从消费者前一次执行的中断处恢复。

服务器端偏移量跟踪​

RabbitMQ Streams 提供服务器端偏移量跟踪，用于存储流中给定消费者的进度。如果消费者因任何原因停止（崩溃、升级等），它将能够从先前停止的位置重新连接，以避免处理相同的消息。

RabbitMQ Streams 提供了偏移量跟踪的 API，但也可以使用其他解决方案来存储正在消耗的应用程序的进度。这可能取决于用例，但关系型数据库也是一个不错的解决方案。

让我们修改接收程序以存储已处理消息的偏移量。已更新的行用注释标出。

var firstOffset int64 = -1
var messageCount int64 = -1 // number of received messages
var lastOffset atomic.Int64
ch := make(chan bool)
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() // store offset every 10 messages
    }
    if string(message.GetData()) == "marker" {
        lastOffset.Store(consumerContext.Consumer.GetOffset())
        consumerContext.Consumer.StoreOffset() // store the offset on consumer closing
        consumerContext.Consumer.Close()
        ch <- true
    }
}

var offsetSpecification stream.OffsetSpecification
consumerName := "offset-tracking-tutorial" // name of the consumer
storedOffset, err := env.QueryOffset(consumerName, streamName) // get last stored offset
if errors.Is(err, stream.OffsetNotFoundError) {
    // start consuming at the beginning of the stream if no stored offset
    offsetSpecification = stream.OffsetSpecification{}.First()
} else {
    // start just after the last stored offset
    offsetSpecification = stream.OffsetSpecification{}.Offset(storedOffset + 1)
}

_, err = env.NewConsumer(streamName, messagesHandler,
    stream.NewConsumerOptions().
        SetManualCommit(). // activate manual offset tracking
        SetConsumerName(consumerName). // the consumer must a have name
        SetOffset(offsetSpecification))
fmt.Println("Started consuming...")
_ = <-ch

fmt.Printf("Done consuming, first offset %d, last offset %d.\n", firstOffset, lastOffset.Load())

最相关的更改是：

  • 程序在创建消费者之前查找最后存储的偏移量。如果没有存储的偏移量（这很可能是该消费者首次启动），它将使用 first。如果存在存储的偏移量，它将使用 offset 规范从存储偏移量之后（stored offset + 1）开始，这假设具有存储偏移量的消息已在应用程序的先前实例中处理。
  • 消费者必须有一个名称。它是存储和检索最后一个存储的偏移量值的关键。
  • 已激活手动跟踪策略，这意味着需要显式调用来存储偏移量。
  • 偏移量每 10 条消息存储一次。对于偏移量存储频率而言，这是一个异常低的值，但对于本教程来说没问题。实际世界中的值通常是几百或几千。
  • 偏移量在关闭消费者之前存储，就在收到标记消息之后。

现在让我们运行更新后的接收程序。

go run offset_tracking_receive.go

这是输出：

Started consuming...
First message received.
Done consuming, first offset 0, last offset 99.

这没什么令人惊讶的：消费者从流的开头获取了消息，并在到达标记消息时停止。

让我们再次启动它。

go run offset_tracking_receive.go

这是输出：

Started consuming...
First message received.
Done consuming, first offset 100, last offset 199.

消费者精确地从中断处恢复：第一次运行的最后一个偏移量是 99，第二次运行的第一个偏移量是 100。消费者在第一次运行时存储了偏移量跟踪信息，因此客户端库在第二次运行时使用它来从正确的位置恢复消费。

本教程关于 RabbitMQ Streams 中的消费语义的内容到此结束。它涵盖了消费者如何附加到流中的任何位置。消费应用程序很可能需要跟踪它们在流中达到的点。它们可以使用本教程中演示的内置服务器端偏移量跟踪功能。它们也可以自由使用任何其他数据存储解决方案来完成此任务。

有关偏移量跟踪的更多信息，请参阅 RabbitMQ 博客 和 stream Go 客户端文档。

© . This site is unofficial and not affiliated with VMware.