Google Pub/Sub 消息排序问题导致延迟超过 10 秒？

对于一个Golang开发者来说，牢固扎实的基础是十分重要的，golang学习网就来带大家一点点的掌握基础知识点。今天本篇文章带大家了解《Google Pub/Sub 消息排序问题导致延迟超过 10 秒？》，主要介绍了，希望对大家的知识积累有所帮助，快点收藏起来吧，否则需要时就找不到了！

我正在尝试制作一个简化的示例，演示如何使用 google pub/sub 的消息排序功能 (https://cloud.google.com/pubsub/docs/ordering)。从这些文档中，在为订阅启用消息排序后，

设置消息排序属性后，pub/sub 服务将按照 pub/sub 服务接收消息的顺序传递具有相同排序键的消息。例如，如果发布者发送具有相同排序键的两条消息，则 pub/sub 服务将首先传送最旧的消息。

我用它来编写以下示例：

package main

import (
    "context"
    "log"
    "time"

    "cloud.google.com/go/pubsub"
    uuid "github.com/satori/go.uuid"
)

func main() {
    client, err := pubsub.newclient(context.background(), "my-project")
    if err != nil {
        log.fatalf("newclient: %v", err)
    }

    topicid := "test-topic-" + uuid.newv4().string()
    topic, err := client.createtopic(context.background(), topicid)
    if err != nil {
        log.fatalf("createtopic: %v", err)
    }
    defer topic.delete(context.background())

    subid := "test-subscription-" + uuid.newv4().string()
    sub, err := client.createsubscription(context.background(), subid, pubsub.subscriptionconfig{
        topic:                 topic,
        enablemessageordering: true,
    })
    if err != nil {
        log.fatalf("createsubscription: %v", err)
    }
    defer sub.delete(context.background())

    ctx, cancel := context.withcancel(context.background())
    defer cancel()

    messagereceived := make(chan struct{})
    go sub.receive(ctx, func(ctx context.context, msg *pubsub.message) {
        log.printf("received message with ordering key %s: %s", msg.orderingkey, msg.data)
        msg.ack()
        messagereceived <- struct{}{}
    })

    topic.publish(context.background(), &pubsub.message{data: []byte("dang1!"), orderingkey: "foobar"})
    topic.publish(context.background(), &pubsub.message{data: []byte("dang2!"), orderingkey: "foobar"})

    for i := 0; i < 2; i++ {
        select {
        case <-messagereceived:
        case <-time.after(10 * time.second):
            log.fatal("expected to receive a message, but timed out after 10 seconds.")
        }
    }
}

首先，我尝试了该程序，但没有在 topic.publish() 调用中指定 orderingkey: "foobar" 。这导致了以下输出：

> go run main.go
2020/08/10 21:40:34 received message with ordering key : dang2!
2020/08/10 21:40:34 received message with ordering key : dang1!

换句话说，消息接收的顺序与发布的顺序不同，这在我的用例中是不可取的，我想通过指定 orderingkey 来防止

但是，当我在发布调用中添加 orderingkeys 时，程序在等待接收 pub/sub 消息 10 秒后超时：

> go run main.go
2020/08/10 21:44:36 Expected to receive a message, but timed out after 10 seconds.
exit status 1

我期望现在首先收到消息 dang1! 然后是 dang2!，但我没有收到任何消息。知道为什么这没有发生吗？

解决方案

发布失败并出现以下错误：无法发布：topic.enablemessageordering=false，但在消息中设置了 orderingkey。请删除 orderingkey 或打开 topic.enablemessageordering。

如果您更改发布调用以检查错误，您可以看到这一点：

res1 := topic.publish(context.background(), &pubsub.message{data: []byte("dang1!"), orderingkey: "foobar"})
res2 := topic.publish(context.background(), &pubsub.message{data: []byte("dang2!"), orderingkey: "foobar"})

_, err = res1.get(ctx)
if err != nil {
    fmt.printf("failed to publish: %v", err)
    return
}

_, err = res2.get(ctx)
if err != nil {
    fmt.printf("failed to publish: %v", err)
    return
}

要修复此问题，请添加一行以启用主题的消息排序。您的主题创建如下：

topic, err := client.createtopic(context.background(), topicid)
if err != nil {
    log.fatalf("createtopic: %v", err)
}
topic.enablemessageordering = true
defer topic.delete(context.background())

我独立提出了与 kamal 相同的解决方案，只是想分享完整修改后的实现：

package main

import (
    "context"
    "flag"
    "log"
    "time"

    "cloud.google.com/go/pubsub"
    uuid "github.com/satori/go.uuid"
)

var enablemessageordering bool

func main() {
    flag.boolvar(&enablemessageordering, "enablemessageordering", false, "enable and use pub/sub message ordering")
    flag.parse()

    client, err := pubsub.newclient(context.background(), "fleetsmith-dev")
    if err != nil {
        log.fatalf("newclient: %v", err)
    }

    topicid := "test-topic-" + uuid.newv4().string()
    topic, err := client.createtopic(context.background(), topicid)
    if err != nil {
        log.fatalf("createtopic: %v", err)
    }
    topic.enablemessageordering = enablemessageordering
    defer topic.delete(context.background())

    subid := "test-subscription-" + uuid.newv4().string()
    sub, err := client.createsubscription(context.background(), subid, pubsub.subscriptionconfig{
        topic:                 topic,
        enablemessageordering: enablemessageordering,
    })
    if err != nil {
        log.fatalf("createsubscription: %v", err)
    }
    defer sub.delete(context.background())

    ctx, cancel := context.withcancel(context.background())
    defer cancel()

    messagereceived := make(chan struct{})
    go sub.receive(ctx, func(ctx context.context, msg *pubsub.message) {
        log.printf("received message with ordering key %s: %s", msg.orderingkey, msg.data)
        msg.ack()
        messagereceived <- struct{}{}
    })

    msg1, msg2 := &pubsub.message{data: []byte("dang1!")}, &pubsub.message{data: []byte("dang2!")}
    if enablemessageordering {
        msg1.orderingkey, msg2.orderingkey = "foobar", "foobar"
    }
    publishmessage(topic, msg1)
    publishmessage(topic, msg2)

    for i := 0; i < 2; i++ {
        select {
        case <-messagereceived:
        case <-time.after(10 * time.second):
            log.fatal("expected to receive a message, but timed out after 10 seconds.")
        }
    }
}

func publishmessage(topic *pubsub.topic, msg *pubsub.message) {
    publishresult := topic.publish(context.background(), msg)
    messageid, err := publishresult.get(context.background())
    if err != nil {
        log.fatalf("get: %v", err)
    }
    log.printf("published message with id %s", messageid)
}

当调用 enablemessageordering 标志设置为 true 时，我首先收到 dang1!，然后收到 dang2!：

> go run main.go --enablemessageordering
2020/08/11 05:38:07 published message with id 1420685949616723
2020/08/11 05:38:08 published message with id 1420726763302425
2020/08/11 05:38:09 received message with ordering key foobar: dang1!
2020/08/11 05:38:11 received message with ordering key foobar: dang2!

如果没有它，我会以与以前相反的顺序收到它们：

> go run main.go
2020/08/11 05:38:47 Published message with ID 1420687395091051
2020/08/11 05:38:47 Published message with ID 1420693737065665
2020/08/11 05:38:48 Received message with ordering key : Dang2!
2020/08/11 05:38:48 Received message with ordering key : Dang1!

好了，本文到此结束，带大家了解了《Google Pub/Sub 消息排序问题导致延迟超过 10 秒？》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！