限制 Google Pub/Sub 传输尝试的方法：使用死信主题进行消息转发

小伙伴们对Golang编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《限制 Google Pub/Sub 传输尝试的方法：使用死信主题进行消息转发》，就很适合你，本篇文章讲解的知识点主要包括。在之后的文章中也会多多分享相关知识点，希望对大家的知识积累有所帮助！

我正在尝试使用死信主题配置 pub/sub 订阅（参见 https://cloud.google.com/pubsub/docs/dead-letter-topics），以限制死信主题的数量当消息被拒绝时，消息会被重新传递。为此，我创建了以下示例程序：

package main

import (
    "context"
    "flag"
    "fmt"
    "log"
    "os"
    "time"

    "cloud.google.com/go/pubsub"
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
)

const (
    topicid           = "my-topic"
    deadlettertopicid = "my-dead-letter-topic"
    subscriptionid    = "my-subscription"
)

var (
    pubsubemulatorhost string
    projectid          string
)

func main() {
    flag.stringvar(&pubsubemulatorhost, "pubsubemulatorhost", "", "pub/sub emulator host (e.g. localhost:8085)")
    flag.stringvar(&projectid, "projectid", "my-project", "google project id")
    flag.parse()

    if pubsubemulatorhost != "" {
        os.setenv("pubsub_emulator_host", pubsubemulatorhost)
        defer os.unsetenv("pubsub_emulator_host")
    }

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

    topic, err := client.createtopic(context.background(), topicid)
    if err != nil {
        if status.code(err) == codes.alreadyexists {
            topic = client.topic(topicid)
            log.printf("topic %s already exists", topicid)
        } else {
            log.fatalf("createtopic: %v", err)
        }
    }
    defer func() {
        topic.stop()
        if err := topic.delete(context.background()); err != nil {
            log.fatalf("delete topic: %v", err)
        }
    }()

    deadlettertopic, err := client.createtopic(context.background(), deadlettertopicid)
    if err != nil {
        if status.code(err) == codes.alreadyexists {
            deadlettertopic = client.topic(deadlettertopicid)
            log.printf("topic %s already exists", deadlettertopicid)
        } else {
            log.fatalf("createtopic: %v", err)
        }
    }
    defer func() {
        deadlettertopic.stop()
        if err := deadlettertopic.delete(context.background()); err != nil {
            log.fatalf("delete dead-letter topic: %v", err)
        }
    }()

    sub, err := client.createsubscription(context.background(), subscriptionid, pubsub.subscriptionconfig{
        topic: topic,
        deadletterpolicy: &pubsub.deadletterpolicy{
            deadlettertopic:     fmt.sprintf("projects/%s/topics/%s", projectid, deadlettertopicid),
            maxdeliveryattempts: 5,
        },
    })
    if err != nil {
        log.fatalf("createsubscription: %v", err)
    }
    defer func() {
        if err := sub.delete(context.background()); err != nil {
            log.fatalf("delete subscription: %v", err)
        }
    }()

    go func() {
        sub.receive(context.background(), func(ctx context.context, msg *pubsub.message) {
            log.printf("got message %q upon delivery attempt %d", msg.data, msg.deliveryattempt)
            msg.nack()
        })
    }()

    result := topic.publish(context.background(), &pubsub.message{data: []byte("hello, world!")})
    messageid, err := result.get(context.background())
    if err != nil {
        log.fatalf("get message id of publish call: %v", err)
    }
    log.printf("published message with id %s", messageid)
    time.sleep(20 * time.second)
}

该脚本以两种模式运行，一种是使用真正的 pub/sub 项目（此处称为 my-project），另一种是通过设置 pubsub_emulator_host 环境变量来使用 gcloud pub/sub 模拟器。我预计，鉴于订阅的 deadletterpolicy 将 maxdeliveryattempts 设置为 5，nack'd pub/sub 消息将被传送大约 5 次（文档表明这是尽力而为）。但是，如果我在真实的 pub/sub 项目上运行该脚本，我会得到以下输出：

