java - 使用 Apache Camel 拆分、聚合然后在一个大文件上进行流式写入

Question

我有一个大型数据库，我可以从中加载大量记录。我使用拆分器和聚合器模式以批处理模式处理它们。

我陷入困境的步骤是将每个批处理流式传输到一个 json 文件，我希望将它们全部存储在其中。步骤如下:

1. 从数据库中获取记录
2. 将它们作为 N 个批处理进行处理
3. 每个处理批处理都写入同一个大 json 文件中(缺少步骤)

我使用 File2 中的追加选项测试了此解决方案，但它确实在一个数组中写入多个数组。我可以扁平化这个 JSON，但这让我想到一个问题。

知道我有两个要求，如何停止路线运行:

1. 运行批处理后，开始时的大小不一定与最终相同。

我尝试使用completionFromConsumer，但不适用于 quartz 消费者。

我有这条路线:

   from(endpointsURL))
        .log(LoggingLevel.INFO, LOGGER, "Start fetching records")
        .bean(DatabaseFetch, "fetch")
        .split().method(InspectionSplittingStrategy.class, "splitItems")
        .aggregate(constant(true), batchAggregationStrategy())
        .completionPredicate(batchSizePredicate())
        .completionTimeout(BATCH_TIME_OUT)
        .log(LoggingLevel.INFO, LOGGER, "Start processing items")
        .bean(ItemProcessor, "process")
        .marshal()
        .json(JsonLibrary.Jackson, true)
        .setHeader(Exchange.FILE_NAME, constant("extract.json")))
        .to("file:/json?doneFileName=${file:name}.done")
        .log(LoggingLevel.INFO, LOGGER, "Processing done");

这里的问题正如我所想的那样，我的 extract.json 会在处理的每个批处理中被覆盖。我想将每个批处理追加到另一个批处理之后。

我不知道如何设计以及使用哪种模式来实现这一点。 Stream和 File具有良好的功能，但我可以以哪种方式使用它们？

Answer 1

您需要告诉 Camel 追加到文件(如果存在)，将 fileExists=Append 添加为文件端点的选项。