Spring Integration：使用多实例重试配置答案

【问题标题】：Spring Integration : retry configuration with multi-instancesSpring Integration：使用多实例重试配置
【发布时间】：2023-04-06 10:16:01
【问题描述】：

我在 4 个不同的服务器上运行 4 个基于 Spring Boot 集成的应用程序实例。过程是：

在共享文件夹中一一读取 XML 文件。
处理文件（检查结构、内容...）、转换数据并发送电子邮件。
在另一个共享文件夹中写一份关于此文件的报告。
删除成功处理的文件。

我正在寻找一种非阻塞且安全的解决方案来处理这些文件。

用例：

如果实例在读取或处理文件时崩溃（因此不结束集成链）：另一个实例必须处理该文件，或者同一实例必须在重新启动后处理该文件。
如果一个实例正在处理文件，则其他实例不得处理该文件。

我已经构建了这个 Spring Integration XML 配置文件（它包括带有共享 H2 数据库的 JDBC 元数据存储）：

    <?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"
    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">

<int:poller default="true" fixed-rate="1000"/>
<int:channel id="inputFilesChannel">
    <int:queue/>
</int:channel>

<!-- Input -->
<int-file:inbound-channel-adapter 
    id="inputFilesAdapter"
    channel="inputFilesChannel"
    directory="file:${input.files.path}" 
    ignore-hidden="true"
    comparator="lastModifiedFileComparator"
    filter="compositeFilter">
   <int:poller fixed-rate="10000" max-messages-per-poll="1"  task-executor="taskExecutor"/>
</int-file:inbound-channel-adapter>

<task:executor id="taskExecutor" pool-size="1"/>

<!-- Metadatastore -->
<bean id="jdbcDataSource" class="org.apache.commons.dbcp.BasicDataSource">
    <property name="url" value="jdbc:h2:file:${database.path}/shared;AUTO_SERVER=TRUE;AUTO_RECONNECT=TRUE;MVCC=TRUE"/>
    <property name="driverClassName" value="org.h2.Driver"/>
    <property name="username" value="${database.username}"/>
    <property name="password" value="${database.password}"/>
    <property name="maxIdle" value="4"/>
</bean>

<bean id="jdbcMetadataStore" class="org.springframework.integration.jdbc.metadata.JdbcMetadataStore">
    <constructor-arg ref="jdbcDataSource"/>
</bean>

<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    <property name="dataSource" ref="jdbcDataSource"/>
</bean>

<bean id="compositeFilter" class="org.springframework.integration.file.filters.CompositeFileListFilter">
    <constructor-arg>
        <list>
            <bean class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter">
                <constructor-arg index="0" ref="jdbcMetadataStore"/>
                <constructor-arg index="1" value="files"/>
            </bean>
        </list>
    </constructor-arg>
</bean>

<!-- Workflow -->
<int:chain input-channel="inputFilesChannel" output-channel="outputFilesChannel">
    <int:service-activator ref="fileActivator" method="fileRead"/>
    <int:service-activator ref="fileActivator" method="fileProcess"/>
    <int:service-activator ref="fileActivator" method="fileAudit"/>
</int:chain>

<bean id="lastModifiedFileComparator" class="org.apache.commons.io.comparator.LastModifiedFileComparator"/>

<int-file:outbound-channel-adapter 
    id="outputFilesChannel" 
    directory="file:${output.files.path}"
    filename-generator-expression ="payload.name">
    <int-file:request-handler-advice-chain>
        <bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice">
             <property name="onSuccessExpressionString" value="headers[file_originalFile].delete()"/>
        </bean>
    </int-file:request-handler-advice-chain>
</int-file:outbound-channel-adapter>

</beans>

问题： 对于多个文件，当成功处理一个文件时，事务提交元数据存储中的其他现有文件（表INT_METADATA_STORE）。因此，如果重新启动应用程序，则永远不会处理其他文件（如果在处理第一个文件时应用程序崩溃，它可以正常工作）。似乎只适用于读取文件，不适用于处理集成链中的文件......如何逐个文件管理JVM崩溃文件的回滚事务？

