带有zookeeper服务器的集群环境中的FTP入站适配器

Question

我有一个FTP入站适配器，它将部署在tomcat的多个实例上。问题是，所有的实例将从远程目录拉文件，但我需要一个单一的文件，只被拉一次，而不是所有的实例。所以我使用Zookeeper元数据存储，但它在启动时给我错误。

误差

java.lang.IllegalStateException: Error processing condition on org.springframework.cloud.zookeeper.ZookeeperAutoConfiguration.zookeeperProperties
    at org.springframework.boot.autoconfigure.condition.SpringBootCondition.matches(SpringBootCondition.java:64) ~[spring-boot-autoconfigure-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.context.annotation.ConditionEvaluator.shouldSkip(ConditionEvaluator.java:102) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForBeanMethod(ConfigurationClassBeanDefinitionReader.java:178) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:140) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:116) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:333) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:243) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:273) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:98) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:678) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:520) ~[spring-context-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.boot.context.embedded.EmbeddedWebApplicationContext.refresh(EmbeddedWebApplicationContext.java:118) ~[spring-boot-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:766) [spring-boot-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.createAndRefreshContext(SpringApplication.java:361) [spring-boot-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:307) [spring-boot-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1191) [spring-boot-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1180) [spring-boot-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at com.deere.sample.main.Application.main(Application.java:21) [classes/:na]
Caused by: java.lang.IllegalArgumentException: @ConditionalOnMissingBean annotations must specify at least one bean (type, name or annotation)
    at org.springframework.util.Assert.isTrue(Assert.java:68) ~[spring-core-4.2.6.RELEASE.jar:4.2.6.RELEASE]
    at org.springframework.boot.autoconfigure.condition.OnBeanCondition$BeanSearchSpec.validate(OnBeanCondition.java:279) ~[spring-boot-autoconfigure-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.autoconfigure.condition.OnBeanCondition$BeanSearchSpec.<init>(OnBeanCondition.java:275) ~[spring-boot-autoconfigure-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.autoconfigure.condition.OnBeanCondition.getMatchOutcome(OnBeanCondition.java:111) ~[spring-boot-autoconfigure-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    at org.springframework.boot.autoconfigure.condition.SpringBootCondition.matches(SpringBootCondition.java:47) ~[spring-boot-autoconfigure-1.3.5.RELEASE.jar:1.3.5.RELEASE]
    ... 17 common frames omitted

下面是配置。

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
    xmlns:jms="http://www.springframework.org/schema/integration/jms"
    xmlns:stream="http://www.springframework.org/schema/integration/stream"
    xmlns:integration="http://www.springframework.org/schema/integration"
    xmlns:int-xml="http://www.springframework.org/schema/integration/xml"
    xmlns:int-ftp="http://www.springframework.org/schema/integration/ftp"
    xmlns:int-file="http://www.springframework.org/schema/integration/file"
    xmlns:int="http://www.springframework.org/schema/integration"
    xmlns:mail="http://www.springframework.org/schema/integration/mail"
    xmlns:jdbc="http://www.springframework.org/schema/integration/jdbc"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/integration
            http://www.springframework.org/schema/integration/spring-integration.xsd
            http://www.springframework.org/schema/integration/jms
            http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
            http://www.springframework.org/schema/integration/stream
            http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd
            http://www.springframework.org/schema/integration/xml
            http://www.springframework.org/schema/integration/xml/spring-integration-xml.xsd
            http://www.springframework.org/schema/integration/mail
            http://www.springframework.org/schema/integration/mail/spring-integration-mail-4.2.xsd 
            http://www.springframework.org/schema/integration/jdbc
            http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
            http://www.springframework.org/schema/integration/ftp
            http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd">

    <beans:bean id="ftpClientFactory"
        class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
        <beans:property name="host" value="*******"/>
        <beans:property name="username" value="*********"/>
        <beans:property name="password" value="******"/>
        <beans:property name="clientMode" value="0"/>
        <beans:property name="fileType" value="2"/>
        <beans:property name="bufferSize" value="100000"/>
    </beans:bean>

    <integration:logging-channel-adapter
        id="nonDeerelogger" level="DEBUG" />
        <integration:wire-tap id="nonDeereWireTap"
        channel="nonDeerelogger" pattern="*" order="2" />
         <integration:logging-channel-adapter
        id="logger" log-full-message="true" level="INFO" />
        <integration:channel id="receiveChannel" />
  <int-ftp:inbound-channel-adapter id="ftpInbound" 
        channel="receiveChannel" session-factory="ftpClientFactory"
        auto-create-local-directory="true" delete-remote-files="false"
        remote-directory="/OUT/SDI402_CARATT_JD" remote-file-separator="/"
        filter="compositeFilter"
        local-directory="src/main/resources/agritalialocal/"
        local-filter="acceptAll">
        <int:poller fixed-rate="5000"  max-messages-per-poll="1" />
    </int-ftp:inbound-channel-adapter>
     <integration:service-activator
        id="nonDeereXmlServiceActivator" input-channel="receiveChannel"
        ref="NonDeereFileListener" method="listen" />

    <beans:bean id="acceptAll" class="org.springframework.integration.file.filters.AcceptAllFileListFilter" />
    <beans:bean id="compositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
        <beans:constructor-arg>
            <beans:list>
                <beans:bean class="org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter">
                    <beans:constructor-arg value="*.xml" />
                </beans:bean>
                <beans:bean class="org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter">
                    <beans:constructor-arg name="store" ref="metadataStore"/>
                    <beans:constructor-arg value="foo/bar/"/>
                </beans:bean>
            </beans:list>
        </beans:constructor-arg>
    </beans:bean>

    <beans:bean id="client" class="org.springframework.integration.zookeeper.config.CuratorFrameworkFactoryBean">
    <beans:constructor-arg value="0.0.0.0:2181" />
</beans:bean>
<beans:bean id="meta" class="org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore">
    <beans:constructor-arg ref="client" />
</beans:bean>
   <!--  <integration:channel id="testChannel" />
    <integration:inbound-channel-adapter
        channel="testChannel"  ref="NonDeereFileListener" method="listen"
        auto-startup="true">
        <integration:poller fixed-delay="1000">
        </integration:poller>
    </integration:inbound-channel-adapter> -->
     <!--  <beans:bean id="NonDeereFileListener"
        class="com.deere.sample.controller.NonDeereFileListener" /> -->

</beans:beans>

POM zookeeper服务器的条目。org.springframework.cloudspring-cloud-zookeeper-config1.0.0.RELEASE如果删除上面的POM条目，然后我得到java.lang.ClassNotFoundException: org.springframework.integration.zookeeper.config.CuratorFramework FactoryBean将POM条目更改为spring-integration-zookeeper和服务器启动后工作正常，但在MetaDataStore中输入值时给出异常

2016-09-06 12:53:38.072 ERROR 11456 --- [ask-scheduler-2] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Problem occurred while synchronizing remote to local directory; nested exception is org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStoreException: Error while trying to set 'foo/bar/20160626S001_20160626.xml':
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:266)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource.doReceive(AbstractInboundFileSynchronizingMessageSource.java:193)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource.doReceive(AbstractInboundFileSynchronizingMessageSource.java:59)
    at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:134)
    at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:175)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:224)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:57)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:176)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:173)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:330)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:51)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:324)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
    at java.util.concurrent.FutureTask.run(Unknown Source)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(Unknown Source)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
