Spring Integration

Question

La nostra applicazione è stata progettata utilizzando Spring Integration Framework. Il flusso completo delle azioni dei messaggi inizia con l'ascolto delle code per le quali sono stati utilizzati adattatori di messaggi JMS dopo quali endpoint della coda i.e. code sono stati definiti e ciascun endpoint viene elaborato dagli attivatori del servizio.

Siamo attualmente in fase di esecuzione, stiamo generando 200 richieste di messaggi. Inizialmente abbiamo osservato che i messaggi non venivano eseguiti in parallelo, dopo aver effettuato alcune letture ho capito che aggiungendo la proprietà concomitante-consumatore e max-concorrente-consumatore alla scheda listener guidata da messaggi JMS sarà possibile abilitare la modalità multi-threading. In effetti questo mi ha aiutato, ma ancora da qualche parte tra il processo continuo a vedere l'effetto Discussione singola. È dovuto al modo in cui è stato definito l'endpoint? Qual è il vantaggio dell'aggiunta della capacità della coda a ciascun endpoint? Pensi che aggiungendo capacità di coda a ogni definizione di endpoint di canale sarà di nuovo utile eseguire in modalità mutli-threading.

L'istantanea design come richiesto:

Answer 1

Sarebbe utile vedere l'esatta definizione dei canali.

Per impostazione predefinita, un canale Spring consuma il suo messaggio nella thread del mittente. In altre parole, è sincrono. Se si desidera che un canale consumi i messaggi in modo asincrono, è necessario specificare un TaskExecutor. Vedi http://static.springsource.org/spring/docs/3.0.5.RELEASE/reference/scheduling.html

Answer 2

Guardando il diagramma di flusso, sembra che il flusso abbia molti elementi a singolo filetto e può essere ottimizzato per essere molto più concorrente con un rendimento sperabilmente maggiore.

Per iniziare con l'adattatore di canale messaggio guidato (non si è mostrata la configurazione per questo), può essere configurato di avere più di 1 default dei consumatori, e può essere fatto per consumare un numero ragionevole di messaggio per consumare ciclo.

Oltre l'adattatore del canale guidato dal messaggio, il thread che inserisce il messaggio nel canale diretto 1, sfortunatamente eseguirà il resto del flusso in quanto non vi è alcun buffer altrove, quindi quando il messaggio viene inserito in "Direct Channel 1" ", invoca immediatamente il router nello stesso thread, quindi richiama l'attivatore del servizio e l'adattatore di posta o l'adattatore del canale in uscita JMS nello stesso thread. La modifica qui potrebbe essere introdurre un canale di accodamento invece del canale diretto 1, in questo modo il thread che sta consumando il messaggio posiziona semplicemente i messaggi nel canale della coda e viene quindi eseguito con esso.

Oltre il canale diretto 1 (modificato in Queue Channel 1), penso che possa essere singolo thread in base alla velocità o alla velocità del flusso, se si dice che la scheda di posta è lenta quindi Direct Channel 4 può essere fatto una coda canale anche, lo stesso con diretto canale 5

Potete vedere se questi cambiamenti evidenziati in aiuto grassetto con rendendo il flusso meglio

Answer 3

Per increas prestazioni posso suggerire di utilizzare un executorchannel con un esecutore compito in controllando il numero della dimensione del pool di thread. In questo modo si ha la situazione in cui quando un messaggio arriva nella coda jms il consumatore prende il messaggio e in un thread separato processa il flusso. Ricorda che in questo tipo di configurazione il lavoro multithreadng viene eseguito dal canale taskexecutor che eseguirà la ricezione del messaggio in un thread separato, per questo motivo hai pensato bene quale grado di multithreading desideri.

Per il canale dei messaggi in coda è infatti necessario un polling che esegue il polling sul canale per l'esecuzione, la capacità della coda è la capacità della coda atomica dietro le quinte.

è possibile configurare il canale esecutore del mattino in questo modo 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> 

o in questo modo in java-dsl

@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))) 
       ..... 

     }