Go GCP Cloud PubSub 不批量发布消息

Posted

技术标签:

【中文标题】Go GCP Cloud PubSub 不批量发布消息【英文标题】:Go GCP Cloud PubSub not batch publishing messages 【发布时间】:2019-07-03 21:11:05 【问题描述】:

我正在开发一个示例项目,该项目从 bigquery 获取输出并将其发布到 pubsub。 bigquery 的行输出可能大于 100,000。我看到有批量发布的选项,并且我在多个地方读到每批 1k 条消息是理想的。我遇到的问题是,在我的一生中,我无法让它批量处理多条消息,我认为解决方案很简单,但我不知道如何去做..

这就是我现在所拥有的,它所做的只是一次发布一条消息。

func publish(client pubsub.Client, data []byte) (string, error) 
    ctx := context.Background()

    topic := client.Topic("topic-name")
    topic.PublishSettings = pubsub.PublishSettings
        // ByteThreshold:  5000,
        CountThreshold: 1000, // no matter what I put here it still sends one per publish
        // DelayThreshold: 1000 * time.Millisecond,
    

    result := topic.Publish(ctx, &pubsub.Message
        Data: data,
    )

    id, err := result.Get(ctx)
    if err != nil 
        return "", err
    

    return id, nil

这个函数被调用:

for _, v := range qr 
        data, err := json.Marshal(v)
        if err != nil 
            log.Printf("Unable to marshal %s", data)
            continue
        
        id, err := publish(*pubsubClient, data)
        if err != nil 
            log.Printf("Unable to publish message: %s", data)
        
        log.Printf("Published message with id: %s", id)
    

其中 qr 是包含从 bigquery 查询返回的数据的结构切片。

现在,是不是因为我调用函数publish 使每条消息都被发布,而topic.PublishSettings 被每个方法调用覆盖,所以它忘记了以前的消息?我在这里不知所措。

我在这里看到了一些批量发布代码:https://github.com/GoogleCloudPlatform/golang-samples/blob/master/pubsub/topics/main.go#L217

但他们实际上并没有在他们的示例中调用它,所以我不知道应该怎么做。

旁注并进一步证明我的观点它不起作用,如果我将 topic.PublishSettings 变量中的 DelayThreshold 设置为 1 秒,它只是每秒发布一条消息,而不是所有消息应该在内存中。

感谢您的帮助,谢谢。

编辑#1:

因此,随着 kingkupps 的评论,我将代码切换为以下代码以用于测试目的:(项目和主题名称从真实名称切换)

func QueryAndPublish(w http.ResponseWriter, r *http.Request) 
    ctx := context.Background()
    // setting up the pubsub client
    pubsubClient, err := pubsub.NewClient(ctx, "fake-project-id")
    if err != nil 
        log.Fatalf("Unable to get pubsub client: %v", err)
    

    // init topic and settings for publishing 1000 messages in batch
    topic := pubsubClient.Topic("fake-topic")
    topic.PublishSettings = pubsub.PublishSettings
        // ByteThreshold:  5000,
        CountThreshold: 1000,
        // DelayThreshold: 1000 * time.Millisecond,
    

    // bq set up
    bqClient, err := bigquery.NewClient(ctx, "fake-project-id")
    if err != nil 
        log.Fatalf("Unable to get bq client: %v", err)
    
    // bq query function call
    qr, err := query(*bqClient)
    if err != nil 
        log.Fatal(err)
    
    log.Printf("Got query results, publishing now")

    // marshalling messages to json format
    messages := make([][]byte, len(qr))
    timeToMarshal := time.Now()
    for i, v := range qr 
        data, err := json.Marshal(v)
        if err != nil 
            log.Printf("Unable to marshal %s", data)
            continue
        
        messages[i] = data
    
    elapsedMarshal := time.Since(timeToMarshal).Nanoseconds() / 1000000
    log.Printf("Took %v ms to marshal %v messages", elapsedMarshal, len(messages))

    // publishing messages
    timeToPublish := time.Now()
    publishCount := 0
    for _, v := range messages 
        // ignore result, err from topic.Publish return, just publish
        topic.Publish(ctx, &pubsub.Message
            Data: v,
        )
        publishCount++
    
    elapsedPublish := time.Since(timeToPublish).Nanoseconds() / 1000000
    log.Printf("Took %v ms to publish %v messages", elapsedPublish, publishCount)

    fmt.Fprint(w, "Job completed")

现在这样做是当我的消息计数为 100,000 时,它将在大约 600 毫秒内完成发布调用,但在后台,它仍将一一发布到 pubsub 端点。

我可以在 StackDriver 和 Wireshark 中看到这一点,其中我在 stackdriver 中的消息/秒大约是 10-16/秒,而 Wireshark 显示每条发送的消息都有新的连接。

【问题讨论】:

您是否尝试过使用相同的 pubsub.Topic 发布您的所有消息?我想知道是否每个主题都有自己的消息队列。这些示例对此有点不清楚,但 godoc 似乎表明每个主题都有自己的资源池。 godoc.org/cloud.google.com/go/pubsub#hdr-Publishing @kingkupps 我或多或少地将publish 代码移到了调用函数中。该主题是在创建客户端后立即定义的,现在在该 for 循环中我使用相同的主题,但结果是相同的。我还删除了result.Get(ctx) 调用,但这只是导致函数快速执行,因为该调用阻塞,但topic.Publish 没有。所有这一切都是在后台发布的消息。我还使用 Wireshark 检查了我的网络流量,这似乎表明每条正在发送的消息都有一个连接请求。 编辑帖子以显示新代码 sn-p。 【参考方案1】:

这很可能是因为当你打电话时

topic.PublishSettings = pubsub.PublishSettings // ByteThreshold: 5000, CountThreshold: 1000, // DelayThreshold: 1000 * time.Millisecond,

您正在将发布设置重置为零初始化结构。这会将 topic.PublishSettings.ByteThreshold 设置为 0,这意味着所有消息将立即发布;你告诉它等到它有 0 个字节,它总是有 0 个字节。

相反,您应该执行以下操作来设置 CountThreshold:

topic.PublishSettings.CountThreshold = 1000

这同样适用于其他领域。如here 所述,它们已经初始化为默认值,如果要更改它们,请直接修改它们,而不是重新分配整个 PublishSetttings 对象。

【讨论】:

就是这样。谢谢! 大家好,好像不是排他性条件。我将 CountThreshold 设置为 2 并将 DelayThreshold 设置为 10 seconds 。其中只有一个控制事件。即使我有 2 条消息要发布,发布者正在等待 10 秒发送第一个消息,其他 10 秒发送第二个消息……我希望发布者在他的 CountThreshold 已满后立即发送事件……你能帮忙吗我?

以上是关于Go GCP Cloud PubSub 不批量发布消息的主要内容,如果未能解决你的问题,请参考以下文章

使用 GCP pubsub 的 Spring Cloud Stream 消费者的并发设置

从 GCP 发布/订阅中捕获错误代码

从 Cloud Function 发布到 GCP PubSub 的正确方法是啥?

在 Spring Cloud GCP pubsub 中创建特定于消息通道的线程

GCP PubSub - 使用 orderingKey (Nodejs) 进行批处理

Golang 的稳定 GCP PubSub API