Spring Integration: опубликовать канал для подписки. Это асинхронно?

Question

Я использую канал публикации подписки после того, как входящий шлюз принимает сообщение String, чтобы отправить его в журнал для регистрации сообщения и трансформатора для преобразования сообщения. Я хочу, чтобы оба эти события происходили параллельно.

Мой вопрос очень прост: параллельный ли канал подписки на подписку в весеннем интеграционном режиме отправляет сообщения в его подсборки?

Ниже приведен фрагмент кода из источника spring-integration-context.xml.

<int:gateway id="gateway" service-interface="com.test.Gateway"> 
    </int:gateway> 
<int:publish-subscribe-channel id="publishsubscribechannel" /> 

<int:service-activator input-channel="publishsubscribechannel" 
     method="transformEvent" ref="transformer" output-channel="transformerreplychannel"> 
</int:service-activator> 
<int:service-activator input-channel="publishsubscribechannel" 
     method="logMessage" ref="logger"> 
</int:service-activator> 

Здесь трансформатор и регистратор являются 2 подписчиками в канале publishsubscribe. В этой настройке поток сообщений для логгера и трансформатора из шлюза будет происходить асинхронно по умолчанию? ... ИЛИ Мне нужно сделать некоторую другую конфигурацию для достижения того же.

Answer 1

По умолчанию он выполняется последовательно. Так что в вашем случае это будет трансформатор, затем logger. Если вы хотите, чтобы запустить его параллельно, необходимо указать задачи-исполнитель

<int:publish-subscribe-channel id="publishsubscribechannel" task-executor="executor" /> 
... 
<task:executor id="executor" pool-size="10" /> 

И с помощью проблемно-исполнителя, обработка сообщений выполняются асинхронно.

Answer 2

Как из документации Spring Integration

PublishSubscribeChannel

Реализация PublishSubscribeChannel передает все сообщения, отправленные на него все свои подписку обработчиков. Это чаще всего используется для отправки сообщений о событиях, основной ролью которых является уведомление в отличие от документов, которые обычно предназначены для обработки одним обработчиком. Обратите внимание, что PublishSubscribeChannel предназначен только для отправки. Поскольку он транслирует свои подписчики непосредственно при вызове метода send (Message), пользователи не могут опросить сообщения (он не реализует PollableChannel и поэтому не имеет метода receive()). Вместо этого любой абонент должен быть сам MessageHandler, и метод callMessage (Message) абонента будет вызываться по очереди.

До версии 3.0, ссылаясь на метод отправки в PublishSubscribeChannel, у которого нет подписчиков, возвращает false. При использовании в сочетании с MessagingTemplate генерируется исключение MessageDeliveryException. Начиная с версии 3.0, поведение изменилось так, что отправка всегда считается успешной, если присутствует хотя бы минимум подписчиков (и успешно обрабатывает сообщение). Такое поведение может быть изменено путем установки свойства minSubscribers, который по умолчанию 0.

[Примечание] Примечание Если используется TaskExecutor, только наличие правильного количества абонентов используются для этого определения, поскольку фактические обработка сообщения выполняется асинхронно.

Обратите внимание на примечание. Он упоминает, если используется TaskExecutor, обработка сообщений будет асинхронной.

Итак, да, вы должны добавить TaskExecutor для того, чтобы это было асинхронным.

Answer 3

... или с помощью JavaConfig

@Bean 
public MessageChannel publishsubscribechannel() { 

return new PublishSubscribeChannel(executor()); 
} 

@Bean 
public ThreadPoolTaskExecutor executor() { 

ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor(); 
pool.setCorePoolSize(10); 
pool.setMaxPoolSize(10); 
pool.setWaitForTasksToCompleteOnShutdown(true); 
return pool; 
}