java - Spring 集成；在复制完成之前复制服务拾取的文件

Question

应用背景

有 2 个服务正在申请中。让我们称呼他们吧
1)服务A
2)服务B

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

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

问题:
最初的问题是，当服务 A 将文件复制到目标目录时，服务 B 拾取复制的一半文件并处理它们。

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

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

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

我对 spring-integration 不太熟悉，我希望实现的两种方法是:
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默认情况下。文件复制后会被重命名。