用于多文件处理的Camel

Question

我是骆驼的新手。我将有一个文件处理与骆驼，但我还没有找到一个现成的解决方案，我的情况。我必须一起处理多个文件，以防它们存在。这些文件上载到特定文件夹时会有一些延迟（例如：我们有两个文件a.csv和b.csv，a.csv上载的时间比b.csv晚10秒，反之亦然）。此外，如果一个文件缺席超过特定的时间，我只需要处理一个文件。有谁能帮我选一个图案吗？据我所知，我可以使用camel过滤器来确保我们已经有了这两个文件a.csv和b.csv，然后才开始处理，但这并不能解决我的问题。

Answer 1

这是聚合器EIP。

from("file:inputFolder")
        .aggregate(constant(true), AggregationStrategies.groupedExchange())
        .completionSize(2) //Wait for two files
        .completionTimeout(60000) //Or process single file, if completionSize was not fulfilled within one minute
            .to("log:do_something") //Here you can access List<Exchange> from message body

若要对消息进行分组，可以使用相关expression。对于您的示例（按_前的文件名前缀分组消息），可能如下所示：

private final Expression CORRELATION_EXPRESSION = new Expression() {
    @Override
    public <T> T evaluate(Exchange exchange, Class<T> type) {
        final String fileName = exchange.getIn().getHeader(Exchange.FILE_NAME, String.class);
        final String correlationExpression = fileName.substring(0, fileName.indexOf('_'));
        return exchange.getContext().getTypeConverter().convertTo(
                type,
                correlationExpression
        );
    }
};

并将其传递给聚合器:

from("file:inputDirectory")
    .aggregate(CORRELATION_EXPRESSION, AggregationStrategies.groupedExchange())
    ...