Spring Integration Aggregator kombiniert mit Publish Subscriber

Question

Ich habe grundlegende Frage in Spring-Integration. Ich habe versucht, in anderen Threads zu suchen, war aber nicht mit den Antworten zufrieden. Also hier sind meine Fragen. Jeder Tipp wird sehr geschätzt.

Ich versuche, Daten aus 4 verschiedenen Quellen parallel zu sammeln und die Daten zu aggregieren.

Mein Verständnis (nach einem der Threads, die ich zuvor gelesen habe), ist jeder Ausgangskanal eines Splitters, wenn es sich um einen direkten Kanal handelt, dann wird die Anfrage sequentiell weitergeleitet.

Aber Ausgangskanal des Splitter ist ein Empfängerlisten-Router. Beim Debuggen wurde die Anfrage sequenziell weitergeleitet. Nach einer Analyse habe ich herausgefunden, dass wir den Payload-Typ-Router parallel verarbeiten können.

Also änderte ich den Router zum Payload-Typ-Router und immer noch wird die Anfrage der Reihe nach geroutet.

Also änderte ich meine Strategie und verwendete einen Publish-Subscriber Channel und verband die Ausgabe aller abonnierenden Kanäle mit einem Aggregator, so dass ich die Ergebnisse aggregieren konnte. Ich konnte die Daten parallel bereitstellen, aber ich sehe nicht, dass die Antwort zu aggregiert weitergeleitet wird. Mache ich es falsch? Bitte vorschlagen!

<int:publish-subscribe-channel id="PublishSubscriberChannel" 
    task-executor="taskExecutor" /> 

<task:executor id="taskExecutor" pool-size="5" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource1" output-channel="lookupAggregatorChannel" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource2" output-channel="lookupAggregatorChannel" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="getVehicleInfoWithAdditionalAttributes" 
    output-channel="lookupAggregatorChannel" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource4" output-channel="lookupAggregatorChannel" /> 

Answer 1

So Schließlich fand ich, was ich falsch mache. Zunächst arbeitete ich an Spring Integration Version 2.0. Dies ist eine alte Anwendung. Wenn mein Code genau dem in der Frage entspricht, konnte ich nicht sehen, dass die Anfrage an den Aggregator weitergeleitet wird. Als ich dann die Spring Version auf 4.3.2 RELEASE umstellte, bekam ich den Fehler, dass Correlation Strategy null sein könnte. Das ist, wenn ich herausgefunden habe, dass der Aggregator eine Korrelations-ID benötigt, und wenn ich Splitter hatte, wird die Korrelations-ID automatisch hinzugefügt und für den Publish-Subscribe-Kanal müssen wir apply-sequence = "true" hinzufügen, um eine Standard-Korrelations-ID hinzuzufügen vom Aggregator verwendet werden. Im Wesentlichen sieht mein Code jetzt so aus und funktioniert vollkommen in Ordnung.

<int:publish-subscribe-channel id="PublishSubscriberChannel" apply-sequence="true" 
    task-executor="taskExecutor" /> 

<task:executor id="taskExecutor" pool-size="5" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource1" output-channel="lookupAggregatorChannel" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource2" output-channel="lookupAggregatorChannel" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource3" 
    output-channel="lookupAggregatorChannel" /> 

<int:service-activator input-channel="PublishSubscriberChannel" 
    ref="lookUpService" method="lookupSource4" output-channel="lookupAggregatorChannel" />