Spring Batch и Spring Integration. Не могу настроить JobListener

Question

Я новенькая весной. В последнее время я пытаюсь сделать так, чтобы весенние партии и весенняя интеграция работали вместе. Я хочу, чтобы JobListener прослушивал сообщения, приходящие на определенный канал, и запускал Spring Batch Job.

Я нашел пример на github (https://github.com/chrisjs/spring-batch-scaling/tree/master/message-job-launch) и попытался каким-то образом настроить копирование Spring Batch и Spring Integration вместе, и это выглядит так:

<!--Incomming channel OneToOne-->
<int:channel id="requests-channel"/>

<!--For multiple consumers OneToMany-->
<int:publish-subscribe-channel id="reply-channel"/>

<!--Channel for file adapter-->
<int:channel id="file-adapter-reply-channel"/>

<int:channel id="statuses">
    <int:queue capacity="10"/>
</int:channel>


<int:channel id="jobLaunchReplyChannel"/>


<!--Intercept request-->
<int-http:inbound-gateway request-channel="requests-channel"
                          supported-methods="PUT"
                          path="/testData/setProfileDescription"
                          reply-timeout="30000"
                          reply-channel="reply-channel">
</int-http:inbound-gateway>


<!--Sending HTTP response back to user OR either 'no reply received within timeout'-->
<bean id="profileDescriptionActivator"
      class="ru.tcsbank.service.integrations.activators.ProfileDescriptionActivator"/>

<int:service-activator ref="profileDescriptionActivator"
                       input-channel="requests-channel"
                       output-channel="reply-channel"
                       method="httpMessageActivator"/>


<!--Write profile description to file-->
<bean id="custom-file-name-generator"
      class="ru.tcsbank.service.integrations.transformers_generators.ProfilesFileAdapterNameGenerator"/>
<file:outbound-channel-adapter channel="file-adapter-reply-channel"
                               directory="file:out"
                               filename-generator="custom-file-name-generator"/>


<bean id="dataSource" class="org.apache.commons.dbcp.BasicDataSource" lazy-init="true" destroy-method="close">
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://localhost:3306/testdb"/>
    <property name="username" value="test_user"/>
    <property name="password" value="qwerty123"/>
</bean>


<bean id="stepScope" class="org.springframework.batch.core.scope.StepScope">
    <property name="autoProxy" value="true"/>
</bean>

<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    <property name="dataSource" ref="dataSource"/>
</bean>

<bean id="jobRepositoryInDB" class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean">
    <property name="dataSource" ref="dataSource"/>
    <property name="transactionManager" ref="transactionManager"/>
</bean>


<bean id="itemProcessor" class="ru.tcsbank.service.batch_processing.CustomItemProcessor"/>

<bean id="itemReader" class="ru.tcsbank.service.batch_processing.CustomReader" scope="step">
    <property name="resource" value="classpath:fileOut/*.csv" />
    <property name="lineMapper">
        <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
            <property name="lineTokenizer">
                <bean class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
                    <property name="delimiter" value=","/>
                    <property name="names" value="id,firstName,lastName"/>
                </bean>
            </property>
            <property name="fieldSetMapper">
                <bean class="ru.tcsbank.service.batch_processing.ProfileDescriptionLineMapper"/>
            </property>
        </bean>
    </property>
</bean>
<bean id="itemWriter" class="ru.tcsbank.service.batch_processing.CustomWriter"/>

<batch:job id="helloWorldJob" job-repository="jobRepositoryInDB">
    <batch:listeners>
        <batch:listener ref="jobListener"/>
    </batch:listeners>
    <batch:step id="step1">
        <batch:tasklet>
            <batch:chunk reader="itemReader" writer="itemWriter" processor="itemProcessor" commit-interval="10"/>
        </batch:tasklet>
    </batch:step>
</batch:job>


<int:transformer input-channel="reply-channel" output-channel="file-adapter-reply-channel">
    <bean class="ru.tcsbank.service.batch_processing.FileMessageToJobRequest">
        <property name="job" ref="helloWorldJob"/>
        <property name="fileParameterName" value="input.file.name"/>
    </bean>
</int:transformer>


<bean id="jobListener" class="ru.tcsbank.service.batch_processing.CustomJobExecutionListener">
    <constructor-arg index="0" ref="notificationSender"/>
</bean>

<batch-int:job-launching-gateway request-channel="reply-channel"
                                 reply-channel="file-adapter-reply-channel"/>

<int:logging-channel-adapter channel="jobLaunchReplyChannel"/>

<int:channel id="notificationsChannel"/>
<int:gateway id="notificationSender"
             service-interface="ru.tcsbank.service.batch_processing.NotificationSender"
             default-request-channel="notificationsChannel"/>

Я ожидаю, что мой helloWorldJob будет запущен, когда (как я правильно понимаю) мой jobListener получит сообщение от notificationsChannel. Но он не работает (не получает сообщение от notificationsChannel). Затем он выдает ошибку вроде:

У Dispatcher нет подписчиков на канал 'Application.notificationsChannel' .; вложенное исключение:> org.springframework.integration.MessageDispatchingException: у Dispatcher> нет подписчиков, failMessage = GenericMessage [payload = TEST. > Задание обработки изображения выполнялось в течение: 0 минут 0 секунд.

Answer 1

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

В примере, который вы упоминаете в ссылке, у нас есть что-то вроде этого:

<int-jms:outbound-channel-adapter id="notifications" destination-name="notifications"
              channel="notificationsChannel"/>

Итак, сообщения, отправленные в notificationsChannel, помещаются в очередь notifications в JMS-брокере. Ваш образец протекает такому подписчику. Поэтому я могу только объяснить причину исключения, но определенно не могу сказать вам, что делать.

UPDATE

Вы не можете использовать notificationSender в своем решении. Похоже, это просто результат CustomJobExecutionListener. Итак, если вам не нужно прослушивать процесс работы, просто удалите этот CustomJobExecutionListener и, следовательно, это объявление notificationSender вместе с определением notificationsChannel.

Все остальное, что вы спрашиваете в комментариях, выходит за рамки этого SO вопроса. Пожалуйста, подумайте над тем, чтобы поднять эти проблемы в отдельной ветке SO.