go - 如何使用GCP在pubsub模型中一次向所有订阅者发送消息

Question

使用谷歌云平台实现pubsub 模型，并使用函数创建主题，订阅者，发布和pullmsg函数。

func pullMsgs(projectID, subID string, jsonPath string) error {
    ctx := context.Background()
    client, err := pubsub.NewClient(ctx, projectID, option.WithCredentialsFile(jsonPath))
    if err != nil {
        return fmt.Errorf("pubsub.NewClient: %v", err)
    }

    // Consume 10 messages.
    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))
        fmt.Println("Got message: n", string(msg.Data))
        msg.Ack()
        received++
        if received == 10 {
            cancel()
        }
    })
    if err != nil {
        return fmt.Errorf("Receive: %v", err)
    }
    return nil
}

pullmsg函数使用订阅ID从发布者获取发布的消息。
假定该模型有3个针对特定主题的订阅者。如果发布者为该主题发布了消息。 pullmsg函数必须执行3次才能为所有订阅者获取该消息。 是否有任何方法可以一次性将发布的消息发送给所有订阅者。

Answer 1

我不明白为什么您确实要一次从3个订阅中提取。由于这些消息可能不会同时到达/大约在同一时间到达。因此，如果要合并这些消息，则代码必须等待，直到收到所有消息。这实际上不是Pub/Sub的目的。另一方面，如果您不想让合并这三则消息，那么我建议您为每个订阅创建一个单独的代码段。
话虽如此，您可以使用同步提取机制或异​​步提取机制来提取所有消息。如果您不希望代码被阻塞，从而使其基本上同时监听所有订阅，则可以使用异步请求。
代码流可能是:

创建一个非阻塞订户函数。

当x条消息到达时(中断while循环)，让此pull函数停止。

创建传递您的三个主题ID的函数的三个实例。

创建另一个函数供您处理。

可以找到更多信息here。