考虑到每次轮询最大消息数和 Cron，轮询如何用于 FTP 入站通道适配器

我有 UC，我需要从 ftp 位置选取文件并将其放入服务器位置 我正在使用 ftp-inbound-channel-adapter （Spring 集成 - 2.0.4）来实现它。 下面是我的xml中的配置

 <bean id="ftpAASessionFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
          <property name="host" value="${ftp.session.host}" />
          <property name="port" value="${ftp.session.port}" />
          <property name="username" value="${ftp.session.username}" />
          <property name="password" value="${ftp.session.password}" />
          <property name="clientMode" value="0" />
          <property name="fileType" value="0" />
   </bean>


   <ftp:inbound-channel-adapter id="ftpAAInbound"
          channel="ftpChannel" session-factory="ftpAASessionFactory" charset="UTF-8"
          auto-create-local-directory="false" delete-remote-files="true"
          remote-directory="${ftp.source.location}" local-directory="file://${ftp.target.location}"
          >
          <int:poller max-messages-per-poll="5" cron="0 */2 * ? * *">
          </int:poller>

   </ftp:inbound-channel-adapter>

   <int:channel id="ftpChannel">
          <int:queue />
          <int:interceptors>
                 <int:wire-tap channel="debugLogger" />
          </int:interceptors>
   </int:channel>

   <int:logging-channel-adapter id="debugLogger"
          level="DEBUG" log-full-message="true" />

   <int:logging-channel-adapter id="errorLogger"
          level="ERROR" log-full-message="true" />

我已将 max-messages-per-poll 配置为 5，并且每隔偶数分钟进行一次轮询（使用 cron 表达式）。

我的问题是，如果我们在 ftp 位置有 6 个文件，则所有 6 个文件都会在第一次轮询时传输到服务器位置（根据每次轮询的最大消息数= 5，它应该只从 Ftp 位置选择 5 个文件）并且有效负载仅由 5 个文件组成。

我希望在第一次轮询时只将 5 个文件传输到我的服务器，在第二次轮询时它应该选择最后一个

请提出解决方案 TIA

当 ftp 位置有 6 个文件时 PFB 日志

    [CRA] [01/03/2017 12:38:00] DEBUG [task-scheduler-8] DefaultFtpSessionFactory.createClient(158) | Connected to server [prgrear01.group.root.ad:21]