> go run main.go
2020/06/22 23:59:37 published message with id 1294186248588871
2020/06/22 23:59:38 got message "hello, world!" upon delivery attempt 824637866440
2020/06/22 23:59:40 got message "hello, world!" upon delivery attempt 824634417896
2020/06/22 23:59:41 got message "hello, world!" upon delivery attempt 824634418592
2020/06/22 23:59:43 got message "hello, world!" upon delivery attempt 824637866928
2020/06/22 23:59:44 got message "hello, world!" upon delivery attempt 824638981864
2020/06/22 23:59:45 got message "hello, world!" upon delivery attempt 824640667960
2020/06/22 23:59:47 got message "hello, world!" upon delivery attempt 824634418712
2020/06/22 23:59:49 got message "hello, world!" upon delivery attempt 824638982160
2020/06/22 23:59:50 got message "hello, world!" upon delivery attempt 824640667760
2020/06/22 23:59:51 got message "hello, world!" upon delivery attempt 824634418000
2020/06/22 23:59:52 got message "hello, world!" upon delivery attempt 824633942168
2020/06/22 23:59:53 got message "hello, world!" upon delivery attempt 824633942712
2020/06/22 23:59:53 got message "hello, world!" upon delivery attempt 824640668296
2020/06/22 23:59:54 got message "hello, world!" upon delivery attempt 824637448352
2020/06/22 23:59:55 got message "hello, world!" upon delivery attempt 824633943336
2020/06/22 23:59:55 got message "hello, world!" upon delivery attempt 824633943448
2020/06/22 23:59:56 got message "hello, world!" upon delivery attempt 824633943560
2020/06/22 23:59:57 got message "hello, world!" upon delivery attempt 824638259688
2020/06/22 23:59:57 got message "hello, world!" upon delivery attempt 824637448752

换句话说，nack'd 消息被传递了 19 次，远低于我预期的 5 次。如果我使用 pub/sub 模拟器运行该程序，我发现传递尝试始终为 0：

> go run main.go --pubsubEmulatorHost=localhost:8085
2020/06/23 00:00:54 Published message with ID 4
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:54 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:55 Got message "Hello, world!" upon delivery attempt 0
2020/06/23 00:00:56 Got message "Hello, world!" upon delivery attempt 0
...

这里为了简洁起见，输出被截断，但消息打印了约 200 次（每秒 10 次，持续 20 秒），再次远远高于我预期的 5 次。

deadletterpolicy 的 maxdeliveryattempts 字段是否不应限制 nack 消息的传送尝试次数？为什么 deliveryattempt 字段是这样一个奇怪的整数序列，而不是每次简单地递增 1 的整数序列（参见 https://pkg.go.dev/cloud.google.com/go/pubsub?tab=doc#message ）？

解决方案

为了进一步为社区做出贡献，我根据我在评论部分中提到的内容发布了这个答案。

您描述的问题通常发生在您未授予所需权限（死信队列选项下的 here）以便 pubsub 可以发布到您的死信主题或订阅您的订阅时。另外，我必须指出，如果写入死信队列主题失败，pubsub 将继续将消息传递给您的订阅者。

为了授予必要的权限，您可以在 shell 环境中使用以下命令：

pubsub_service_account="service-${project_number}@gcp-sa-pubsub.iam.gserviceaccount.com"

gcloud pubsub topics add-iam-policy-binding  \  --member="serviceaccount:${pubsub_service_account}"\  --role='roles/pubsub.publisher'

gcloud pubsub subscriptions add-iam-policy-binding  \  --member="serviceaccount:${pubsub_service_account}"\  --role='roles/pubsub.subscriber'

另外，我想添加 @mahesh gattani 评论，其中提到模拟器目前不支持死信主题。

您可以通过 gcp ui 实现它。

Go to Subscription -> Click on Edit -> 
Go to Dead lettering -> 
Add max attempt(minimum number is 5) -> 
add the topic (where you want to send message once your maximum attempt exceeds)

终于介绍完啦！小伙伴们，这篇关于《限制 Google Pub/Sub 传输尝试的方法：使用死信主题进行消息转发》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！