Caused by: org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStoreException: Error while trying to set 'foo/bar/20160626S001_20160626.xml':
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:419)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.synchronizeToLocalDirectory(AbstractInboundFileSynchronizer.java:223)
    ... 22 more
Caused by: org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStoreException: Error while trying to set 'foo/bar/20160626S001_20160626.xml':
    at org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore.putIfAbsent(ZookeeperMetadataStore.java:132)
    at org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter.accept(AbstractPersistentAcceptOnceFileListFilter.java:77)
    at org.springframework.integration.file.filters.AbstractFileListFilter.filterFiles(AbstractFileListFilter.java:40)
    at org.springframework.integration.file.filters.CompositeFileListFilter.filterFiles(CompositeFileListFilter.java:108)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.filterFiles(AbstractInboundFileSynchronizer.java:206)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer$1.doInSession(AbstractInboundFileSynchronizer.java:230)
    at org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer$1.doInSession(AbstractInboundFileSynchronizer.java:223)
    at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:410)
    ... 23 more
Caused by: org.apache.zookeeper.KeeperException$NoNodeException: KeeperErrorCode = NoNode for /SpringIntegration-MetadataStore/foo/bar/20160626S001_20160626.xml
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:111)
    at org.apache.zookeeper.KeeperException.create(KeeperException.java:51)
    at org.apache.zookeeper.ZooKeeper.create(ZooKeeper.java:783)
    at org.apache.curator.framework.imps.CreateBuilderImpl$11.call(CreateBuilderImpl.java:696)
    at org.apache.curator.framework.imps.CreateBuilderImpl$11.call(CreateBuilderImpl.java:679)
    at org.apache.curator.RetryLoop.callWithRetry(RetryLoop.java:107)
    at org.apache.curator.framework.imps.CreateBuilderImpl.pathInForeground(CreateBuilderImpl.java:676)
    at org.apache.curator.framework.imps.CreateBuilderImpl.protectedPathInForeground(CreateBuilderImpl.java:453)
    at org.apache.curator.framework.imps.CreateBuilderImpl.forPath(CreateBuilderImpl.java:443)
    at org.apache.curator.framework.imps.CreateBuilderImpl.forPath(CreateBuilderImpl.java:44)
    at org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore.createNode(ZookeeperMetadataStore.java:256)
    at org.springframework.integration.zookeeper.metadata.ZookeeperMetadataStore.putIfAbsent(ZookeeperMetadataStore.java:118)
    ... 30 more

Answer 1

我的问题现已解决。在将FtpPersistentAcceptOnceFileListFilter的构造函数参数从foo/bar/更改为空字符串后，它得到了解决。

<beans:bean class="org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter">
                    <beans:constructor-arg name="store" ref="metadataStore"/>
                    <beans:constructor-arg value=""/>
</beans:bean>