Camel中的几个重要概念之 Components

http://jnn.iteye.com/blog/320094

Components

Component 是一个容易混淆的名词，可能使用EndpointFactory会更合适，因为Component是创建Endpoint实力的工厂类。例如如果一个Camel应用使用了几个JMS 队列，那么这个应用首先需要创建一个叫JmsComponent（实现了Component接口）的实例，然后应用会调用这个JMSComponent对象的createEndpoint()方法来创建一个JmsEndpoint对象（这个对象实现了Endpoint接口）。事实上，应用代码并没直接调用Component.createEndoint() 方法，而是CamelContext对象通过找到对应的Component对象（我马上会在后续的文章中介绍）,并调用createEndpoint() 方法来实现的。

myCamelContext.getEndpoint("pop3://john.smith@mailserv.example.com?password=myPassword");

在getEndpoint()中使用的参数就是URI。这个URI的前缀(: 之前的那部分内容）描述了一个组件的名字，CamelContext对象内部维护着一个组件名字与Component对象的映射表。对于上面给定的URI例子来说，CamelContext对象会根据pop3前缀找到MailComponent类，然后CamelContext对会调用MailComponent的createEndpoint("pop3://john.smith@mailserv.example.com?password=myPassword") 方法。在createEndpoint()方法中，将把URI分割成一段段小的参数，这些小参数将被用来设置生成的Endpoint对象。

在上一段中，我提到的CamelContext对象维护了一个组件名到Component对象的映射表。但这个映射表是如何产生的呢？这里可以在通过代码调用CamelContext.addComponent(String componentName, Component component)来实现。下面的例子就是展示了如何给一个MailComponent对象注册上三个不同的名字。
Component mailComponent = new org.apache.camel.component.mail.MailComponent();
myCamelContext.addComponent("pop3", mailComponent);
myCamelContext.addComponent("imap", mailComponent);
myCamelContext.addComponent("smtp", mailComponent);

第二个方法也是最常用的方法，就是通过CamelContext对象来实现一个懒初始化。这个方法依赖于一套Camel内部的定义Component发现规则，开发者只要在实现Component接口的时候按照这一规则设置，就可以保证CamelContext能够正常发现这一Component。这里我们假设你所写的Class名字为 com.example.myproject.FooComponent，并且你想让Camel自动将这个component和"foo”这个名字相对应。为了做到这一点，你需要先写一个叫做"META-INF/services/org/apache/camel/component/foo" 属性文件，注意这个文件没有".properties"作为后缀名，在这个属性文件中只有一个class的条目，而这个条目的只就是你所写的类的全名。如下所示

META-INF/services/org/apache/camel/component/foo
class=com.example.myproject.FooComponent

如果你还想让Camel将上面的类和”bar” 这个名字联系起来，那你需要在同样的目录下在创建一个相同内容叫bar的文件。一旦完成了这些配置，你可以把 com.example.myproject.FooComponent class和这些配置文件一同打成一个jar 包，然后把这个jar包放你的CLASSPATH中。这样Camel就会通过分析这些属性文件的class 项目，通过使用reflectionAPI创建这个指定的类的实例。

正如我在Endpoint中说描述的， Camel提供了对多种通信协议一个开箱即用的支持。这种支持是建立在实现了Component接口的类以及让CamelContext对象自动建立映射关系的配置文件基础之上的。

在这一节的开始，我使用的这个例子来调用CamelContext.getEndpoint()。
myCamelContext.getEndpoint("pop3://john.smith@mailserv.example.com?password=myPassword");

在最开始举这个例子的时候，我说这个getEndpoint()方法的参数是一个URI。我这么说是因为Camel的在线问答以及Camel的源代码就把这个参数声明为一个URI。在现实生活中，这个参数是按照URL来定义的。这是因为Camel会从参数中通过一个简单的算法查找第一：来分析出组件名。为了了解其中的奥妙，大家可以回想一下我在前面 URL，URI，URN和IRI是什么中谈到的一个URI可以是URL或者URN。现在让我们来看一下下面的getEndpoint()调用。
myCamelContext.getEndpoint("pop3:...");
myCamelContext.getEndpoint("jms:...");
myCamelContext.getEndpoint("urn:foo:...");
myCamelContext.getEndpoint("urn:bar:...");

Camel会先找出这些component的标识，例如 "pop3", "jms", "urn" 和 "urn"。如果"urn:foo" 和"urn:bar" 能够别用来识别component，或者是使用"foo" 和"bar" （这一可以跳过这个"urn:"前缀）。所以在实际的编程中，大家更喜欢使用URL来制定一个Endpoint（使用":..."来描述的字符串）而不是用一个URN（使用"urn::..."来描述的字符串）。正因为我们没有安全按照URN的规定的参数来调用getEndpoint() 方法，所以这个方法的参数更像一个URL而不是一个URI。

===========================

1.component.createEndpoint()

2.if it's in from(...) clause, then call endpoint.createConsumer()

2.1 consumer will call doStart(),

2.2 create message and exchange, set exchange.in,then call processor to process this exchange.

you can create several exchange, each exchange mean one process.

3.if it's in to(...), then call endpoint.createProducer()

3.1producer call process(), then you can process this exchange, set this exchange.out, then it will be processed

Component

public class MemoryComponent extends DefaultComponent{

@Override
//memory:myMemory?param=value
//remaining = myMemory
protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
  System.out.println("createEndpoint(String uri, String remaining, Map<String, Object> parameters)");
  MemoryEndpoint endpoint = new MemoryEndpoint(uri, this);
  return endpoint;
}

}

