分佈式彈簧整合ftp入站適配器

Question

我正在使用Spring Integration: FTP Inbound Channel Adapter從遠程FTP服務器讀取文件。我的問題是，它每天能處理大約5百萬個文件嗎？

如果我在分佈式模式下在2臺不同的服務器上部署項目戰，那麼會出現問題嗎？因爲在這兩臺服務器上都會運行FTP Inbound Channel Adapter。然後這兩個適配器將兩次讀取相同的文件。 請幫我設置分佈式系統。

編輯：
我已經建立了我的Spring Integration項目戰爭2臺服務器。它使用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> 

1服務器的輸出日誌：

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 

第2服務器生成相同的日誌，不同的消息ID 。 我錯過了什麼嗎？
我是否需要爲此編寫自定義過濾器？

Answer 1

我的問題是，它每天能處理大約5百萬個文件嗎？

這取決於文件大小和網絡帶寬; 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" 

（每一次民意調查 - 檢查，看看是否時間戳改變）。