探索Apache Camel Core

探索Apache Camel Core – Seda组件

Apache Camel中的seda组件与我在之前的博客中介绍的direct组件非常相似，但是以异步的方式。为此，它使用java.util.concurrent.BlockingQueue作为默认实现来使消息排队并与主Route线程断开连接，然后在单独的线程中处理消息。由于此BlockingQueue ，您需要了解用法和config选项。

需要注意的一个选项是异步处理，默认情况下队列大小是不受限制的，这意味着它将增长到您允许的内存大小。为此，请设置size=1000 。让我们来看一个例子。

package camelcoredemo;import org.slf4j.*;
import org.apache.camel.*;
import org.apache.camel.builder.*;
import org.apache.camel.main.Main;
import java.io.*;public class SedaDemoCamel extends Main {static Logger LOG = LoggerFactory.getLogger(SedaDemoCamel.class);public static void main(String[] args) throws Exception {SedaDemoCamel main = new SedaDemoCamel();main.enableHangupSupport();main.addRouteBuilder(createRouteBuilder1());main.addRouteBuilder(createRouteBuilder2());main.addRouteBuilder(createRouteBuilder3());main.run(args);}// The file poller routestatic RouteBuilder createRouteBuilder1() {return new RouteBuilder() {public void configure() {from("file://target/input?preMove=staging&move=.processed").process(new Processor() {public void process(Exchange msg) {CamelContext camelContext = msg.getContext();ProducerTemplate producer = camelContext.createProducerTemplate();String text = msg.getIn().getBody(String.class);String fileName = (String)msg.getIn().getHeader("CamelFileName");boolean specialFile = fileName.endsWith("_SPECIAL.dat");if (specialFile)producer.sendBody("seda:specialRoute", text);elseproducer.sendBody("seda:normalRoute", text);}});}};}// The special file processing routestatic RouteBuilder createRouteBuilder2() {return new RouteBuilder() {public void configure() {from("seda:specialRoute").process(new Processor() {public void process(Exchange msg) {LOG.info("Processing special file: " + msg);}});}};}// The normal file processing routestatic RouteBuilder createRouteBuilder3() {return new RouteBuilder() {public void configure() {from("seda:normalRoute").process(new Processor() {public void process(Exchange msg) {LOG.info("Processing normal file: " + msg);}});}};}
}

您会注意到，该演示代码与direct组件演示非常相似，几乎没有区别。首先，我们使用seda端点。其次，在文件轮询器中，我们读取了整个文件的内容文本。我们之所以这样做，是因为我们现在要传递到将在单独的线程上运行的异步Route 。轮询器配置为在第一个Route结束后立即将处理的文件移动到其他文件夹中。因此，我们必须确保处理Route不依赖于File的路径，因此我们将加载整个文本。

另一个有趣的seda选项是，您可以设置并发线程的数量以接收消息来处理它们！假设您的普通文件流量很大，那么您可以配置为在该部分使用更多线程（默认仅为一个线程）。

from("seda:normalRoute?concurrentConsumers=10")
.process(new Processor() {public void process(Exchange msg) {LOG.info("Processing normal file: " + msg);}
});

要验证您正在同时运行，可以轻松配置记录器以显示线程名称。例如，使用log4j ，可以使用以下模式：

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p %t [%c] - %m%n

您可以探索Seda组件中可用的更多选项。尝试一下路线，亲自体验一下。

参考：在A Programmer's Journal博客上，从我们的JCG合作伙伴 Zemian Deng 探索Apache Camel Core – Seda组件。