Endpoint

public class MemoryEndpoint extends DefaultEndpoint{

public MemoryEndpoint() {
super();
System.out.println("MemoryEndpoint()");
}

public MemoryEndpoint(String endpointUri, Component component) {
super(endpointUri, component);
System.out.println("MemoryEndpoint(String endpointUri, Component component)");
}

@Override
public Consumer createConsumer(Processor processor) throws Exception {
  System.out.println("createConsumer(Processor processor)");
  Consumer memConsumer = new MemoryConsumer(this,processor);
  return memConsumer;
}

@Override
public Producer createProducer() throws Exception {
  System.out.println("createProducer()");
  Producer memProducer = new MemoryProducer(this);
  return memProducer;
}

@Override
public boolean isSingleton() {
System.out.println("isSingleton()");
return false;
}

}

Producer

public class MemoryProducer extends DefaultProducer{

public MemoryProducer(Endpoint endpoint) {
super(endpoint);
System.out.println("MemoryProducer(Endpoint endpoint)");
}

@Override
public void process(Exchange exchange) throws Exception {
  System.out.println("process(Exchange exchange)");
  Object body = exchange.getIn().getBody();
  System.out.println("add to queue:" + (String)body);
  exchange.getOut().setBody("producer-" + (String)body);
  StaticQueue.producerQueue.add("producer-" + (String)body);
}

}

Consumer

public class MemoryConsumer extends DefaultConsumer {

public MemoryConsumer(Endpoint endpoint, Processor processor) {
super(endpoint, processor);
System.out.println("MemoryConsumer(Endpoint endpoint, Processor processor)");
}

@Override
protected void doStart() throws Exception {
  super.doStart();

  System.out.println("doStart()");
  for (String str : StaticQueue.consumerQueue) {
   Exchange ex = this.getEndpoint().createExchange();
   Message msg = new DefaultMessage();
   msg.setBody(str);
   ex.setIn(msg);
   getProcessor().process(ex);
  }
}

@Override
protected void doStop() throws Exception {
  super.doStop();
  System.out.println("doStop()");

}

}

Processor

public class MemoryProcessor implements Processor {

@Override
public void process(Exchange exchange) throws Exception {
  System.out.println("MemoryProcessor-process(Exchange exchange)");
  Message in = exchange.getIn();
  System.out.println("in:" +in.getBody());
  Message out = exchange.getOut();
  System.out.println("out:" +out.getBody());

  //self defined processor need to transmit the msg from in to out,

//to("memory:mymemory").process(processor)

out.setBody(in.getBody() + " out processed");

  //in.setBody(in.getBody() + " in processed");

}

}

Test