非常感谢任何帮助。这会让我发疯:(

谢谢！

编辑/注释：

灵感来自https://github.com/caoimhindenais/spring-integration-files/blob/master/src/main/resources/context.xml
我已经用 Artem Bilan 的回答更新了我的配置。并删除 poller 块中的 transactional 块：我在实例之间发生了事务冲突（丑陋的表锁异常）。虽然行为是一样的。

我在 poller 块中测试此配置失败（行为相同）：

<int:advice-chain>
    <tx:advice id="txAdvice" transaction-manager="transactionManager">
        <tx:attributes>
            <tx:method name="file*" timeout="30000" propagation="REQUIRED"/>
        </tx:attributes>
    </tx:advice>
</int:advice-chain>

也许基于Idempotent Receiver Enterprise Integration Pattern 的解决方案可以工作。但我没有设法配置它……我没有找到准确的文档。

【问题讨论】：

标签： java spring spring-boot spring-integration

【解决方案1】：

您不应使用PseudoTransactionManager，而应使用DataSourceTransactionManager。

由于您使用JdbcMetadataStore，它将参与事务，如果下游流失败，元数据存储中的条目也将被回滚。

【讨论】：

很高兴知道，但问题仍然存在:(

【解决方案2】：

好的。我找到了一个可行的解决方案。也许不是最干净的，但它有效：

不同服务器上的多实例，共享同一个 H2 数据库（网络文件夹挂载）。我认为它应该通过远程 TCP 工作。 MVCC 已在 H2 上激活（查看其 doc）。
inbound-channel-adapter 已激活 scan-each-poll 选项以允许重新轮询以前可能被忽略的文件（如果该过程已由另一个实例开始）。因此，如果另一个实例崩溃，则可以再次轮询和处理该文件，而无需为此实例重新启动。
选项 defaultAutoCommit 在 DB 上设置为 false。
我没有使用FileSystemPersistentAcceptOnceFileListFilter，因为当一个文件被成功处理时，它会聚合元数据存储中的所有读取文件。我没有设法在我的上下文中使用它...

我通过过滤器和事务同步将自己的条件和动作写在表达式中。

<!-- Input -->
<bean id="lastModifiedFileComparator" class="org.apache.commons.io.comparator.LastModifiedFileComparator"/>
<int-file:inbound-channel-adapter
    id="inputAdapter"
    channel="inputChannel"
    directory="file:${input.files.path}"
    comparator="lastModifiedFileComparator"
    scan-each-poll="true">
   <int:poller max-messages-per-poll="1" fixed-rate="5000">
        <int:transactional transaction-manager="transactionManager"  isolation="READ_COMMITTED" propagation="REQUIRED" timeout="60000" synchronization-factory="syncFactory"/>
   </int:poller>
</int-file:inbound-channel-adapter>

<!-- Continue only if the concurrentmetadatastore doesn't contain the file. If if is not the case : insert it in the metadatastore -->
<int:filter input-channel="inputChannel" output-channel="processChannel"  discard-channel="nullChannel" throw-exception-on-rejection="false" expression="@jdbcMetadataStore.putIfAbsent(headers[file_name], headers[timestamp]) == null"/>

<!-- Rollback by removing the file from the metadatastore -->
<int:transaction-synchronization-factory id="syncFactory">
    <int:after-rollback expression="@jdbcMetadataStore.remove(headers[file_name])" />
</int:transaction-synchronization-factory>

<!-- Metadatastore configuration -->
<bean id="jdbcDataSource" class="org.apache.commons.dbcp.BasicDataSource">
    <property name="url" value="jdbc:h2:file:${database.path}/shared;AUTO_SERVER=TRUE;AUTO_RECONNECT=TRUE;MVCC=TRUE"/>
    <property name="driverClassName" value="org.h2.Driver"/>
    <property name="username" value="${database.username}"/>
    <property name="password" value="${database.password}"/>
    <property name="maxIdle" value="4"/>
    <property name="defaultAutoCommit" value="false"/>
</bean>

<bean id="jdbcMetadataStore" class="org.springframework.integration.jdbc.metadata.JdbcMetadataStore">
    <constructor-arg ref="jdbcDataSource"/>
</bean>

<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
    <property name="dataSource" ref="jdbcDataSource"/>
</bean>

<!-- Workflow -->
<int:chain input-channel="processChannel" output-channel="outputChannel">
    <int:service-activator ref="fileActivator" method="fileRead"/>
    <int:service-activator ref="fileActivator" method="fileProcess"/>
    <int:service-activator ref="fileActivator" method="fileAudit"/>
</int:chain>


<!-- Output -->
<int-file:outbound-channel-adapter 
    id="outputChannel" 
    directory="file:${output.files.path}"
    filename-generator-expression ="payload.name">
    <!-- Delete the source file -->
    <int-file:request-handler-advice-chain>
        <bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice">
             <property name="onSuccessExpressionString" value="headers[file_originalFile].delete()"/>
        </bean>
    </int-file:request-handler-advice-chain>
</int-file:outbound-channel-adapter>

欢迎任何改进或其他解决方案。

【讨论】：