Spring Batch 和 Spring 集成。无法配置 JobListener

Question

我是春天的新人。最近我确实尝试让 Spring Batch 和 Spring Integration 一起工作。我想让 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 >没有订阅者， failedMessage=GenericMessage [payload=TEST. >图像处理作业运行时间：0 分钟，0 秒。

Answer 1

很难理解您希望使用所有这些自定义代码实现什么目标，但我可以说的是，您的配置中没有该 notificationsChannel 的订阅者。您确实通过notificationSender 网关向它发送消息，但您没有提供任何端点来使用该notificationsChannel。

在您在链接中提到的示例中，我们有这样的内容：

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

因此，发送到notificationsChannel 的消息将到达 JMS 代理上的 notifications 队列中。您的样本正在泄漏这样的订阅者。因此我只能解释异常的原因，但绝对不能告诉你做什么。

更新

您不得在解决方案中使用notificationSender。看起来这只是CustomJobExecutionListener 的结果。因此，如果您不需要侦听作业流程，只需删除 CustomJobExecutionListener，因此，将 notificationSender 声明与 notificationsChannel 定义一起删除。

您在 cmets 中提出的所有其他问题都超出了这个 SO 问题的范围。请考虑在单独的 SO 线程中提出这些问题。