Spring Integration Java DSL流Splitter/Aggregator在处理所有行后删除文件

Question

使用 Spring Integration Java DSL，我构建了一个流程，在该流程中我与 FileSplitter 同步处理文件。在将File 中的每一行转换为Message 以进行后续处理后，我已经能够使用AbstractFilePayloadTransformer 上的setDeleteFiles 标志来删除文件，如下所示：

@Bean
protected IntegrationFlow s3ChannelFlow() {
    // do not exhaust filesystem w/ files downloaded from S3
    FileToInputStreamTransformer transformer = new FileToInputStreamTransformer();
    transformer.setDeleteFiles(true);

    // @see http://docs.spring.io/spring-integration/reference/html/files.html#file-reading
    // @formatter:off
    return IntegrationFlows
        .from(s3Channel())
        .channel(StatsUtil.createRunStatsChannel(runStatsRepository))
        .transform(transformer)
        .split(new FileSplitter())
        .transform(new JsonToObjectViaTypeHeaderTransformer(new Jackson2JsonObjectMapper(objectMapper), typeSupport))
        .publishSubscribeChannel(p -> p.subscribe(persistenceSubFlow()))
        .get();
    // @formatter:on
}

这很好用，但速度很慢。所以我尝试在上面的.split 之后添加一个ExecutorChannel，如下所示：

.channel(c -> c.executor(Executors.newFixedThreadPool(10)))

但是前面提到的删除标志不允许流程在完全读取文件之前完成成功删除文件。

如果我删除该标志，我可能会耗尽从 S3 同步文件的本地文件系统。

我可以在上面介绍什么 a) 完全处理每个文件 b) 完成后从本地文件系统中删除文件？ 换句话说，有没有办法准确知道文件何时完全处理（当它的行已通过池中的线程异步处理时）？

如果你好奇，这里是我对 FileToInputStreamTransformer 的暗示：

public class FileToInputStreamTransformer extends AbstractFilePayloadTransformer<InputStream> {

    private static final int BUFFER_SIZE = 64 * 1024; // 64 kB

    @Override
    // @see http://java-performance.info/java-io-bufferedinputstream-and-java-util-zip-gzipinputstream/
    protected InputStream transformFile(File payload) throws Exception {
        return new GZIPInputStream(new FileInputStream(payload), BUFFER_SIZE);
    }
}

更新

那么，下游流中的某些东西是如何知道要请求什么的呢？

顺便说一句，如果我正确地遵循了你的建议，当我用上面的new FileSplitter(true, true) 更新.split 时，我得到了

