分布式模式下的spring集成ftp入站适配器

Question

我正在使用Spring Integration: FTP Inbound Channel Adapter 从远程 FTP 服务器读取文件。我的问题是，它每天能处理大约 500 万个文件吗？

如果我以分布式模式将我的项目战争部署在 2 个不同的服务器上，那会不会有问题？因为在两台服务器上FTP Inbound Channel Adapter 都将运行。然后两个适配器将读取相同的文件两次。 请帮我在分布式模式下设置这个系统。

编辑：
我已经在 2 台服务器上设置了我的 Spring Integration 项目战争。它正在使用FTP Inbound Channel Adapter。两台服务器的适配器remote-location 都指向同一个ftp 文件位置。当我启动两个服务器时，两个服务器都开始传输相同的文件并多次生成消息。我根据Gary's 的建议使用Redis MetaData Store。 我在两台服务器上的Ftp Inbound Channel Adapter 如下所示：

<bean id="redisMessageStore" class="org.springframework.integration.redis.store.RedisMessageStore">
    <constructor-arg ref="redisConnectionFactory"/>
</bean>

<bean name="metadataStore" class="org.springframework.integration.redis.metadata.RedisMetadataStore">
    <constructor-arg name="connectionFactory" ref="redisConnectionFactory"/>
</bean>

<bean id="fileSystemPersistantFilter" class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter"> 
    <constructor-arg name="store" ref="metadataStore"/> <constructor-arg name="prefix" value="" />
</bean>

<bean id="ftpPersistantFilter" class="org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter"> 
    <constructor-arg name="store" ref="metadataStore"/> <constructor-arg name="prefix" value="" />
</bean>
<int-ftp:inbound-channel-adapter id="ftpInboundAdapter"
    session-factory="ftpClientFactory" channel="ftpChannel"
    filter="ftpPersistantFilter"
    local-filter="fileSystemPersistantFilter" delete-remote-files="false"
    remote-directory="${ftp.remote_directory}/test/" local-directory="${ftp.local_directory}/test/"
    temporary-file-suffix=".writing" auto-create-local-directory="true">
    <int:poller fixed-rate="1000" max-messages-per-poll="-1" />
</int-ftp:inbound-channel-adapter>

第一台服务器的输出日志是：

19-Feb-2016 10:34:41.634 INFO [task-scheduler-1] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=1793c207-2d8a-542c-c5a7-eac9165e4cc5, timestamp=1455858281634}]]
19-Feb-2016 10:34:42.886 INFO [task-scheduler-4] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=c909b6cc-9f78-2f6f-2a27-036f0186b959, timestamp=1455858282886}]]
File /home/harsh/test/test_input_file1.txt transformed by 1st war 1793c207-2d8a-542c-c5a7-eac9165e4cc5
File /home/harsh/test/test_input_file1.txt transformed by 1st war c909b6cc-9f78-2f6f-2a27-036f0186b959
19-Feb-2016 10:34:47.892 INFO [task-scheduler-4] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=8c5c8941-fbfd-91d8-9a25-75d46e450930, timestamp=1455858287892}]]
19-Feb-2016 10:34:49.325 INFO [task-scheduler-2] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=dbdddd0f-1ac5-0753-8873-f0f9c77cb48b, timestamp=1455858289325}]]
Service Activator /home/harsh/test/test_input_file1.txt 1st war 24632436-d297-db0c-c9ea-ac596c57a91e
19-Feb-2016 10:34:50.372 INFO [task-scheduler-2] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=5cc843ae-c1d7-814f-b9fd-a7c5c2515674, timestamp=1455858290372}]]
19-Feb-2016 10:34:51.759 INFO [task-scheduler-2] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=428ba015-e2f3-6948-fc13-ca0df31ee9c0, timestamp=1455858291759}]]
19-Feb-2016 10:34:53.670 INFO [task-scheduler-2] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=ac1fca37-838f-39fc-f9ed-cc373f8f8b12, timestamp=1455858293670}]]
19-Feb-2016 10:34:55.648 INFO [task-scheduler-8] org.springframework.integration.file.FileReadingMessageSource.receive Created message: [GenericMessage [payload=/home/harsh/test/test_input_file1.txt, headers={id=f9197ec2-e73a-19be-e94b-94bffe515569, timestamp=1455858295647}]]
File /home/harsh/test/test_input_file1.txt transformed by 1st war 45718961-2a99-d368-d88a-9bc2ceb955cd

第二台服务器正在生成具有不同消息 ID 的相同日志。 我错过了什么吗？
我需要为此编写自定义过滤器吗？

Answer 1

我的问题是，它每天能处理大约 500 万个文件吗？

这取决于文件的大小和网络的带宽；使用 Spring Integration 不太可能是一个因素。

不过，您可能应该在处理后在本地删除文件，以避免大目录扫描。

为避免集群中出现重复，您需要使用由共享元数据存储支持的FtpPersistentAcceptOnceFileListFilter，这样每个实例都会跳过其他实例处理的文件。

更多信息请参见the documentation。

编辑：

我刚刚使用您的配置进行了测试，没有发现任何问题。您确定两个实例都使用同一个 Redis 服务器吗？

如果您运行redis-cli，然后运行monitor，您应该会看到如下内容：

1459258131.934949 [0 127.0.0.1:55237] "HSETNX" "MetaData" "bar.txt" "1384837200000"
1459258131.935129 [0 127.0.0.1:55237] "HSETNX" "MetaData" "baz.txt" "1384837200000"
1459258131.940125 [0 127.0.0.1:55237] "HSETNX" "MetaData" "/tmp/test/bar.txt" "1459258131000"
1459258131.940353 [0 127.0.0.1:55237] "HSETNX" "MetaData" "/tmp/test/baz.txt" "1459258131000"

在这种情况下，远程目录有 2 个文件；前两行来自远程过滤器，后两行来自本地过滤器（设置初始值）。

然后你应该会看到一堆

1459258142.073316 [0 127.0.0.1:55237] "HSETNX" "MetaData" "bar.txt" "1384837200000"
1459258142.073506 [0 127.0.0.1:55237] "HGET" "MetaData" "bar.txt"

（每次轮询一次 - 检查时间戳是否更改）。