java - spring 集成 - 从 FTP 读取文件并处理它。如何实现metastore并在不轮询的情况下进行处理

Question

在 Spring 集成中 - 我尝试从 FTP 读取文件并处理它。我们如何实现metastore并在不轮询的情况下对其进行处理。 在下面的配置中，为了避免读取相同的文件，如果服务器重新启动，我在ftpChannel中引入了消息存储。 现在，文件的处理器是服务激活器，需要轮询。如何避免在服务激活器中进行轮询并立即从 ftpChannel 队列中读取文件。如果我使用 int:dispatcher 那么我就无法使用消息存储。

我们如何解决这个问题？

<int:channel id="ftpChannel">
    <int:queue message-store="mongoDbMessageStore" />
    <!-- <int:dispatcher task-executor="taskExecutor"/> -->
</int:channel>

<bean id="mongoDbMessageStore"
    class="org.springframework.integration.mongodb.store.MongoDbMessageStore">
    <constructor-arg name="mongoDbFactory" ref="mongoDbFactory" />
    <constructor-arg name="collectionName" value="ftpInfo" />
</bean>

<int-ftp:inbound-channel-adapter id="ftpInbound"
    channel="ftpChannel" session-factory="ftpClientFactory" charset="UTF-8"
    auto-create-local-directory="true" delete-remote-files="false"
    filename-pattern="*.gz" remote-directory="/myfilerepo/#{istDate.getISTDate()}"
    remote-file-separator="/" local-filename-generator-expression="#this.toUpperCase()"
    temporary-file-suffix=".writing" preserve-timestamp="true"
    local-directory="/temp/spring/#{istDate.getISTDate()}">

     <int:poller cron="0-5 0/5 * * * ?" max-messages-per-poll="-1"/>
</int-ftp:inbound-channel-adapter>

<int:service-activator id="jobServiceActivator"
    input-channel="ftpChannel" ref="triggerJobLauncher" method="launch">
    <int:poller fixed-delay="10" />
</int:service-activator>

<!-- job context -->
<bean id="jobRepository"
    class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
    <property name="transactionManager" ref="transactionManager" />
</bean>

<bean id="transactionManager"
    class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />

<bean id="jobLauncher"
    class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository" />
</bean>
<!-- job context -->

Answer 1

您不需要 channel 上的消息存储；您需要使用 FtpPersistentAcceptOnceFileListFilter在过滤器和/或FileSystemPersistentAcceptOnceFileListFilter中在 local-filter 中以避免系统重新启动后重新处理文件。

他们需要一个MetadataStore； 如果你想使用 mongo，你需要实现一个；该框架目前没有 mongo 实现。

编辑:

从版本 4.2 开始，框架 now has a mongo MetadataStore .