***[CRA] [01/03/2017 12:38:00] INFO [task-scheduler-8] FtpSession.read(79) | File has been successfully transfered from: /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_808_2015_02_01_07_50_01_102_20.txt***
[CRA] [01/03/2017 12:38:00] DEBUG [task-scheduler-8] FtpInboundFileSynchronizer.copyFileToLocalDirectory(219) | deleted /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_808_2015_02_01_07_50_01_102_20.txt
[CRA] [01/03/2017 12:38:00] INFO [task-scheduler-8] FtpSession.read(79) | File has been successfully transfered from: /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_808_2017_02_22_07_50_01_102_02.txt
[CRA] [01/03/2017 12:38:00] DEBUG [task-scheduler-8] FtpInboundFileSynchronizer.copyFileToLocalDirectory(219) | deleted /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_808_2017_02_22_07_50_01_102_02.txt
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FtpSession.read(79) | File has been successfully transfered from: /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_809_2015_02_01_07_50_01_102_01.txt
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] FtpInboundFileSynchronizer.copyFileToLocalDirectory(219) | deleted /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_809_2015_02_01_07_50_01_102_01.txt
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FtpSession.read(79) | File has been successfully transfered from: /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_809_2015_02_01_07_50_01_102_21.txt
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] FtpInboundFileSynchronizer.copyFileToLocalDirectory(219) | deleted /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_809_2015_02_01_07_50_01_102_21.txt
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FtpSession.read(79) | File has been successfully transfered from: /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_809_2017_02_22_07_50_01_102_02.txt
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] FtpInboundFileSynchronizer.copyFileToLocalDirectory(219) | deleted /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_809_2017_02_22_07_50_01_102_02.txt
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FtpSession.read(79) | File has been successfully transfered from: /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_810_2017_02_22_07_50_01_102_02.txt
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] FtpInboundFileSynchronizer.copyFileToLocalDirectory(219) | deleted /FTPManifestSrcDev/R2_2a/FTPManifestSrcDev/ABC_810_2017_02_22_07_50_01_102_02.txt
***[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] FileReadingMessageSource.scanInputDirectory(272) | Added to queue: [D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt, D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt, D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt, D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt, D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt, D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt]
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FileReadingMessageSource.receive(260) | Created message: [[Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8]*** QueueChannel.preSend(224) | preSend on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.preSend(224) | preSend on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessage(67) | org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessageInternal(141) | [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.postSend(237) | postSend (sent=true) on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.postSend(237) | postSend (sent=true) on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_808_2015_02_01_07_50_01_102_20.txt][Headers={timestamp=1488352081732, id=46536ab1-c0bd-4cf4-9867-b7d99e462ed5}]
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FileReadingMessageSource.receive(260) | Created message: [[Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.preSend(224) | preSend on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.preSend(224) | preSend on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessage(67) | org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessageInternal(141) | [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.postSend(237) | postSend (sent=true) on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.postSend(237) | postSend (sent=true) on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_808_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081784, id=336045cf-0abd-4b1d-b698-d82c230e4b1f}]
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FileReadingMessageSource.receive(260) | Created message: [[Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.preSend(224) | preSend on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.preSend(224) | preSend on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessage(67) | org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessageInternal(141) | [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.postSend(237) | postSend (sent=true) on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.postSend(237) | postSend (sent=true) on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_01.txt][Headers={timestamp=1488352081786, id=75029ba5-4857-4a4e-832f-b8c657b539e3}]
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FileReadingMessageSource.receive(260) | Created message: [[Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.preSend(224) | preSend on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.preSend(224) | preSend on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessage(67) | org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessageInternal(141) | [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.postSend(237) | postSend (sent=true) on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.postSend(237) | postSend (sent=true) on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_809_2015_02_01_07_50_01_102_21.txt][Headers={timestamp=1488352081789, id=edea505f-37a2-4c96-8034-b3c74f55f9de}]
[CRA] [01/03/2017 12:38:01] INFO [task-scheduler-8] FileReadingMessageSource.receive(260) | Created message: [[Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.preSend(224) | preSend on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.preSend(224) | preSend on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessage(67) | org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] LoggingHandler.handleMessageInternal(141) | [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] DirectChannel.postSend(237) | postSend (sent=true) on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:38:01] DEBUG [task-scheduler-8] QueueChannel.postSend(237) | postSend (sent=true) on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_809_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352081792, id=5123c737-02d1-4846-9001-011796d92aa0}]
[CRA] [01/03/2017 12:40:00] INFO [task-scheduler-8] FileReadingMessageSource.receive(260) | Created message: [[Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] QueueChannel.preSend(224) | preSend on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] DirectChannel.preSend(224) | preSend on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] LoggingHandler.handleMessage(67) | org.springframework.integration.handler.LoggingHandler#0 received message: [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] LoggingHandler.handleMessageInternal(141) | [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] DirectChannel.postSend(237) | postSend (sent=true) on channel 'debugLogger', message: [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] QueueChannel.postSend(237) | postSend (sent=true) on channel 'ftpChannel', message: [Payload=D:\applications\files\local\ABC_810_2017_02_22_07_50_01_102_02.txt][Headers={timestamp=1488352200005, id=7a0a0ea6-e573-4981-9e2f-89ae0f646b50}]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] DefaultFtpSessionFactory.createClient(158) | Connected to server [prgrear01.group.root.ad:21]
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(91) | Poll resulted in Message: null
[CRA] [01/03/2017 12:40:00] DEBUG [task-scheduler-8] SourcePollingChannelAdapter.doPoll(101) | Received no Message during the poll, returning 'fal

.......

该日志仅显示 4 个文件；看起来它被配置为每次轮询 3 条消息，并且清楚地显示在 12:38 以及 1 和 12:40 发送了 3 个文件（没有找到第五个文件）。

Poll resulted in Message: null

EDIT

这是实现所需结果的另一种方法（使用出站网关）。该版本使用 Java DSL。

如果您更喜欢使用 XML 配置，则在中提供了类似的流程（以 XML 形式）ftp-样本 https://github.com/spring-projects/spring-integration-samples/tree/master/basic/ftp- 您必须在之间插入文件限制器ls网关和分路器。

@SpringBootApplication
public class So42528316Application {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = SpringApplication.run(So42528316Application.class, args);
        try {
            context.getBean(So42528316Application.class).runDemo();
        }
        finally {
            context.close();
        }
    }

    @Autowired
    private FileGateway fetchAndProcess;

    private void runDemo() throws Exception {
        Collection<Boolean> rmResults = this.fetchAndProcess.processFilesAt("so42528316");
        while (rmResults != null) {
            System.out.println("Processed " + rmResults.size() + " files");
            Thread.sleep(10_000);
            rmResults = this.fetchAndProcess.processFilesAt("so42528316");
        }
        System.out.println("No more files available");
    }

    @MessagingGateway(defaultRequestChannel = "flow.input", defaultReplyTimeout = "0")
    public interface FileGateway {

        Collection<Boolean> processFilesAt(String pattern);

    }

    @Bean
    public DefaultFtpSessionFactory sessionFactory() {
        DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
        factory.setHost("10.0.0.3");
        factory.setUsername("ftptest");
        factory.setPassword("ftptest");
        factory.setClientMode(FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE);
        return factory;
    }

    @Bean
    public IntegrationFlow flow() {
        return f -> f
                .handle(Ftp.outboundGateway(sessionFactory(), "ls", "payload"))
                .handle("so42528316Application", "limitFiles")
                .split()
                .handle(Ftp.outboundGateway(sessionFactory(), "get",
                                "payload.remoteDirectory + '/' + payload.filename")
                        .localDirectory(new File("/tmp", "so42528316")))
                .handle("so42528316Application", "process")
                .handle(Ftp.outboundGateway(sessionFactory(), "rm",
                        "headers['file_remoteDirectory'] + '/' + headers['file_remoteFile']"))
                .aggregate();
    }

    public List<FtpFileInfo> limitFiles(List<FtpFileInfo> files) {
        // Add any logic you want here, e.g. check if file already on local disk.
        if (files.size() == 0) {
            return null;
        }
        else if (files.size() > 2) {
            System.out.println("Reducing fetch list from " + files.size() + " to 2");
            return files.stream().limit(2).collect(Collectors.toList());
        }
        else {
            return files;
        }
    }

    public String process(File file) {
        System.out.println("Processing " + file);
        file.delete();
        return file.getName();
    }

}

Result:

Reducing fetch list from 3 to 2
Processing /tmp/so42528316/bar.txt
Processing /tmp/so42528316/baz.txt
Processed 2 files
Processing /tmp/so42528316/foo.txt
Processed 1 files
No more files available