Spring Integration

Question

Unsere Anwendung wurde mit Spring Integration Framework entwickelt. Der vollständige Nachrichtenaktionsfluss beginnt mit dem Abhören von Warteschlangen, für die JMS-nachrichtengesteuerte Adapter verwendet wurden, nachdem kanalbasierte, d. H. Warteschlangenendpunkte, definiert wurden und jeder Endpunkt von Dienstaktivatoren verarbeitet wird.

Wir sind derzeit in der Leistungsphase, wir erzeugen 200 Nachricht Anfrage. Anfangs haben wir beobachtet, dass Nachrichten nicht parallel ausgeführt wurden, nachdem wir einige Lesevorgänge durchgeführt haben, und herausgefunden haben, dass durch das Hinzufügen von "concurrent-consumer" und "max-concurrent-consumer" -Eigenschaft zu JMS der nachrichtengesteuerte Listener-Adapter dazu beiträgt, den Multi-Threading-Modus zu aktivieren. In der Tat hat dies geholfen, aber immer noch irgendwo zwischen dem Prozess sehe ich immer noch Single-Thread-Effekt. Liegt das daran, dass der Endpunkt definiert wurde? Welchen Vorteil bietet das Hinzufügen von Warteschlangenkapazität zu jedem Endpunkt? Glauben Sie, dass das Hinzufügen von Warteschlangenkapazität zu jeder Channel-Endpunktdefinition wieder dazu beiträgt, im Modus für mehrere Threads zu laufen?

Das Design Snapshot wie gewünscht:

Answer 1

Es wäre hilfreich sein, die genaue Definition Ihrer Kanäle zu sehen.

Standardmäßig verwendet ein Spring-Kanal seine Nachricht im Thread des Absenders. Mit anderen Worten, es ist synchron. Wenn ein Kanal Nachrichten asynchron konsumieren soll, müssen Sie einen TaskExecutor angeben. Siehe http://static.springsource.org/spring/docs/3.0.5.RELEASE/reference/scheduling.html

Answer 2

Wenn Sie Ihr Flussdiagramm betrachten, sieht es so aus, als ob der Fluss viele einzelne Threaded-Elemente enthält und so optimiert werden kann, dass er mit einem hoffentlich höheren Durchsatz viel mehr übereinstimmt.

mit der Nachricht angetrieben Kanaladapter zu starten (Sie haben die Konfiguration für das nicht gezeigt), kann konfiguriert werden, um mehr als 1 Standard Verbraucher, haben und kann gemacht werden, um eine angemessene Anzahl von Nachrichten verbrauchen pro verbrauchen Zyklus.

Nach dem Message Driven Channel Adapter, der Thread, der die Nachricht in den direkten Kanal 1 setzt, wird der Rest des Flusses leider laufen, da es keine Pufferung irgendwo anders gibt, also wenn Ihre Nachricht in "Direct Channel 1 ", ruft es sofort den Router im selben Thread auf und ruft dann den Service-Aktivator und den Mail-Adapter oder den JMS Outbound-Channel-Adapter im selben Thread auf. Die Änderung könnte hier sein, um einen Warteschlangenkanal anstelle des direkten Kanals 1 einzuführen, auf diese Weise wird der Thread, der die Nachricht verbraucht, nur die Nachrichten in den Warteschlangenkanal stellen, ist dann damit erledigt.

Über direkten Kanal 1 (geändert in Warteschlange Kanal 1), ich denke, kann single thread basierend darauf, wie schnell oder langsam Ihr Fluss ist, wenn sagen, der Mail-Adapter ist langsam dann kann Direct Channel 4 eine Warteschlange gemacht werden Kanal auch, das gleiche mit Direct-Channel 5

können Sie sich bitte, ob diese Änderungen mit der Herstellung der Strömung in fett Hilfe hervorgehoben besser

Answer 3

Denn ich increas Leistung Verwendung einen executorchannel mit einer Aufgabe Testamentsvollstrecker vorschlagen kann, in welche die Anzahl der Thread-Pool-Größe steuern. Auf diese Weise haben Sie die Situation, in der, wenn eine Nachricht in der JMS-Warteschlange ankommt, der Verbraucher die Nachricht und in einem separaten Thread den Ablauf übernimmt. Denken Sie daran, dass bei dieser Art von Konfiguration die Multithread-Arbeit vom TaskExecutor-Kanal ausgeführt wird, der das Empfangen der Nachricht in einem separaten Thread ausführt, aus diesem Grund haben Sie gut überlegt, welche Klasse von Multithreading Sie wollen.

Für den Warteschlangen-Nachrichtenkanal in der Tat benötigen Sie einen Poller, der auf dem Kanal für den Empfang abfragen, die Warteschlangenkapazität ist die Kapazität der atomischen Warteschlange hinter den Kulissen.

Sie können auf diese Weise in xml

<?xml version="1.0" encoding="UTF-8"?> 
<beans xmlns="http://www.springframework.org/schema/beans" 
     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
     xmlns:si="http://www.springframework.org/schema/integration" 
     xmlns:tx="http://www.springframework.org/schema/task" 
     xsi:schemaLocation="http://www.springframework.org/schema/beans 
     http://www.springframework.org/schema/beans/spring-beans.xsd 
     http://www.springframework.org/schema/integration 
     http://www.springframework.org/schema/integration/spring-integration.xsd 
     http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"> 

     <tx:executor id="taskExecutor" pool-size="10" queue-capacity="10"/> 
     <si:channel id="ch" > 
       <si:dispatcher task-executor="taskExecutor"/>  
     </si:channel> 
</beans> 

oder auf diese Weise in java-dsl am Testamentsvollstrecker Kanal konfigurieren

@Bean 
    public IntegrationFlow storeBookPageByPage(ConnectionFactory connectionFactory, 
               @Qualifier("createBookQueue") Destination createBookQueue, 
               @Qualifier("createBookResultQueue") Destination createBookResultQueue, 
               PdfBookMasterRepository pdfBookMasterRepository, 
               BookRepository bookRepository){ 
     String tempFilePathBaseDir = environment.getProperty("bookService.storeBookPageByPage.tempFilePathBaseDir"); 

     return IntegrationFlows.from(Jms.messageDriverChannelAdapter(connectionFactory) 
       .destination(createBookQueue) 
       .errorChannel(storeBookPageByPageErrorChannel())) 
       .channel(channels -> channels.executor(Executors.newScheduledThreadPool(5))) 
       ..... 

     }