我使用“int-file: inbound-channel-adapter”来加载目录中存在的文件。我喜欢按顺序处理文件:这意味着当第一个文件的处理完成时,我加载第二个文件......等等。
I see a sample https://github.com/spring-projects/spring-integration-samples/blob/master/intermediate/file-processing/src/main/java/org/springframework/integration/samples/fileprocessing/FileProcessor.java但我无法预测处理一个文件所需的时间,这取决于文件的大小。
我的源代码:
<int-file:inbound-channel-adapter
directory="${directory.files.local}" id="filesIn" channel="channel.filesIn">
<int:poller fixed-delay="1000" max-messages-per-poll="1" />
</int-file:inbound-channel-adapter>
一个文件的处理过程是 file:inbound-channel-adapter--->transformer--->splitter---->http:outbound-gateway--->outbound-mail-adapter---->的处理一个文件处理完毕,那么此时,我要处理下一个文件。
我的项目配置太复杂了。
下面,我向您展示更多配置:
配置的第一部分是:
<int-sftp:inbound-channel-adapter id="sftpInbondAdapter"
auto-startup="true" channel="receiveChannel" session-factory="sftpSessionFactory"
local-directory="file:${directory.files.local}" remote-directory="${directory.files.remote}"
auto-create-local-directory="true" delete-remote-files="true"
filename-regex=".*\.txt$">
<int:poller fixed-delay="${sftp.interval.request}"
max-messages-per-poll="-1" />
</int-sftp:inbound-channel-adapter>
<!-- <int:poller cron="0 * 17 * * ?"></int:poller> -->
<int-file:inbound-channel-adapter
filter="compositeFileFilter" directory="${directory.files.local}" id="filesIn"
channel="channel.filesIn" prevent-duplicates="true">
<int:poller fixed-delay="1000" max-messages-per-poll="1" />
</int-file:inbound-channel-adapter>
<int:transformer input-channel="channel.filesIn"
output-channel="channel.file.router" ref="fileTransformer" method="transform" />
<int:recipient-list-router id="fileRouter"
input-channel="channel.file.router">
<int:recipient channel="channel.empty.files"
selector-expression="payload.length()==0" />
<int:recipient channel="channel.filesRejected"
selector-expression="payload.toString().contains('rejected')" />
<int:recipient channel="toSplitter"
selector-expression="(payload.length()>0) and(!payload.toString().contains('rejected'))" />
</int:recipient-list-router>
然后从通道 tosplitter 中,我的程序逐行读取文件:
<int-file:splitter input-channel="toSplitter"
output-channel="router" requires-reply="false" />
<int:recipient-list-router id="recipentRouter"
input-channel="router">
<int:recipient channel="channelA"
selector-expression="headers['file_name'].startsWith('${filenameA.prefix}')" />
<int:recipient channel="channelB"
selector-expression="headers['file_name'].startsWith('${filenameB.prefix}')" />
</int:recipient-list-router>
每个通道 A 和 B 应该为每条线路调用两个不同的 WAYS。
每个文件都使用异步调用 ws 代码,文件 A 如下:
<int:header-enricher input-channel="channelA"
output-channel="channelA.withHeader">
<int:header name="content-type" value="application/json" />
<int:header name="key1" expression="payload.split('${line.column.separator}')[0]" />
<int:header name="key2" expression="payload"></int:header>
</int:header-enricher>
<int:transformer input-channel="channelA.withHeader"
output-channel="channelA.request" ref="imsiMsgTransformer"
method="transform">
</int:transformer>
<int:channel id="channelA.request">
<int:queue capacity="10" />
<int-http:outbound-gateway id="maspUpdatorSimChangedGateway"
request-channel="channelA.request"
url="${url}"
http-method="PUT" expected-response-type="java.lang.String" charset="UTF-8"
reply-timeout="${ws.reply.timeout}" reply-channel="channelA.reply">
<int-http:uri-variable name="foo" expression="headers['key1']" />
<int:poller fixed-delay="1000" error-channel="channelA.error"
task-executor="executorA" />
<int-http:request-handler-advice-chain>
<int:retry-advice max-attempts="${ws.max.attempts}"
recovery-channel="recovery.channelA">
<int:fixed-back-off interval="${ws.interval.attempts}" />
</int:retry-advice>
</int-http:request-handler-advice-chain>
</int-http:outbound-gateway>
<int:service-activator input-channel="recovery.channelA"
ref="updateImsiHttpResponseErrorHandler" method="handleMessage" output-channel="updateImsi.channel.error.toenricher">
</int:service-activator>
<int:service-activator input-channel="channelA.reply"
ref="updateImsiHttpResponseMessageHandler" method="handleMessage">
<int:poller fixed-delay="1000"></int:poller>
</int:service-activator>
在(回复通道和恢复通道)的每个激活器中,我计算文件的进度,直到文件完成,此时我应该加载第二个文件 A2 或文件 B ...等