處理多條消息

我有一個spring sd源模塊，它從s3中提取文件並逐行分割。我的spring配置如下。但是我有3個容器和1個管理服務器。現在，我看到正在處理的重複消息每個容器都有自己的副本。我可以解決與源s3模塊部署計數爲1，但我的郵件處理變慢。？解決此問題的任何投入？處理多條消息

<int:poller fixed-delay="${fixedDelay}" default="true"> 
    <int:advice-chain> 
      <ref bean="pollAdvise"/> 

      </int:advice-chain> 
    </int:poller> 


    <bean id="pollAdvise" 

    </bean> 






    <bean id="credentials" class="org.springframework.integration.aws.core.BasicAWSCredentials"> 
     <property name="accessKey" value="#{encryptedDatum.decryptBase64Encoded('${accessKey}')}"/> 
     <property name="secretKey" value="${secretKey}"/> 
    </bean> 


    <bean id="clientConfiguration" class="com.amazonaws.ClientConfiguration"> 
     <property name="proxyHost" value="${proxyHost}"/> 
     <property name="proxyPort" value="${proxyPort}"/> 
    <property name="preemptiveBasicProxyAuth" value="false"/> 
    </bean> 

    <bean id="s3Operations" class="org.springframework.integration.aws.s3.core.CustomC1AmazonS3Operations"> 
     <constructor-arg index="0" ref="credentials"/> 
     <constructor-arg index="1" ref="clientConfiguration"/> 
     <property name="awsEndpoint" value="s3.amazonaws.com"/> 
     <property name="temporaryDirectory" value="${temporaryDirectory}"/> 
     <property name="awsSecurityKey" value="${awsSecurityKey}"/> 
    </bean> 

    <bean id="encryptedDatum" class="abc"/> 

    <!-- aws-endpoint="https://s3.amazonaws.com" --> 
    <int-aws:s3-inbound-channel-adapter aws-endpoint="s3.amazonaws.com" 
             bucket="${bucket}" 
             s3-operations="s3Operations" 
             credentials-ref="credentials" 
             file-name-wildcard="${fileNameWildcard}" 
             remote-directory="${remoteDirectory}" 
             channel="splitChannel" 
             local-directory="${localDirectory}" 
             accept-sub-folders="false" 
             delete-source-files="true" 
             archive-bucket="${archiveBucket}" 
             archive-directory="${archiveDirectory}"> 
    </int-aws:s3-inbound-channel-adapter> 

    <int-file:splitter input-channel="splitChannel" output-channel="output" markers="false" charset="UTF-8"> 

     <int-file:request-handler-advice-chain> 
      <bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice"> 
       <property name="onSuccessExpression" value="payload.delete()"/> 
      </bean> 
     </int-file:request-handler-advice-chain> 

</int-file:splitter> 

    <int:channel id="output"/>

[更新] 我增加了冪等的建議由你與元數據store.But因爲我的XD在3容器集羣中運行與兔子會簡單metadatastore工作？我想我應該用紅色/蒙戈元source.If我使用mongo/redis metadatastore howcan我驅逐/刪除消息，因爲消息會隨着時間堆積起來？

<int:idempotent-receiver id="expressionInterceptor" endpoint="output" 
           metadata-store="store" 
          discard-channel="nullChannel" 
          throw-exception-on-rejection="false" 
           key-expression="payload"/> 

    <bean id="store" class="org.springframework.integration.metadata.SimpleMetadataStore"/>

來源

2016-01-20 constantlearner