GCP Pub / sub: использование горутин для запуска нескольких подписчиков в одном приложении

gcp.go

package gcp

import (
    "context"
    "path"
    "runtime"

    "google.golang.org/api/option"

    "cloud.google.com/go/pubsub"
)

// PubsubClient is the GCP pubsub service client.
var PubsubClient *pubsub.Client

// Initialize initializes GCP client service using the environment.
func Initialize(env, projectName string) error {
    var err error
    ctx := context.Background()
    credentialOpt := option.WithCredentialsFile(getFilePathByEnv(env))

    PubsubClient, err = pubsub.NewClient(ctx, projectName, credentialOpt)
    return err
}

// GetTopic returns the specified topic in GCP pub/sub service and create it if it not exist.
func GetTopic(topicName string) (*pubsub.Topic, error) {
    topic := PubsubClient.Topic(topicName)
    ctx := context.Background()
    isTopicExist, err := topic.Exists(ctx)

    if err != nil {
        return topic, err
    }

    if !isTopicExist {
        ctx = context.Background()
        topic, err = PubsubClient.CreateTopic(ctx, topicName)
    }

    return topic, err
}

// GetSubscription returns the specified subscription in GCP pub/sub service and creates it if it not exist.
func GetSubscription(subName string, topic *pubsub.Topic) (*pubsub.Subscription, error) {
    sub := PubsubClient.Subscription(subName)
    ctx := context.Background()
    isSubExist, err := sub.Exists(ctx)

    if err != nil {
        return sub, err
    }

    if !isSubExist {
        ctx = context.Background()
        sub, err = PubsubClient.CreateSubscription(ctx, subName, pubsub.SubscriptionConfig{Topic: topic})
    }

    return sub, err
}

func getFilePathByEnv(env string) string {
    _, filename, _, _ := runtime.Caller(1)

    switch env {
    case "local":
        return path.Join(path.Dir(filename), "local.json")
    case "development":
        return path.Join(path.Dir(filename), "development.json")
    case "staging":
        return path.Join(path.Dir(filename), "staging.json")
    case "production":
        return path.Join(path.Dir(filename), "production.json")
    default:
        return path.Join(path.Dir(filename), "local.json")
    }
}

main.go

package main

import (
    "context"
    "fmt"
    "log"
    "net/http"
    "runtime"
    "runtime/debug"
    "runtime/pprof"
    "time"

    "rpriambudi/pubsub-receiver/gcp"

    "cloud.google.com/go/pubsub"
    "github.com/go-chi/chi"
)

func main() {
    log.Fatal(http.ListenAndServe(":4001", Route()))
}

func Route() *chi.Mux {
    InitializeSubscription()
    chiRoute := chi.NewRouter()

    chiRoute.Route("/api", func(r chi.Router) {
        r.Get("/_count", func(w http.ResponseWriter, r *http.Request) {
            fmt.Fprintf(w, "Number of goroutines: %v", runtime.NumGoroutine())
        })

        r.Get("/_stack", getStackTraceHandler)
    })

    return chiRoute
}

func InitializeSubscription() {
    gcp.Initialize("local", "fifth-bonbon-277102")

    go pubsubHandler("test-topic-1", "test-topic-1-subs")
    go pubsubHandler("test-topic-2", "test-topic-2-subs")
    go pubsubHandler("test-topic-3", "test-topic-3-subs")
    // ....

    return
}

func getStackTraceHandler(w http.ResponseWriter, r *http.Request) {
    stack := debug.Stack()
    w.Write(stack)

    pprof.Lookup("goroutine").WriteTo(w, 2)
}

func pubsubHandler(topicID string, subscriptionID string) {
    topic, err := gcp.GetTopic(topicID)
    fmt.Println("topic: ", topic)
    if err != nil {
        fmt.Println("Failed get topic: ", err)
        return
    }

    sub, err := gcp.GetSubscription(subscriptionID, topic)
    fmt.Println("subscription: ", sub)
    if err != nil {
        fmt.Println("Get subscription err: ", err)
        return
    }

    err = sub.Receive(context.Background(), func(ctx context.Context, msg *pubsub.Message) {
        messageHandler(subscriptionID, ctx, msg)
    })
    if err != nil {
        fmt.Println("receive error: ", err)
    }
}

func messageHandler(subscriptionID string, ctx context.Context, msg *pubsub.Message) {
    defer func() {
        if r := recover(); r != nil {
            fmt.Println("recovered from panic.")
            msg.Ack()
        }
    }()

    fmt.Println("message of subscription: ", subscriptionID)
    fmt.Println("Message ID: ", string(msg.ID))
    fmt.Println("Message received: ", string(msg.Data))

    msg.Ack()
    time.Sleep(10 * time.Second)
}

Он отлично работает, когда у меня всего несколько pubsubHandler внутри InitializeSubscription. Но когда я добавил еще pubsubHandler в функцию инициализации (примерно 10 или более обработчиков), все стало интересно. Подтверждение никогда не достигает сервера pubsub, поэтому сообщение просто не подтверждается (я проверил AcknowledgeRequest в обозревателе метрик, и никаких запросов подтверждения не поступало). Таким образом, сообщение продолжает возвращаться к подписчику. Кроме того, когда я перезапускаю приложение, иногда оно не получает никаких сообщений, ни новых, ни неподтвержденных.