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

Question

我是 Spring 的新人。最近我确实尝试让 Spring Batch 和 Spring Integration 一起工作。我想要 JobListener 来监听来自特定 channel 的消息并启动 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 has no subscribers for channel 'application.notificationsChannel'.; nested exception is >org.springframework.integration.MessageDispatchingException: Dispatcher >has no subscribers, failedMessage=GenericMessage [payload=TEST. >Image processing job ran for: 0 minutes, 0 seconds.

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定义。

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