什么是Camel？

Camel框架的核心是一个路由引擎，或者更确切地说是一个路由引擎构建器。它允许您定义自己的路由规则，决定从哪个源接收消息，并确定如何处理这些消息并将其发送到其他目标。

Camel提供更高层次的抽象，使您可以使用相同的API与各种系统进行交互，而不管系统使用的协议或数据类型如何。 Camel中的组件提供了针对不同协议和数据类型的API的特定实现。开箱即用，Camel支持80多种协议和数据类型。

Getting started

源码地址：https://github.com/camelinaction/camelinaction.git

下面是一个拷贝文件的例子，将文件从data/inbox拷贝到data/outbox

1 添加maven依赖

<dependencies>
  <dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-core</artifactId>
    <version>2.15.6</version>
  </dependency>
</dependencies>

2 代码

public class FileCopierWithCamel {

    public static void main(String args[]) throws Exception {
        // create CamelContext
        CamelContext context = new DefaultCamelContext();

        // add our route to the CamelContext
        context.addRoutes(new RouteBuilder() {
            public void configure() {
                /**
                  file: 表示使用文件Component
                  from 表示从哪里获取数据，进行消费
                  to  表示将数据生产到哪里
                 */
                from("file:data/inbox?noop=true").to("file:data/outbox");
            }
        });

        // start the route and let it do its work
        context.start();
        Thread.sleep(10000);

        // stop the CamelContext
        context.stop();
    }
}

Camel概念

CamelContext

Camel的容器，通过CamelContext可以访问内部服务：Components，Endpoints，Endpoints，Registry等等

image.png

Routes

通过路由可以实现：客户端与服务端，生产者与消费者的解耦

比如：从ftp服务上获取订单信息，将其发送到JMS队列，可以通过如下路由表示

    //from可以理解成消费者：表示从ftp服务上获取数据进行消费
    from("ftp://rider.com/orders?username=rider&password=secret")
    //to可以理解成生产者：表示将数据发送给jms
    .to("jms:incomingOrders");

image.png

endpoint URI

可以简单理解成消息的地址

对于消费者(from方法)来说,表示消息从哪里来
对于生产者(to方法)来说，表示消息到哪里去

image.png

如上图所示

Scheme：指明使用的是FtpComponent

Context path: ftp服务和端口号，以及文件路径

Options:一些操作配置，每个组件都不同

Exchange

Message的容器，其的内部属性，如下图所示

image.png

Message

消息数据的基本实体

MEP

Exchange支持多种消息交换模式 (MEPs)，通过其内部持有的pattern属性进行区分

下面介绍2种常用的交互模式

InOnly ：单向消息模式(也称为事件消息)，简言之：不需要等待对方的响应
InOut ：请求响应模式，例如：基于http的传输，通常是此模式，客户端请求web页面，等待服务端的回应

InOut模式包含In message 与 Out message，而InOnly模式只包含In message

Exception

如果路由期间发生错误，此属性将被赋值

Properties

Exchange的消息头，Camel本身和开发者可以设置或读取属性值

Endpoints

Endpoints是模拟通道末端的camel抽象，充当一个工厂，用于创建消息的producer和consumer

Component

创建Endpoints的工厂，一个Component的实现，通常有一些传输属性需要设置。例如，JMS-Component要求在其上设置ConnectionFactory，以便对所有JMS通信使用相同的消息代理

Component，Endpoints和Exchange的关系如下图所示：

image.png

内部组件介绍

Direct Component

基于内存的同步消息组件

使用Direct组件，生产者直接调用消费者。因此使用Direct组件的唯一开销是方法调用。

Direct的线程模型

由于生产者直接调用消费者

因此：调用者与camel的消费者共用一个线程

image.png

SEDA Component

基于内存的异步消息组件：生产者和消费者通过BlockingQueue交换消息，生产者与消费者是不同的线程

如果VM在消息尚未处理时终止，则seda不会实现消息的持久化或恢复，因此有丢失消息的风险

消费者视角

Consumer thread pool

