Mule Aggregator - Streaming Aggregation

Question

Die Sammlung Aggregator im Mule 2.0-Framework verwendet, funktioniert ein bisschen wie folgt aus:

• Eine eingehende Router eine Sammlung von Nachrichten nimmt und es teilt sich in eine Reihe von kleineren Nachrichten bis - je kleinere Nachricht mit einer Korrelations-ID entsprechend der Elternnachricht gestempelt bekommen

• Diese Nachrichten fließen durch verschiedene Dienste

• Schließlich diese Meldungen in einem eingehenden Aggregator gelangen, die sammelt die Nachrichten basierend auf der Korrelations-ID der übergeordneten Nachricht und der Anzahl der erwarteten Nachrichten. Sobald alle erwarteten Nachrichten empfangen wurden, wird die Aggregationsfunktion aufgerufen und das Ergebnis wird zurückgegeben.

Jetzt funktioniert das gut, wenn die Anzahl der Nachrichten in einer Gruppe relativ klein ist. Sobald jedoch die Anzahl von Nachrichten in einer Gruppe ~ 100k groß wird, wird viel Speicher gebunden, der auf die Gruppe von Nachrichten wartet, die darauf warten, dass die späteren Nachrichten ankommen. Dies wird schlechter, wenn mehrere Gruppen gleichzeitig aggregiert werden.

Ein Weg um dieses Problem wäre, einen Streaming-Aggregator zu implementieren. In meinem Anwendungsfall summiere ich im Wesentlichen die verschiedenen Nachrichten basierend auf einem Schlüssel, und dies könnte durchgeführt werden, ohne alle Nachrichten in der Gruppe gleichzeitig sehen zu müssen. Ich möchte nur wissen, dass alle Nachrichten empfangen wurden, bevor das Ergebnis an den Endpunkt weitergeleitet wird.

Hört sich das nach einer vernünftigen Lösung für das Problem an?

Ist das schon irgendwo in Mule implementiert?

Gibt es bessere Möglichkeiten, dies zu tun?

Answer 1

Dies scheint ein vernünftiger Ansatz (ich bin kein Mule-Experte auf jeden Fall), und ich habe alle Mule-Dokumentation gelesen und denke nicht, dass es so etwas gibt (die Streaming-Unterstützung ist begrenzt zu ein paar Konnektoren und Transformatoren - es ist ziemlich einfach, dass es nur um einen InputStream geht. Nur ein paar Dinge in Mule streamen, also müssen Sie möglicherweise andere modifizierte Transformatoren (wenn Sie sie verwenden) diesen Stream haben. Sie würden einfach den Aggregator implementieren, der einen InputStream bereitstellt, und das Streaming starten, sobald eine fortlaufende Nachrichtensequenz vorliegt.

jedoch ein Satz in der Beschreibung „... alle Nachrichten empfangen worden ist, um den Endpunkt die Ergebnisse vor der Weiterleitung“ beunruhigend sein könnte. Dies ist natürlich der Zweck des Streams, es sei denn, Sie meinen, dass Sie (vermutlich in Ihrer Dienstkomponente) den Überblick behalten, dass Sie alles haben, bevor Sie das (vermutlich viel kleinere) verarbeitete Ergebnis weiterleiten.