Verwenden der benutzerdefinierten Handler-Nats-Golang-Beibehaltungs-Abonnementmethode

php-Editor Banana stellt Ihnen in diesem Artikel vor, wie Sie einen benutzerdefinierten Handler Nats Golang verwenden, um die Abonnementmethode beizubehalten. Während des Entwicklungsprozesses müssen wir häufig Nachrichtenwarteschlangen verwenden, um asynchrone Aufgaben abzuwickeln. Nats Golang ist ein leichtes Nachrichtenwarteschlangensystem mit hoher Leistung und Skalierbarkeit. Durch die Anpassung des Handlers können wir die Abonnementmethode beibehalten und eine flexiblere Nachrichtenverarbeitung und Prozesssteuerung erreichen. Im Folgenden stellen wir detailliert vor, wie diese Funktion in Nats Golang implementiert wird.

Frageninhalt

Ich schreibe einen Wrapper über dem NATs-Client in Golang und möchte eine Handlerfunktion erhalten, die ich vom Verbraucher aufrufen kann, sobald ich eine Nachricht vom NATs-Server erhalte. Ich möchte die benutzerdefinierte Abonnementmethode beibehalten, bis die Nachricht von nats eingeht.

Veröffentlicht von:

func (busconfig busconfig) publish(service string, data []byte) error {
    puberr := conn.publish(service, data)
    if puberr != nil {
        return puberr
    }
    return nil
}

Abonnieren:

func (busconfig busconfig) subscribe(subject string, handler func(msg []byte)) {
    fmt.println("subscrbing on : ", subject)

    //wg := sync.waitgroup{}
    //wg.add(1)
    subscription, err := conn.subscribe(subject, func(msg *nats.msg) {
        go func() {
            handler(msg.data)
        }()
        //wg.done()
    })
    if err != nil {
        fmt.println("subscriber error : ", err)
    }
    //wg.wait()
    defer subscription.unsubscribe()

}

Testfall:

func TestLifeCycleEvent(t *testing.T) {
    busClient := GetBusClient()
    busClient.Subscribe(SUBJECT, func(input []byte) {
        fmt.Println("Life cycle event received :", string(input))
    })

    busClient.Publish(SUBJECT, []byte("complete notification"))
}

Ich sehe, dass die Nachricht veröffentlicht, aber nicht abonniert ist. Ich habe versucht, Waitgroup zu verwenden, um die Abonnementmethode beizubehalten, aber ich denke, das ist nicht die richtige Lösung.

Problemumgehung

Sie können die zugestellte Nachricht nicht sehen, da Subscribe eine asynchrone Methode ist, die eine Goroutine erzeugt, um die eingehende Nachricht zu verarbeiten und den Rückruf aufzurufen.

Rufen Sie busClient.Publish() 之后，您的应用程序立即退出。它不会等待 Subscribe() alles an, was intern passiert.

Wenn Sie nats.Subscribe() verwenden, haben Sie normalerweise eine Anwendung mit langer Laufzeit, die unter bestimmten Bedingungen beendet wird (z. B. beim Empfang eines Signals zum Herunterfahren). WaitGroup wird hier funktionieren, aber wahrscheinlich nicht für echte Anwendungen, sondern nur zum Testen.

Sie sollten auch die Methode Flush() auf der NATS-Verbindung aufrufen, um sicherzustellen, dass alle gepufferten Nachrichten gesendet werden, bevor Sie das Programm verlassen.

Wenn Sie eine synchronisierte Methode wünschen, können Sie nats.SubscribeSync()

verwenden

Beispiel ansehen: https://natsbyexample.com/examples/messaging/publish-subscribe/execute

Das obige ist der detaillierte Inhalt vonVerwenden der benutzerdefinierten Handler-Nats-Golang-Beibehaltungs-Abonnementmethode. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!