在 Spring Integration 中成功持久化到 MongoDB 后删除文件

Question

我有一个 Spring Integration 流程，它从目录中读取 csv 文件，拆分行，然后处理每一行并从每一行中提取 2 个对象。然后将这两个对象发送到两个单独的int-mongodb:outbound-channel-adapter。我想在处理并保留所有行之后删除传入的文件。我已经看到使用事务管理器对入站适配器执行此操作的示例，但对出站适配器则没有。有没有办法做到这一点？

我的配置如下所示：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       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/task http://www.springframework.org/schema/task/spring-task.xsd
                http://www.springframework.org/schema/integration/mongodb http://www.springframework.org/schema/integration/mongodb/spring-integration-mongodb.xsd
                http://www.springframework.org/schema/data/mongo http://www.springframework.org/schema/data/mongo/spring-mongo.xsd"
       xmlns:int="http://www.springframework.org/schema/integration"
       xmlns:int-file="http://www.springframework.org/schema/integration/file"
       xmlns:task="http://www.springframework.org/schema/task"
       xmlns:int-mongodb="http://www.springframework.org/schema/integration/mongodb"
       xmlns:mongo="http://www.springframework.org/schema/data/mongo">

    <int:poller default="true" fixed-delay="50"/>

    <int-file:inbound-channel-adapter id="filesInChannel"
                                      directory="file:${file.ingest.directory}"
                                      auto-create-directory="true">
        <int:poller id="poller" fixed-rate="100">
        </int:poller>
    </int-file:inbound-channel-adapter>

    <task:executor id="executor" pool-size="10" queue-capacity="50" />
    <int:channel id="executorChannel">
        <int:queue capacity="50"/>
    </int:channel>

    <int:splitter input-channel="filesInChannel" output-channel="executorChannel"
                  expression="T(org.apache.commons.io.FileUtils).lineIterator(payload)"/>

    <int:service-activator id="lineParserActivator"  ref="lineParser" method="parseLine"
                           input-channel="executorChannel" output-channel="lineChannel">
        <int:poller task-executor="executor" fixed-delay="500">
        </int:poller>
    </int:service-activator>

    <bean name="lineParser" class="com.xxx.LineParser"/>

    <int:channel id="lineChannel">
        <int:queue/>
    </int:channel>
    <int:channel id="lineMongoOutput">
        <int:queue/>
    </int:channel>
    <int:channel id="actionMongoOutput">
        <int:queue/>
    </int:channel>

    <int:transformer input-channel="lineChannel" output-channel="lineMongoOutput">
        <bean id="lineTransformer" class="com.xxx.transformer.LineTransformer"></bean>  
    </int:transformer>
    <int:transformer input-channel="lineChannel" output-channel="actionMongoOutput">
        <bean id="actionTransformer" class="com.xxx.transformer.ActionTransformer"></bean>  
    </int:transformer>

    <mongo:db-factory id="mongoDbFactory" dbname="${mongo.db.name}" password="${mongo.db.pass}" username="${mongo.db.user}" port="${mongo.db.port}" host="${mongo.db.host}"/>

    <int-mongodb:outbound-channel-adapter id="lineMongoOutput"
                                          collection-name="full"
                                          mongodb-factory="mongoDbFactory" />
    <int-mongodb:outbound-channel-adapter id="actionMongoOutput"
                                          collection-name="action"
                                          mongodb-factory="mongoDbFactory" />
</beans>

Answer 1

您无法真正在出站适配器上执行此操作，因为您不知道何时“完成”。鉴于您正在异步切换到下游流（通过执行程序和队列通道），您也不能在入站适配器上执行此操作，因为一旦发送了所有拆分，轮询器线程就会返回到适配器。

除此之外，我在您的流程中发现了一些问题：

您似乎有过多的线程切换 - 您确实不需要下游流中的队列通道，因为您的执行由 exec 控制。频道。

让每个频道都成为 QueueChannel 是很不寻常的。

最后，您有 2 个变形金刚订阅了同一个频道。

您是否意识到发送到lineChannel 的消息会交替循环方式。

根据您的描述，也许这是您的意图，但对我来说似乎有点脆弱；我希望看到不同的数据类型进入不同的渠道。

如果您避免使用队列通道，并使用服务激活器中的网关将数据发送到 mongo 适配器，您的服务激活器将知道它何时完成并能够在那时删除文件。

编辑：

这是一种解决方案（它写入日志而不是 mongo，但您应该明白这一点）...

<int-file:inbound-channel-adapter directory="/tmp/foo" channel="toSplitter">
    <int:poller fixed-delay="1000">
        <int:transactional synchronization-factory="sf" transaction-manager="ptxMgr" />
    </int:poller>
</int-file:inbound-channel-adapter>

<int:transaction-synchronization-factory id="sf">
    <int:after-commit expression="payload.delete()" />
    <int:after-rollback expression="payload.renameTo(new java.io.File('/tmp/bad/' + payload.name))" />
</int:transaction-synchronization-factory>

<bean id="ptxMgr" class="org.springframework.integration.transaction.PseudoTransactionManager" />

<int:splitter input-channel="toSplitter" output-channel="processChannel">
    <bean class="org.springframework.integration.file.splitter.FileSplitter" />
</int:splitter>

<int:service-activator input-channel="processChannel">
    <bean class="foo.Foo">
        <constructor-arg ref="gate" />
    </bean>
</int:service-activator>

<int:gateway id="gate" service-interface="foo.Foo$Gate">
    <int:method name="toLine" request-channel="toLine" />
    <int:method name="toAction" request-channel="toAction" />
</int:gateway>

<int:channel id="toLine" />

<int:logging-channel-adapter channel="toLine" expression="'LINE:' + payload" level="WARN"/>

<int:channel id="toAction" />

<int:logging-channel-adapter channel="toAction" expression="'ACTION:' + payload" level="WARN"/>

.

public class Foo {

    private final Gate gateway;

    public Foo(Gate gateway) {
        this.gateway = gateway;
    }

    public void parse(String payload) {
        String[] split = payload.split(",");
        if (split.length != 2) {
            throw new RuntimeException("Bad row size: " + split.length);
        }
        this.gateway.toLine(split[0]);
        this.gateway.toAction(split[1]);
    }

    public interface Gate {

        void toLine(String line);

        void toAction(String action);
    }

}

.

@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class FooTests {

    @Test
    public void testGood() throws Exception {
        File file = new File("/tmp/foo/x.txt");
        FileOutputStream fos = new FileOutputStream(file);
        fos.write("foo,bar".getBytes());
        fos.close();
        int n = 0;
        while(n++ < 100 && file.exists()) {
            Thread.sleep(100);
        }
        assertFalse(file.exists());
    }

    @Test
    public void testBad() throws Exception {
        File file = new File("/tmp/foo/y.txt");
        FileOutputStream fos = new FileOutputStream(file);
        fos.write("foo".getBytes());
        fos.close();
        int n = 0;
        while(n++ < 100 && file.exists()) {
            Thread.sleep(100);
        }
        assertFalse(file.exists());
        file = new File("/tmp/bad/y.txt");
        assertTrue(file.exists());
        file.delete();
    }

}

在<poller/> 中添加一个任务执行器来同时处理多个文件。根据需要添加路由器。