弹簧集成；在复制完成之前复制服务拾取的文件

Question

应用背景

应用中有 2 个服务。让我们打电话给他们
1) 服务 A
2）服务B

服务 A 基本上将符合特定条件的文件从源目录复制到目标目录。在这里使用弹簧集成。 服务 A 的目标目录是服务 B 的源目录。

服务 B 不断地轮询目录中的文件并处理它们，然后将它们移动到另一个名为“已处理”的子目录。

问题：
最初的问题是，当服务 A 将文件复制到目标目录时，服务 B 会拾取一半复制的文件并对其进行处理。

尝试过的解决方案
请参阅下面的服务 Bintegration-context.xml。我将复合过滤器附加到入站通道。我向这个复合过滤器添加了一个名为LastCreatedFileListFilter 的自定义过滤器。此过滤器基于spring-integration-file 提供的LastModifiedFileListFilter 行，这基本上会丢弃任何年龄（按创建时间）小于30 秒的文件。

此过滤器运行良好，不会选择 30 秒前的文件。但现在的问题是我使用的是prevent-duplicates="true"。所以会发生什么是 service B 第一次轮询文件夹并且文件的年龄小于 30 秒它会过滤掉文件但 30 秒后，过滤器不会过滤掉文件正确的行为，但现在服务已将其标记为重复并拒绝它。

所以，我的问题是我确实想保留防止重复检查，并且在文件完全复制之前不处理文件。

我对弹簧集成不是很熟悉，我希望实现的两种方法是：
1）在应用程序处理之前，在上述场景中不将文件标记为重复？这可能吗？是这样，这是建议吗？
2) 在服务 A 中，如果我可以先创建具有临时名称的文件，然后在复制完成后重命名它们？这里的事情是服务 B 不会拿起文件，直到它们以配置的名称开始。通过这种方式，我不必信任不能 100% 决定性的年龄属性，因为源和目标位于不同的网络上，并且复制时间可能很长（#MurphysLaw）。 但这里的问题是，我在概念化如何通过 spring 集成最好地实现这个解决方案时遇到了麻烦。有什么指导或建议吗？

请参阅服务 A 的集成上下文以了解当前实现。让我知道是否有任何事情需要澄清。

服务 A

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-file="http://www.springframework.org/schema/integration/file"
       xmlns:context="http://www.springframework.org/schema/context"
       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/file
    http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

    <context:property-placeholder location="classpath:application.properties"/>


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

    <bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
        <property name="port" value="6379" />
    </bean>

    <int-file:inbound-channel-adapter id = "filesIn"
                                      channel="fileChannel"
                                      directory="file:${input.directory}"
                                      filter="incomingCompositeFilter">

        <int:poller id="fileInboudPoller" fixed-rate="${in.interval}" time-unit="SECONDS"  />
    </int-file:inbound-channel-adapter>

    <bean id="incomingCompositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
        <constructor-arg>
            <list>
                <bean id="acceptOnceFilter" class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter">
                    <constructor-arg ref="redisMetaDataStore"/>
                    <constructor-arg value="*"/>
                </bean>
                <bean id="notOlderThanDateFilter" class="com.fexco.bgeadmin.file.filter.NotOlderThanDateFilter">
                    <constructor-arg value="${file.lastModified.ignoreBeforeDate}"/>
                </bean>
                <bean id="documentConfigFilter" class="com.fexco.bgeadmin.file.filter.DocumentConfigFilter">
                </bean>
            </list>
        </constructor-arg></bean>

    <int:channel id="fileChannel"/>

    <int-file:outbound-channel-adapter id="save-as-file"
                                       auto-create-directory="true"
                                       channel="fileChannel"
                                       directory="file:${output.directory}"/>
</beans>

服务 B

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-file="http://www.springframework.org/schema/integration/file"
       xmlns:batch-int="http://www.springframework.org/schema/batch-integration"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       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/file
         http://www.springframework.org/schema/integration/file/spring-integration-file.xsd
         http://www.springframework.org/schema/batch-integration
         http://www.springframework.org/schema/batch-integration/spring-batch-integration.xsd">

    <int:channel id="inboundFileChannel"/>
    <int:channel id="outboundJobRequestChannel"/>
    <int:channel id="jobLaunchReplyChannel"/>

    <int-file:inbound-channel-adapter id="filePoller"
                                      channel="inboundFileChannel"
                                      directory="${app.file.source}"
                                      auto-create-directory="true"
                                      prevent-duplicates="true"
                                      filter="incomingCompositeFilter">
        <int:poller fixed-rate="5000"/>
    </int-file:inbound-channel-adapter>

    <bean id="incomingCompositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
        <constructor-arg>
            <list>
                <bean id="fileNameFilter" class="org.springframework.integration.file.filters.RegexPatternFileListFilter">
                    <constructor-arg value=".*\.(xls|xlsx|csv)$" />
                </bean>
                <bean id="ageFilter" class="com.fexco.bgeadmin.integration.filter.LastCreatedFileListFilter">
                    <property name="age" value="30"/>
                </bean>
            </list>
        </constructor-arg></bean>

    <int:transformer input-channel="inboundFileChannel"
                     output-channel="outboundJobRequestChannel" method="toRequest">
        <bean class="com.fexco.bgeadmin.integration.FileMessageToJobRequest"/>
    </int:transformer>

    <batch-int:job-launching-gateway request-channel="outboundJobRequestChannel"
                                     reply-channel="jobLaunchReplyChannel"/>

    <int:logging-channel-adapter channel="jobLaunchReplyChannel"/>

</beans>

Answer 1

此类问题的最佳解决方案是＃2 - A 不要“就地”将文件写入 B。使用上次修改时间是不可靠的。

这是<int:file-outbound-channel-adapter/> 的标准过程 - 它在下面使用FileWritingMessageHandler。它有一个属性temporaryFileSuffix，默认为.writing。文件复制后重命名。