如何在spring批处理和spring集成中从itemWriter编写输出通道？

Question

首先感谢您的关注，我结合了spring批处理和spring集成，我定义了一个工作流并从ftp适配器检索文件并发送到jobChannel，并使用spring批处理对其进行处理，我想写入输出通道并使用通道处理后，我的代码是：

     <int-ftp:outbound-gateway id="gatewayGET"
                                  local-directory-expression="'./backup/' +#remoteDirectory"
                                  session-factory="ftpSessionFactory"
                                  request-channel="toGetChannel"
                                  reply-channel="toProcessChannel"
                                  command="get"
                                  temporary-file-suffix=".writing"
                                  command-options="-P"
                                  expression="payload.remoteDirectory + '/' + payload.filename"/>
        <int:channel id="toProcessChannel">
            <int:interceptors>
                <int:wire-tap channel="logger2"/>
            </int:interceptors>
        </int:channel>
    <int:channel id="outboundJobRequestChannel"/>
    <int:channel id="replyJobChannel"/>
   <service-activator input-channel="jobLaunchReplyChannel"/>
 <int:transformer input-channel="toProcessChannel" output-channel="outboundJobRequestChannel">
        <bean class="ir.isc.macna.configuration.FileMessageToJobRequest">
            <property name="fileParameterName" value="fileName"/>
        </bean>
    </int:transformer>
    <batch-int:job-launching-gateway request-channel="outboundJobRequestChannel"
                                     reply-channel="jobLaunchReplyChannel"/>

我的作家代码是：

@Component
@StepScope
public class MacnaFileWriter implements ChunkMessageChannelItemWriter<FileInfo> {

    @Autowired
    @Qualifier("replyJobChannel")
    private MessageChannel messageChannel;
    @Override
    public void write(List<? extends FileInfo> list) throws Exception {

       // send Message to replyJobChannel channel with Send method
    }
}

并使用 ftp 适配器在服务器上写入文件：

<int-ftp:outbound-gateway session-factory="ftpSessionFactory"
                              request-channel="replyJob"
                              command="mput"
                              auto-create-directory="true"
                              expression="payload"
                              remote-directory-expression="payload.remoteDirectory + '/' + payload.filename + '.out'"
                              reply-channel="output"/>

这是运行批处理作业并使用结果的标准方式吗？

Answer 1

MessageChannel 有 send 方法。为此，您可以使用MessageBuilder 为您的payload 创建一个Message<?>。

因为它只是一个<channel>，所以与任何其他MessageChannel 消费没有区别。标准的方式定义一个消费者是这样的：

<service-activator input-channel="jobLaunchReplyChannel"/>

但是，当此通道特定于来自<batch-int:job-launching-gateway>（JobLaunchingGateway 类）的结果时，尚不清楚为什么要将消息发送到jobLaunchReplyChannel：

jobExecution = this.jobLaunchingMessageHandler.launch(jobLaunchRequest);

所以，jobExecution 被发送到jobLaunchReplyChannel。

请参阅 Spring Batch Integration 文档。

也许您需要AsyncItemWriter 或ChunkMessageChannelItemWriter？