读取 Google Pub Sub 中的所有消息，使用 Go 实现

编程并不是一个机械性的工作，而是需要有思考，有创新的工作，语法是固定的，但解决问题的思路则是依靠人的思维，这就需要我们坚持学习和更新自己的知识。今天golang学习网就整理分享《读取 Google Pub Sub 中的所有消息，使用 Go 实现》，文章讲解的知识点主要包括，如果你对Golang方面的知识点感兴趣，就不要错过golang学习网，在这可以对大家的知识积累有所帮助，助力开发能力的提升。

我正在尝试从 google pub-sub 中的某个主题获取所有可用消息。 但在 go 中，我无法找到一个配置，一旦 pub-sub 中没有剩余消息，就可以取消接收回调。

我认为的一种方法是使用此答案 google pubsub - counting messages in topic and then keep a count of the messages read and calling cancel 中描述的 google cloud monitoring api 获取来自 pub-sub 的消息总数如果计数等于数字，但我不太确定这是否是继续前进的正确方法。

var mu sync.mutex
    received := 0
    sub := client.subscription(subid)
    cctx, cancel := context.withcancel(ctx)
    err = sub.receive(cctx, func(ctx context.context, msg *pubsub.message) {
            mu.lock()
            defer mu.unlock()
            fmt.fprintf(w, "got message: %q\n", string(msg.data))
            msg.ack()
            received++
            if received == totalnumberofmessages {
                    cancel()
            }
    })
    if err != nil {
            return fmt.errorf("receive: %v", err)
    }

我也尝试过使用带有超时的上下文，即获取直到未满足此上下文截止日期，然后取消。

ctx, cancel := context.WithTimeout(ctx, 100*time.Second)
defer cancel()
err = subscription.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) {
}

但是话又说回来，这并不能让我确定所有消息都已被处理。

请提出一个解决方案，以确保当 pub-sub 中没有剩余消息时，subscription.receive 函数停止。

正确答案

我已经在我以前的公司中实现了这一点（遗憾的是我不再拥有代码，它在我以前的公司 git 中......）。不过它确实有效。

原理如下

msg := make(chan *pubsub.Message, 1)
sub := client.Subscription(subID)
cctx, cancel := context.WithCancel(ctx)
go sub.Receive(cctx, func(ctx context.Context, m *pubsub.Message) {
    msg <- m
    })
for {
  select {
    case res := <-msg:
      fmt.Fprintf(w, "Got message: %q\n", string(res.Data))
      res.Ack()
  
    case <-time.After(3 * time.Second):
        fmt.Println("timeout")
        cancel()
    }
}

好了，本文到此结束，带大家了解了《读取 Google Pub Sub 中的所有消息，使用 Go 实现》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！