Apache Camel 调研
什么是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");
}
};
}
}
