spring - 在Spring Integration中成功持久保存到MongoDB后删除文件

spring - 在Spring Integration中成功持久保存到MongoDB后删除文件,第1张

我有一个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>

最佳答案:

1 个答案:

答案 0 :(得分: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/>添加任务执行程序以同时处理多个文件。根据需要添加路由器。

DABAN RP主题是一个优秀的主题,极致后台体验,无插件,集成会员系统
U19学习网站 » spring - 在Spring Integration中成功持久保存到MongoDB后删除文件