Apache camel：路由中的多线程

Question

String postProcessor=“file：

from（processFiles）.threads（10）.routeid（“someid”）
.to（“bean：someBean”）；

从（postProcessor).routeid（“PostProcress”）.到（“bean：PostProcessorBean”）；

解决方案已经到位。但目前需要更多的时间。因此，我们尝试在camel路由级别使用线程，以便可以同时处理多个文件。现在有了这个，我们可以最小化时间，但不能进行后处理（即步骤2）

Answer 1

使用聚合器模式：

from("file:src/main/resources/data/parallel-file-processing?noop=true")
    .threads(10)
    .process(new PreProcessor())
    .aggregate(constant(true), new ArrayListAggregationStrategy())
    .completionFromBatchConsumer()
    .process(new PostProcessor());

preprocessor实现processor接口，在10个独立的线程中执行耗时的工作。

预处理的结果用aggrege聚合，使用取自Camel聚合器文档的ArrayListAggregatorStrategy:

//simply combines Exchange body values into an ArrayList<Object>
public class ArrayListAggregationStrategy implements AggregationStrategy {
    public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
        Object newBody = newExchange.getIn().getBody();
        List<Object> list = null;
        if (oldExchange == null) {
            list = new ArrayList<Object>();
            list.add(newBody);
            newExchange.getIn().setBody(list);
            return newExchange;
        } else {
            list = oldExchange.getIn().getBody(List.class);
            list.add(newBody);
            return oldExchange;
        }
    }
}

private static class PostProcessor implements Processor {
    @SuppressWarnings("unchecked")
    @Override
    public void process(final Exchange exchange) throws Exception {
        final Object body = exchange.getIn().getBody();
        final List<GenericFile<File>> list = (List<GenericFile<File>>) body;
        for (final GenericFile<File> genericFile : list) {
            LOG.info("file = " + genericFile.getAbsoluteFilePath());
        }
    }
}

from("file:src/main/resources/data/parallel-file-processing?noop=true")
    .threads(10)
    .process(new PreProcessor())
    .aggregate(constant(true), new ArrayListAggregationStrategy())
    .completionFromBatchConsumer()
    .parallelProcessing()  // <- for parallel processing
    .process(new PostProcessor());