Wie kann ich die neuesten Ereignisse mit dem Azure Event Hubs Go SDK (azeventhubs) nutzen?

Der

php-Editor Strawberry bietet Ihnen eine Anleitung zur Nutzung der neuesten Ereignisse mit dem Azure Event Hubs Go SDK (azeventhubs). Azure Event Hubs ist ein hoch skalierbarer Echtzeit-Datenübertragungsdienst, mit dem große Mengen an Ereignisdaten verarbeitet werden können. In diesem Leitfaden erfahren Sie, wie Sie das azeventhubs-Paket für die Interaktion mit Azure Event Hubs verwenden und wie Sie die Funktion „Neueste Ereignisse“ verwenden, um Echtzeitdaten abzurufen. Mithilfe der Anleitung in diesem Artikel können Sie Azure Event Hubs problemlos in Go-Anwendungen verwenden und die neuesten Ereignisfunktionen nutzen, um Echtzeitdaten abzurufen und zu verarbeiten.

Frageninhalt

Ich migriere von azure-event-hubs-go/v3 zum neueren azeventhubs Go SDK. In älteren Versionen des SDK gab es einen ReceiveOption-Parameter, mit dem ich angeben konnte, wo mit der Verarbeitung von Ereignissen begonnen werden soll.

Im neuen SDK verwende ich den folgenden Code, um den Prozessor zu initialisieren:

processor, err := azeventhubs.NewProcessor(
    e.ConsumerClient, 
    checkpointStore, 
    &azeventhubs.ProcessorOptions{
        UpdateInterval: time.Second, 
        Prefetch: 0, 
        StartPositions: azeventhubs.StartPositions{
            Default: azeventhubs.StartPosition{
                Latest: to.Ptr(true), 
                EnqueuedTime: to.Ptr(time.Now()), 
                Inclusive: true
            }
        }
    }
)

Allerdings ist mir aufgefallen, dass Ereignisse vom letzten Kontrollpunkt und nicht vom zuletzt gesendeten Ereignis verbraucht werden.

Was ich versucht habe: Ich habe die Beispiele ConsumingEventsUsingConsumerClient und ConsumingEventsWithCheckpoints ausprobiert, aber sie verhalten sich gleich und verbrauchen Ereignisse vom letzten Kontrollpunkt statt vom aktuellsten.

Meine Erwartungen: Ich möchte, dass der Prozessor die neuesten Ereignisse verwendet, die vom Gerät gesendet werden, das jede Sekunde eine Nachricht sendet. Wie kann ich dieses Verhalten mit dem azeventhubs Go SDK erreichen?

Lösung

Anfangs fiel es mir schwer, den zugrunde liegenden Mechanismus von AMQP zu verstehen. Ich freue mich jedoch, Ihnen mitteilen zu können, dass das Problem erfolgreich gelöst wurde.

var wg sync.WaitGroup
wg.Add(1)

for _, partition := range p.PartitionIDs {
    go func(partition string) {
        defer wg.Done()

        partitionClient, err := consumerClient.NewPartitionClient(partition, nil)
        if err != nil {
            panic(err)
        }

        receiveCtx, cancel := context.WithTimeout(context.TODO(), time.Second*30)
        defer cancel()

        for {
            events, err := partitionClient.ReceiveEvents(receiveCtx, 1, nil)

            if err != nil && !errors.Is(err, context.DeadlineExceeded) {
                panic(err)
            }

            for _, evt := range events {
                fmt.Printf("partition: %s\n", partition)
                fmt.Printf("Body: %s\n", string(evt.Body))
            }
        }
    }(partition)
}

wg.Wait()

Ich möchte dem Azure-Kundensupport-Team meinen Dank für die wertvolle Hilfe aussprechen.

Das obige ist der detaillierte Inhalt vonWie kann ich die neuesten Ereignisse mit dem Azure Event Hubs Go SDK (azeventhubs) nutzen?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!