使 Camel 路由并行运行 [英] Making Camel routes run in parallel

问题描述

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

I have an application which does some basic ETL using camel routes. Each route is configured to take some data from one table do some transformation and safe it into the same table on a different schema. So there is an one to one relationship between a camel route and a table.

假设我有这两条路线:

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 消息时，一切运行正常，数据移动到目标模式中终点.

Everything runs OK and the data is moved into target schema when sending a start processing message to both direct:table_1 and direct:table_2 end points.

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

However looking at the logs I can see table 2 records start being moved only after table 1 records are finished. That is definitely a no no for my application as some tables are quite big and and moving one table at a time would take a very long time to run.

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

My question is what I am doing wrong and how can I address this so the data movement happens in parallel.

推荐答案

我会尝试这样的事情:

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

基本上我有:

1. 使用多播发送给多个接收者，并使用并行处理尝试并行发送给两个端点.
2. 我已经用 seda 端点替换了您的直接端点.如果您不需要同步端点，则改用 seda 会很有好处.

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

You can also experiment with .threads() syntax for multithreading.

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

If you want to compute your table endpoints at runtime you can replace .multicast() with .recipientlist()

这篇关于使 Camel 路由并行运行的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！