java - Apache Camel Splitter、线程池和 JMS

Question

我在 spring xml 中定义了以下路由来拆分文本文件中的行并将每行发送到 JMS 队列

<bean id="myPool" class="java.util.concurrent.Executors" factory-method="newCachedThreadPool"/>

<camelContext id="concurrent-route-context" xmlns="http://camel.apache.org/schema/spring" trace="true">
    <route id="inbox-threadpool-split-route">
        <from uri="{{inbox.uri}}" />
            <log message="Starting to process file: ${header.CamelFileName}" />
            <split streaming="true" executorServiceRef="myPool">
                <tokenize token="\n" />
                <to uri="{{inventory.queue.uri}}" />    
            </split>
            <log message="Done processing file: ${header.CamelFileName}" />
    </route>
</camelContext>

inbox.uri 是监听目录中文件的文件组件 uri，而 inventory.queue.uri 是连接到 JMS 服务器(Tibco EMS 6.X 版本)中队列的 JmsComponent uri。 JmsComponent uri 很简单，如“JmsComponent:queue:?username=&password="

上面的路由可以正确运行，但是从文件中分割出来的行不会作为 JMS 消息发送到队列(即程序运行后队列仍然为空)

如果我从拆分器定义中删除 executorServiceRef="myPool"(其余定义如 )，则拆分的消息可以一一传递到 JMS 队列。

如果我用“直接”端点替换“to”uri，那么无论分割器中是否使用线程池，都可以传递分割的消息

JmsComponent 中是否需要任何特殊设置才能使其与 Splitter + 线程池一起使用？或者我错过的任何其他配置？

======= 编辑于 20150731 =======

在使用 1000 行的大型 CSV 文件进行测试时，我遇到了上述问题。如果我用一个小文件(例如仅 10 行)进行测试，我可以看到消息被传递到 inventory.queue，但从日志来看，完成拆分并将消息传递到队列似乎需要 10 秒。 .下面捕获了日志:

2015-07-31 11:02:07,210 [main           ] INFO  SpringCamelContext             - Apache Camel 2.15.0 (CamelContext: concurrent-route-context) started in 1.301 seconds
2015-07-31 11:02:07,220 [main           ] INFO  MainSupport                    - Apache Camel 2.15.0 starting
2015-07-31 11:02:17,250 [://target/inbox] INFO  inbox-threadpool-split-route   - Done processing file: smallfile.csv

查看从 11:02:07 开始的路线，并在 11:02:17(即 10 秒)显示“完成处理...”语句

如果我再次使用 5 行的 CSV 进行测试，将需要 5 秒...似乎每行需要 1 秒才能拆分并传递到 JMS 队列...这非常慢

如果我将“to uri”更改为“direct”而不是“JMS”，则拆分可以在一秒钟内很快完成

此外，从 JMS 监听器日志来看，它能够在同一秒内接收到所有 10 条消息。看起来 Splitter 会读取并拆分整个文件，为所有 10 行“准备”10 条 JMS 消息，然后将所有消息传递到队列，但不会“拆分 1 行并立即传递 1 条 JMS 消息”。 .

是否有任何选项或配置可以改变拆分器行为并增强拆分性能？

Answer 1

我在使用带有标记化的拆分器处理 14G 文件时遇到了类似的问题。正如 Claus 在 Parsing Large Files with Apache Camel 上的帖子所指出的那样，我能够通过使用聚合器来克服性能难题。

聚合批量消息后，我使用生产者模板将这些消息路由到消息传递系统。希望有帮助。