2015-10-20 14:26:45,288 [pool-6-thread-1] org.springframework.integration.handler.LoggingHandler 错误 org.springframework.integration.transformer.MessageTransformationException：转换消息失败；嵌套异常是 java.lang.IllegalArgumentException：'json' 参数必须是以下实例：[class java.lang.String，class [B，class java.io.File，class java.net.URL，class java.io。 InputStream, class java.io.Reader] ，但得到：class org.springframework.integration.file.splitter.FileSplitter$FileMarker 在 org.springframework.integration.transformer.AbstractTransformer.transform(AbstractTransformer.java:44)

Answer 1

FileSplitter 具有 markers option 正是为此目的：

设置为 true 以在文件数据之前和之后发出文件标记消息的开始/结束。标记是具有FileSplitter.FileMarker 有效负载的消息（在标记属性中具有START 和END 值）。在过滤某些行的下游流中按顺序处理文件时，可能会使用标记。它们使下游处理能够知道文件何时已被完全处理。 END 标记包括行数。默认值：false。当true时，apply-sequence默认为false。

您可以在下游流程中使用它来确定是否可以删除文件。

Answer 2

谢谢阿尔乔姆。

我确实设法解决了这个问题，但也许是以更重要的方式。

引入ExecutorChannel 引起了相当多的实施调整，包括：关闭AbtractFilePayloadTransformer 上的setDeleteFiles 标志，更新JPA @Entity、RunStats 和存储库，以捕获文件处理状态以及整个运行的处理状态。综合处理状态更新，让流程知道何时从本地文件系统中删除文件（即，它们何时被完全处理）并在/stats/{run} 端点中返回状态，以便用户知道何时完成运行。

这是我的实现中的 sn-ps（如果有人好奇的话）...

class FileToInputStreamTransformer extends AbstractFilePayloadTransformer<InputStream> {

private static final int BUFFER_SIZE = 64 * 1024; // 64 kB

@Override
// @see http://java-performance.info/java-io-bufferedinputstream-and-java-util-zip-gzipinputstream/
protected InputStream transformFile(File payload) throws Exception {
    return new GZIPInputStream(new FileInputStream(payload), BUFFER_SIZE);
}
}

public class RunStatsHandler extends AbstractMessageHandler {

private final SplunkSlf4jLogger log = new SplunkSlf4jLogger(LoggerFactory.getLogger(getClass()));
private static final int BUFFER_SIZE = 64 * 1024; // 64 kB

private final RunStatsRepository runStatsRepository;

public RunStatsHandler(RunStatsRepository runStatsRepository) {
    this.runStatsRepository = runStatsRepository;
}

// Memory efficient routine, @see http://www.baeldung.com/java-read-lines-large-file
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
    RunStats runStats = message.getHeaders().get(RunStats.RUN, RunStats.class);
    String token = message.getHeaders().get(RunStats.FILE_TOKEN, String.class);
    if (runStats != null) {
        File compressedFile = (File) message.getPayload();
        String compressedFileName = compressedFile.getCanonicalPath();
        LongAdder lineCount = new LongAdder();
        // Streams and Scanner implement java.lang.AutoCloseable
        InputStream fs = new FileInputStream(compressedFile);
        InputStream gzfs = new GZIPInputStream(fs, BUFFER_SIZE);
        try (Scanner sc = new Scanner(gzfs, "UTF-8")) {
            while (sc.hasNextLine()) {
                sc.nextLine();
                lineCount.increment();
            }
            // note that Scanner suppresses exceptions
            if (sc.ioException() != null) {
                log.warn("file.lineCount", ImmutableMap.of("run", runStats.getRun(), "file", compressedFileName, 
                        "exception", sc.ioException().getMessage()));
                throw sc.ioException();
            }
            runStats.addFile(compressedFileName, token, lineCount.longValue());
            runStatsRepository.updateRunStats(runStats);
            log.info("file.lineCount",
                    ImmutableMap.of("run", runStats.getRun(), "file", compressedFileName, "lineCount", lineCount.intValue()));
        }
    }
}

}

更新流程

@Bean
protected IntegrationFlow s3ChannelFlow() {
    // @see http://docs.spring.io/spring-integration/reference/html/files.html#file-reading
    // @formatter:off
    return IntegrationFlows
        .from(s3Channel())
        .enrichHeaders(h -> h.headerFunction(RunStats.FILE_TOKEN, f -> UUID.randomUUID().toString()))
        .channel(runStatsChannel())
        .channel(c -> c.executor(Executors.newFixedThreadPool(persistencePoolSize)))
        .transform(new FileToInputStreamTransformer())
        .split(new FileSplitter())
        .transform(new JsonToObjectViaTypeHeaderTransformer(new Jackson2JsonObjectMapper(objectMapper), typeSupport))
        .publishSubscribeChannel(p -> p.subscribe(persistenceSubFlow()))
        .get();
    // @formatter:on
}

@Bean
public MessageChannel runStatsChannel() {
    DirectChannel wiretapChannel = new DirectChannel();
    wiretapChannel.subscribe(new RunStatsHandler(runStatsRepository));
    DirectChannel loggingChannel = new DirectChannel();
    loggingChannel.addInterceptor(new WireTap(wiretapChannel));
    return loggingChannel;
}

很遗憾，我无法分享 RunStats 和 repo 实现。