SedaConsumer内部持有一个线程池，默认是1个线程，可以通过concurrentConsumers指定线程数

代码如下所示

from("seda:start?concurrentConsumers=2")
    .to("log:A")
    .to("log:B");

image.png

Threads thread pool

Consumer thread pool中的每个线程，还可以开启新的线程池，代码如下所示

from("seda:start?concurrentConsumers=2")
            .to("log:A")
            // create a thread pool with a pool size of 5 and a maxi- mum size of 10.
            .threads(5, 10) 
            .to("log:B");

image.png

如上图所示：consumer线程执行完"log:A"后，将后续任务提交给"Threads thead pool",然后就直接返回了

生产者视角

异步发送消息

生产者发完消息，立刻返回，不需要等待消息消费成功

 //InOnly消息模式
 producerTemplate.sendBody("seda:start", body);

同步发送消息

生产者发完消息，会阻塞，直到消费成功

 //InOut消息模式
 producerTemplate.requestBody("seda:start", body);

实现原理：SedaProducer通过CountDownLatch信号量进行等待，当数据消费成功后，消费者修改CountDownLatch信号量，唤醒SedaProducer，然后消费者才返回。

Camel使用

消息发送

Camel可以使用ProducerTemplate将消息发送到endpoint，或从endpoint请求数据

我们可以使用@Produce创建ProducerTemplate，代码如下

   import org.apache.camel.Produce;
   import org.apache.camel.ProducerTemplate;
   public class ProducePojo {
     @Produce
     private ProducerTemplate template;
     public String sayHello(String name) {
       //发消息到一个activemq端点
       return template.requestBody("activemq:queue:sayhello",
                                   name, String.class);
   } }

为了确保ProducerTemplate可以注入到ProducePojo类，需要将ProducePojo配置到spring上下文

    <beans xmlns="http://www.springframework.org/schema/beans" ...>
     <bean id="activemq"
           class="org.apache.activemq.camel.component
                  .ActiveMQComponent">
       <property name="brokerURL"
                 value="tcp://localhost:61616"/>
     </bean>
     <bean id="producer"
           class="org.camelcookbook.extend.produce
                  .ProducePojo"/>
     <camelContext xmlns="http://camel.apache.org/schema/spring"/>
   </beans>

方法调用

比如我要调用MyBean的myMethon,可以通过注解或java DSL

如果参数是对象类型，camel也会自动转型

以下代码表示接收到someEndpoint的消息后，调用myBean.myMethod方法

    //注意：要确保MyBean被camelContext或springContext加载
  public class MyBean {
     //注解的方式
     @Consume(uri="someEndpoint")
     public String myMethod(ParamBean message) {
       //...
  } } 
  //java DSL
  from("someEndpoint")
     .bean(MyBean.class, "myMethod");

  //通过ProducerTemplate调用此方法
  ParamBean param = genTestParam();
  template.requestBody("someEndpoint", param);

这里其实用的的是camel的内部组件Bean Component，具体用法可以参考如下官方文档

Bean Component: http://camel.apache.org/bean.html

关于参数的传递，可以参考

Bean Binding: http://camel.apache.org/bean-binding.html

自定义Processor

Processor是camel中的基本功能元素，自定义Processor非常易于在路由中编写和使用
定义一个将订单数据转成csv格式的Processor

public class OrderToCsvProcessor implements Processor {

    public void process(Exchange exchange) throws Exception {
        String custom = exchange.getIn().getBody(String.class);

        String id = custom.substring(0, 9);
        String customerId = custom.substring(10, 19);
        String date = custom.substring(20, 29);
        String items = custom.substring(30);
        String[] itemIds = items.split("@");

        StringBuilder csv = new StringBuilder();
        csv.append(id.trim());
        csv.append(",").append(date.trim());
        csv.append(",").append(customerId.trim());
        for (String item : itemIds) {
            csv.append(",").append(item.trim());
        }

        exchange.getIn().setBody(csv.toString());
    }

}

定义路由规则