public class Test {

/**
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
  if ( System.getProperty("log4j.configuration") != null )
   PropertyConfigurator.configure(System.getProperty("log4j.configuration"));
  else
   BasicConfigurator.configure();

  // TODO Auto-generated method stub
  Component memoryCom = new MemoryComponent();
  CamelContext context = new DefaultCamelContext();
  context.addComponent("memory", memoryCom);


  Component streamCom = new StreamComponent();
  context.addComponent("in", streamCom);
  context.addComponent("out", streamCom);

  context.addRoutes(new MemoryRouteBuilder());
  context.start();

  Runtime.getRuntime().addShutdownHook(new Thread(){
   @Override
   public void run() {
    System.out.println("queues:");
    for (String str : StaticQueue.producerQueue) {
     System.out.println("queue:"+str);
    }
   }

  });
}

}
class MemoryRouteBuilder extends RouteBuilder {
/** * A main() so we can easily run these routing rules in our IDE */

/** * Lets configure the Camel routing rules using Java code... */
public void configure() {
  //producer
//  from("timer://myTimer?period=2000").setBody().simple("myTimertest1").
//  to("memory:mymemory");

  //consumer
  MemoryProcessor processor = new MemoryProcessor();
  //from("memory:mymemory").to("file://C://test");
  //from("memory:mymemory").to("memory:mymemory").to("file://C://test");
  from("memory:mymemory").to("memory:mymemory").process(processor).to("file://test");
    //to("log:out");
  //  to("stream:out");

  //from("stream:in?promptMessage=Enter&promptDelay=5000").to("stream:out");
}
}

class StaticQueue {
public static Queue<String> consumerQueue = new LinkedBlockingQueue<String>();
public static Queue<String> producerQueue = new LinkedBlockingQueue<String>();
static {
consumerQueue.add("consumer-1");
consumerQueue.add("consumer-2");
}

}

Camel中的几个重要概念之 Components相关推荐

Camel中的几个重要概念之Routes, RouteBuilders 和 Java DSL
http://jnn.iteye.com/blog/320097 Route 一个route就是将一个输入队列中获得的消息,通过一步一步地设置好的逻辑判断(例如过滤器和路由规则)到达一个目标队列中(如 ...
图解面向对象中的聚合与耦合概念
图解面向对象中的聚合与耦合概念简单理解聚合与耦合在面向对象的设计中,我们经常会听到或用到聚合.耦合的概念.面向对象的目标就是设计出高聚合.低耦合的程序.然而,究竟什么是聚合.什么是耦合,恐怕每个 ...
oracle元数据存储在表空间,oracle db中数据和元数据的概念
本帖最后由 warehouse 于 2014-8-10 11:58 编辑 02年研究dw相关的产品和业务时第一次听说了元数据的概念,很多资料上是这样描述元数据的概念的:元数据是定义或者描述数据的数据. ...
apache camel_Apache Camel中的短重试与长重试
apache camel <骆驼设计模式>一书介绍了20种模式以及用于设计基于Apache Camel的集成解决方案的众多技巧和最佳实践. 每种模式都基于真实的用例,并提供了Camel特定 ...
Apache Camel中的短重试与长重试
<骆驼设计模式>一书介绍了20种模式以及用于设计基于Apache Camel的集成解决方案的众多技巧和最佳实践. 每个模式都基于真实的用例,并提供了Camel特定的实现细节和最佳实践. 为 ...
锁是网络数据库中的一个非常重要的概念
锁是网络数据库中的一个非常重要的概念,它主要用于多用户环境下保证数据库完整性和一致性.各种大型数据库所采用的锁的基本理论是一致的,但在具体实现上各有差别.目前,大多数数据库管理系统都或多或少具有自我 ...
Java中抽象类和接口在概念、语法和应用上的区别和关系
2019独角兽企业重金招聘Python工程师标准>>> 春招开始了,盆友们都忙着准备笔试.准备面试,复习学过的知识点,当然我也不例外,在这里祝每一个"有心人"心想 ...
NIO中的几个基础概念
一.NIO中的几个基础概念在NIO中有几个比较关键的概念:Channel(通道),Buffer(缓冲区),Selector(选择器). 首先从Channel说起吧,通道,顾名思义,就是通向什么的道 ...
Java正则表达式中的捕获组的概念及相关API使用
要弄清这三个方法,首先要弄清Java正则表达式中的捕获组的概念.捕获组也就是Pattern中以括号对"()"分割出的子Pattern.至于为什么要用捕获组呢,主要是为了能找出在一次 ...

Camel中的几个重要概念之 Components

Camel中的几个重要概念之 Components相关推荐

最新文章

热门文章