我正在嘗試從谷歌發(fā)布訂閱中的主題中獲取所有可用消息。但是在前進中,我無法找到一個配置,一旦Pub-Sub中沒有更多消息剩余,就可以取消接收回調(diào)。我認(rèn)為一種方法是使用Google云監(jiān)控API從Pub-Sub獲取消息總數(shù),該答案中描述了Google PubSub - 計算主題中的消息數(shù),然后保留已讀消息數(shù)的計數(shù),如果計數(shù)等于該數(shù)字,則調(diào)用取消,但我不太確定這是否是正確的方法。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) {}但話又說回來,這并不能確定所有消息都已得到處理。請建議一個可以確保該訂閱的解決方案。當(dāng) Pub-Sub 中不再有剩余消息時,接收功能將停止。
1 回答
慕桂英546537
TA貢獻1848條經(jīng)驗 獲得超10個贊
我已經(jīng)在我以前的公司中實現(xiàn)了它(可悲的是,我不再有代碼,它是在我以前的公司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()
}
}
- 1 回答
- 0 關(guān)注
- 128 瀏覽
添加回答
舉報
0/150
提交
取消