from("quartz://report?cron=0+0+6+*+*+?")
    .to("http://riders.com/orders/cmd=received&date=yesterday")
    .process(new OrderToCsvProcessor())
    .to("file://riders/orders?fileName=report-${header.Date}.csv");

异常处理

基本用法

camel支持"异步重试，延迟重试"等多种处理方式

//通用异常处理
errorHandler(defaultErrorHandler()
                    //异步重试(默认同步)
                    .asyncDelayedRedelivery()
                    .maximumRedeliveries(2)
                    .redeliveryDelay(1000)
                    .retryAttemptedLogLevel(LoggingLevel.WARN));

//如果是JmsException，则只需自定义processer
onException(JmsException.class)
    .handled(true)
    .process(new GenerateFailueResponse());

//如果IOException,则重试3次，依旧失败，则执行to对应的动作
onException(IOException.class).maximumRedeliveries(3)
    .handled(true)
    .to("ftp://gear@ftp.rider.com?password=secret");

from("file:/rider/files/upload?delay=3600000")
    .to("http://rider.com?user=gear&password=secret");

上面的代码，其异常处理的作用域是整个context

Camel也支持route作用域的异常处理，如下代码所示

from("direct:step1")
        .bean(Step1.class, "success")
         //异常处理，作用域是当前路由
        .onCompletion().onFailureOnly()
         //如果失败，则执行onFailure方法
        .bean(Step1.class, "onFailure")
        .end()
        .to("direct:step2");

注意onCompletion的方式，是异步的，如果想同步处理异常可以参考camel的Synchronization使用方式

一个异常处理的例子

场景描述

顺序执行step1,step2,step3,如果某一步失败，回滚之前的每一步

比如step3执行失败，回滚step2,step1

解决方案

通过 onCompletion().onFailureOnly()方法对每一步设置失败回调函数,

下面的代码模拟了step3执行失败的场景，从日志可以看出camel按顺序执行了step2和step1的失败回调方法

public class RollbackTest extends CamelTestSupport {
  
    @Override
    public void setUp() throws Exception {
        deleteDirectory("target/mail/backup");
        super.setUp();
    }

    @Test
    public void testRollback() throws Exception {
        template.sendBodyAndHeader("direct:step1", "bumper", "to", "FATAL");
    }

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:step1")
                        .bean(Step1.class, "success")
                        .onCompletion().onFailureOnly()
                         //如果失败，调用step1的onFailure方法
                        .bean(Step1.class, "onFailure")
                        .end()
                        .to("direct:step2");

                from("direct:step2")
                        .bean(Step2.class, "success")
                        .onCompletion().onFailureOnly()
                        .bean(Step2.class, "onFailure")
                        .end()
                        .to("direct:step3");

                from("direct:step3")
                        .bean(Step3.class, "fail")
                        .onCompletion().onFailureOnly()
                        .bean(Step3.class, "onFailure")
                        .end()
                        .log("888:end");
            }
        };
    }
}

Apache Camel 调研

什么是Camel？

Getting started

Camel概念

CamelContext

Routes

endpoint URI

Exchange

Message

MEP

Exception

Properties

Endpoints

Component

内部组件介绍

Direct Component

Direct的线程模型

SEDA Component

消费者视角

Consumer thread pool

Threads thread pool

生产者视角

异步发送消息

同步发送消息

Camel使用

消息发送

方法调用

自定义Processor

异常处理

基本用法

一个异常处理的例子

查找文章

文章分类

最新发布

联系我们

谢谢，您的信息已成功发送。

请填写信息。

小范资讯

Apache Camel 调研

什么是Camel？

Getting started

Camel概念

CamelContext

Routes

endpoint URI

Exchange

Message

MEP

Exception

Properties

Endpoints

Component

内部组件介绍

Direct Component

Direct的线程模型

SEDA Component

消费者视角

Consumer thread pool

Threads thread pool

生产者视角

异步发送消息

同步发送消息

Camel使用

消息发送

方法调用

自定义Processor

异常处理

基本用法

一个异常处理的例子

查找文章

文章分类

最新发布

联系我们

谢谢，您的信息已成功发送。

请填写信息。