使骆驼路线并行运行答案

【问题标题】：Making Camel routes run in parallel使骆驼路线并行运行
【发布时间】：2016-11-12 21:07:06
【问题描述】：

我有一个使用骆驼路线做一些基本 ETL 的应用程序。每个路由都配置为从一个表中获取一些数据，进行一些转换并将其安全地放入不同模式的同一个表中。所以骆驼路线和桌子之间是一对一的关系。

假设我有这两条路线：

from("direct:table_1").routeId(table1Route)
    .setBody("SELECT * FROM table_1)
    .to("jdbc:source_schema").split(body()).streaming()
    .process("someProcessor")
    .to("sql:INSERT INTO table_1 ... ?dataSource=target_schema");

from("direct:table_2").routeId(table2Route)
    .setBody("SELECT * FROM table_2)
    .to("jdbc:source_schema").split(body()).streaming()
    .process("someProcessor")
    .to("sql:INSERT INTO table_2 ... ?dataSource=target_schema");

当向direct:table_1 和direct:table_2 端点发送start processing 消息时，一切运行正常并且数据被移动到目标架构中。

但是查看日志我可以看到表 2 记录仅在表 1 记录完成后才开始移动。这对我的应用程序来说绝对是不行的，因为有些表非常大，一次移动一个表需要很长时间才能运行。

我的问题是我做错了什么，我该如何解决这个问题，以便数据移动并行发生。

【问题讨论】：

我不明白为什么如果您并行触发两条路由不应并行执行。如何触发路由？
这正是我的想法。向两条路由发送start-processing 将使它们并行运行。即使发送发生相隔纳秒，路由 2 也不会在路由 1 完成之前开始处理。
那么您的代码将start-processing 发送到两条路由是从两个不同的线程执行的吗？如果你不使用两个线程，那么路由当然不会并行运行。
是因为你在两条路线上使用了相同的 id 吗？ routeId(table1Route)
没有。那是我修复的复制和粘贴问题。感谢您发现它。

标签： apache-camel

【解决方案1】：

我会尝试这样的：

from("start").multicast().parallelProcessing().to("seda:table1", "seda:table2");

基本上我有：

使用多播发送到多个接收者，并使用并行处理尝试并行发送到两个端点。
我已将您的直接端点替换为 seda 端点。如果您不需要同步端点，则改用 seda 可能会有所帮助。

您还可以尝试使用.threads() 语法进行多线程处理。

如果您想在运行时计算您的表端点，您可以将 .multicast() 替换为 .recipientlist()

【讨论】：

非常感谢。所以使用seda 是解决方案。看来我对直接可以用于什么有误解。在我看来，direct 就像一个队列。
它确实像一个内存队列，但默认情况下它比异步的 seda 更同步。当您想与 camelcontext 之外的端点进行通信时，您还可以使用 vm 和 direct-vm 端点。

【解决方案2】：

或者，如果使用 xml，可以通过以下方式实现：

<routeContext id="xxxRoute" xmlns="http://camel.apache.org/schema/spring">
    <route id="xxxRouteId">
        <from uri="activemq:queue:{{xxx.queue}}" />
        <multicast parallelProcessing="true">
            <pipeline>
                <to uri="file://?fileExist=Append"></to>
            </pipeline>
            <pipeline>
                <to uri="sql:{{sql.xxxx.insertQuery}}"></to>
            </pipeline>
        </multicast>
    </route>
</routeContext>

【讨论】：