EIP-Apache camel2.18使用篇

捐助

3625 次浏览

2019-3-7

编辑推荐:

本文来自于csdn，本文介绍了Apache Camel是什么以及Camel 要素、对一些错误处理的相关内容。

1、简介

Apache Camel 是一个非常强大的基于规则的路由以及媒介引擎，该引擎提供了一个基于POJO的企业应用模式(Enterprise Integration Patterns)的实现，你可以采用其异常强大且十分易用的API (可以说是一种Java的领域定义语言 Domain Specific Language)来配置其路由或者中介的规则。通过这种领域定义语言，你可以在你的IDE中用简单的Java Code就可以写出一个类型安全并具有一定智能的规则描述文件。这与那种复杂的XML配置相比极大简化了规则定义开发。当然Apache Camel也提供了一个对Spring 配置文件的支持。

Apache Camel 采用URI来描述各种组件，这样你可以很方便地与各种传输或者消息模块进行交互，其中包含的模块有 HTTP, ActiveMQ, JMS, JBI, SCA, MINA or CXF Bus API。这些模块是采用可插拔的方式进行工作的。Apache Camel的核心十分小巧你可以很容易地将其集成在各种Java应用中。

2、Camel 要素

2-1、Endpoint 端点

Camel中的Endpoint类似webservice中的endpoint，即某个资源的位置。Camel使用URI来定位一个endpoint. 比如from(“file://xxxx”), 可以是file:///edi/po/?include=.*\.txt, 代表/EDI/PO/下的所有txt文件，即为一个endpoint。

不同的endpoint都是通过URI格式进行描述的，并且通过Camel中的org.apache.camel.Component（endpoint构建器）接口的响应实现进行endpoint实例的创建。需要注意的是，Camel通过plug方式提供对某种协议的endpoint支持，所以如果读者需要使用某种Camel的endpoint，就必须确定自己已经在工程中引入了相应的plug。例如，如果要使用Camel对Netty4-Endpoint的支持，就需要在工程中引入Camel对Netty4的支持

<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-netty4</artifactId>
</dependency>

更多的端点可以参考:http://camel.apache.org/http.html

2-2、消息模型

在Camel中，由两部分抽象类构成消息

org.apache.camel.Message——在 Camel中，该基础实体包含着数据进行传输和路由

org.apache.camel.Exchange——在Camel的route中,消息在Route的各个节点中是以Exchange的形式传递的

2-1、Message

每一个Message（无论是inMessage还是outMessage）对象主要包括四个属性：MessageID、Header、Body和Attachment。

各个子元素描述如下

- MessageID

在系统开发阶段，提供给开发人员使用的标示消息对象唯一性的属性，这个属性可以没有值。

- Header

消息结构中的“头部”信息，在这个属性中的信息采用K-V的方式进行存储，并可以随着Message对象的传递将信息带到下一个参与路由的元素中。

- Body

Message的业务消息内容存放在这里

- Attachment

Message中使用attachment属性存储各种文件内容信息，以便这些文件内容在Camel路由的各个元素间进行流转。attachment同样使用K-V键值对形式进行文件内容的存储。但不同的是，这里的V是一个javax.activation.DataHandler类型的对象。

如下所示为Message的一个子类实现

public class DefaultMessage extends MessageSupport {
private boolean fault;
private Map<String, Object> headers;
private Map<String, DataHandler> attachments;
private Map<String, Attachment> attachmentObjects;
...
}

2-2、Exchange

在Camel的route中,消息在Route的各个节点中是以Exchange的形式传递的，所以对Exchange结构的理解对使用Camel来说是很重要的。

- ExchangeID

一个Exchange贯穿着整个编排的路由规则，ExchangeID就是它的唯一编号信息，同一个路由规则的不同实例（对路由规则分别独立的两次执行），ExchangeID不相同。

- MEP

message exchange pattern，Exchange中的pattern属性非常重要，它的全称是：ExchangePattern（交换器工作模式）。其实现是一个枚举类型：org.apache.camel.ExchangePattern。可以使用的值包括：InOnly, RobustInOnly, InOut, InOptionalOut, OutOnly, RobustOutOnly, OutIn, OutOptionalIn。从Camel官方已公布的文档来看，这个属性描述了Exchange中消息的传播方式。

- In message 输入消息

- Out message 输出消息

Exchange的结构如下图所示

2-4、CamelContext上下文

CamelContext从英文字面上理解，是Camel服务上下文的意思。CamelContext在Apache Camel中的重要性，就像ApplicationContext之于spring、ServletContext之于Servlet。

CamelContext横跨了Camel服务的整个生命周期，并且为Camel服务的工作环境提供支撑。

2-4-1 、CamelContext实现结构

上图是Apache Camel中实现了org.apache.camel.CamelContext接口的主要类。其中有两个实现类需要特别说明一下：SpringCamelContext和DefaultCamelContext。Camel可以和Spring框架进行无缝集成，例如可以将您的某个Processor处理器以Spring Bean的形式注入到Spring Ioc容器中，然后Camel服务就可以通过在Spring Ioc容器中定义的bean id（XML方式或者注解方式都行）取得这个Processor处理器的实例。

为了实现以上描述的功能，需要Camel服务能够从Spring的ApplicationContext取得Bean，而SpringCamelContext可以帮助Camel服务完成这个关键动作：通过SpringCamelContext中重写的createRegistry方法创建一个ApplicationContextRegistry实例，并通过后者从ApplicationContext的“getBean”方法中获取Spring Ioc容器中符合指定的Bean id的实例。这就是Camel服务和Spring进行无缝集成的一个关键点，如以下代码片段所示：

public class SpringCamelContext extends DefaultCamelContext implements InitializingBean, DisposableBean,
ApplicationContextAware {
private static final Logger LOG = LoggerFactory.getLogger(SpringCamelContext.class);
private static final ThreadLocal<Boolean> NO_START = new ThreadLocal<Boolean>();
private ApplicationContext applicationContext;
private EventComponent eventComponent;
private boolean shutdownEager = true;
public SpringCamelContext() {
}
public SpringCamelContext(ApplicationContext applicationContext) {
setApplicationContext(applicationContext);
}
}
...

2-5 数据转化在Camel中的应用

2-5-1、概念

数据转换包括两方面的内容：

- 1、数据格式变换—消息体的格式从一种形式转换为另一种形式：XML—->json。

- 2、数据类型的转换—消息体的格式从一种类型转换为另一种类型:java.lang.String—->javax.jms.TextMessage.

如下图所示

图说明了消息体转换从一种形式到另中形式的原则。这种转换原则适用于任何格式转换和类型转换。在Camel应用中你面临的数据转换大多数情况下都是格式转换，此时你需要在两个协议之间做格式转换。Camel内置了类型转换机制，可以自动在两种类型之间转换，这样就大大减轻了终端用户处理类型转换的负担。

2-5-2、使用EIP(企业集成模式)和java代码进行数据转化

数据映射是指两个不同的数据模型之间的映射的过程，数据映射是数据集成的一个关键因素。目前关于数据模型的标准有很多，这些标准由不同的机构或者委员会制定。因此，你会经常发现自己需要从公司的自定义数据模型映射到一个标准的数据模型。

Camel中的数据映射给用户提供了很大的自由，因为它允许你使用java代码，并不局限于使用特定数据映射工具,使用工具，起初可能看起来优雅，但结果可能出现不可能解决的问题。

在本节中，我们将会看到如何使用Processor进行数据映射，Processor是一个Camel API。Camel也可以使用bean进行数据映射,这是一个好的做法,因为它允许你的映射逻辑独立于Camel API

Camel提供了三种方式来使用此模式：

1、使用Processor

2、使用Bean

3、使用transform

2-5-2-1、Processor 处理器

Processor是Camel中的一个接口使用Processor，此接口只有一个方法：

public void process(Exchange exchange) throws Exception;

从上述方法可以看出，Processor是一个可以直接操作Camel的Exchange对象的API。它让你可以访问所有在Camel的CamelContext中传输的部分。CamelContext对象可以通过Exchange的getCamelContext方法得到。

让我们看一个例子。在骑士汽车零部件系统中，每天都会把收到的订单输出为CSV格式的文件。公司使用了一个自定义格式的订单实体，但是为了使事情变得简单，他们使用了一个file协议，此协议会根据输入的日期参数返回一个订单列表。你面对的挑战就在于：把file协议返回的数据映射为CSV格式，并写到文件中。

因为你想从一个快速原型开始,您决定使用Camel的Processor。

代码中使用Processor将一个自定义格式转换为CSV格式

public class OrderToCsvProcessor implements Processor {
public void process(Exchange exchange) throws Exception {
String custom = exchange.getIn().getBody(String.class);
String id = custom.substring(0, 10);
String customerId = custom.substring(10, 20);
String date = custom.substring(20, 30);
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());
}
}

首先从exchange中获取自定义格式的内容。它是String类型的，所以你传入了String参数，Exchange返回了String类型的结果。接着你从自定义格式中提取数据到本地变量。自定义格式的内容可以是任意的，但在这个例子中，它是一个定长的自定义格式。接着你通过构建一个以逗号分隔的字符串将自定义格式映射为CSV格式。最后，使用CSV格式的payload替换了自定义格式的payload。

你可以在下列路由中使用上面的OrderToCsvProcessor

public class OrderToCsvProcessorTest extends
CamelTestSupport {
@Test
public void testOrderToCsvProcessor()
throws Exception {
// this is the inhouse format we want to transform to CSV
String inhouse = "0000004444000001212320091208 1217@1478@2132";
template.sendBodyAndHeader("direct:start",
inhouse, "Date", "20091208");
File file = new File("target/orders/received
/report-20091208.csv");
assertTrue("File should exist", file.exists());
// compare the expected file content
String body = context.getTypeConverter().convertTo(String.class,
file);
assertEquals("0000004444,20091208,0000012123,1217
,1478,2132", body);
}
@Override
protected RouteBuilder createRouteBuilder() throws
Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
from("direct:start")
// format inhouse to csv using a processor
.process(new OrderToCsvProcessor())
// and save it to a file
.to("file://target/orders/received?fileName=report-${header.Date}.csv");
}
};
}

2-5-2-2、使用bean进行数据转换

使用bean进行数据转换是一个很好的实践，因为这种方式允许你使用任何你需要的java代码或者java类库。Camel对这点没有强加任何限制。Camel可以调用你开发的任意bean，甚至你可以使用已经存在的bean，而不需要重写或者重新编译他们。

让我们使用 bean代替前面的那个Processor。

/**
* A bean which translates an order in custom inhouse format
* to a CSV format.
*/
public class OrderToCsvBean {
public String map(String custom) {
String id = custom.substring(0, 10);
String customerId = custom.substring(10, 20);
String date = custom.substring(20, 30);
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());
}
return csv.toString();
}
}

2-5-2-3、使用transform方法进行数据转换

transform是可以用在路由中的一个方法，可以用来进行消息转换。利用表达式，transform()具有极大的灵活性，有时还可以节省时间。例如，假设你需要使用<br/>标记取代所有HTML格式数据中的的换行符。此时可以使用Camel内置的表达式和正则表达式的搜索和替换:

from("direct:start")
.transform(body().regexReplaceAll("\n", "<br/>"))
.to("mock:result");

上述路由使用transform()方法告诉Camel：消息要使用表达式进行转换。Camel利用建造者模式从路由中的表达式创建了整个表达式。这是通过方法调用链接在一起,这是建造者模式的本质。

在这个例子中，你联合使用了body()和regexReplaceAll()表达式，表达式应该这样读：获取body，并执行一个正则表达式，使用<br/>替换所有的\n，两个表达式组成了一个联合Camel表达式。

2-5-3、 Camel 类型转换器(Converter)

Camel提供了一个内置的类型转换系统,对常用类型之间进行自动转换。这个类型转换系统使Camel的各个组件直接可以很容易的协调工作，不会出现类型转换错误。从Camel用户的角度来看，在很多地方类型转换是内置在API中的，没有侵入性。例如，有如下代码：

String custom = exchange.getIn().getBody(String.class);

getBody方法接收了你所期望返回的类型为参数。在底层，类型转换系统将返回的结果转换成了Sring类型(如果需要的话)。

在本节中,我们将看看类型转换程序的内部机制。我们将解释Camel在启动时如何扫描类路径，进行动态注册类型转换器。我们还将向您展示在Camel路由中如何使用它,以及如何建立你自己的类型转换器。

2-5-3-1、 Camel类型转换器的运行机制

理解类型转换器的运行机制，首先需要理解在Camel中类型转换器是什么。图展示TypeConverterRegistry和TypeConverters之间的关系

在Camel启动时，所有的类型转换器都会注册到TypeConverterRegistry中。在运行时，Camel使用TypeConverterRegistry的lookup方法查找一个合适的TypeConverter来使用：

TypeConverter lookup(Class<?> toType, Class<?> fromType);

使用TypeConverter的convertTo方法，Camel可以将一种类型转换为另一种类型：

<T> T convertTo(Class<T> type, Object value);

2-5-3-2、使用Camel类型转换器

如前所述，Camel类型转换器被广泛使用，而且常常自动起作用。你可能要在一个路由中使用它们转换一个特定的类型，假设你需要将一些文件路由到JMS队列，并且使用javax.jmx.TextMessage类型的消息。你可以将每一个文件转换为String类型，迫使JMS组件使用TextMessage，Camel是很容易做到的—使用convertBodyTo方法，如下

from("file://riders/inbox")
.convertBodyTo(String.class)
.to("activemq:queue:inbox");

如果使用Spring XML，如下：

你可以省略java.lang.前缀 <convertBodyTo type="String"/>.

2-5-3-3、自定义类型转换器

在Camel中编写自己的类型转换器是很容易的。你已经看到类型转换器是什么样子。假设你想写一个converter用于将一个byte[]转换为PurchaseOrder对象。那么，你需要创建一个@Converter注解的类，类中包含转换方法，代码如下

@Converter
public final class PurchaseOrderConverter {
@Converter
public static PurchaseOrder toPurchaseOrder(byte[] data, Exchange exchange) {
TypeConverter converter = exchange.getContext().getTypeConverter();
String s = converter.convertTo(String.class, data);
if (s == null || s.length() < 30) {
throw new IllegalArgumentException("data is invalid");
}
s = s.replaceAll("##START##", "");
s = s.replaceAll("##END##", "");
String name = s.substring(0, 10).trim();
String s2 = s.substring(10, 20).trim();
String s3 = s.substring(20).trim();
BigDecimal price = new BigDecimal(s2);
price.setScale(2);
Integer amount = converter.convertTo(Integer.class, s3);
PurchaseOrder order = new PurchaseOrder(name, price, amount);
return order;
}
}

代码中，通过Exchange获取CamelContext，进而获取上级TypeConverter，用其进行String和byte转换。

现在你所需要做的是添加服务发现文件,文件名为：TypeConverter，位于META-INF目录中。正如前面所解释的那样,这个文件包含一行内容，用于Camel扫描这个包进而发现@Converter注解类。如图所示

3、路由

3-1、如何使用RouteBuilder

通过继承抽像类org.apache.camel.builder.RouteBuilder是最常见的一种方式。然后重载configure方法如下代码所示

public class MyRouteBuilder extends RouteBuilder {
/**
* Let's configure the Camel routing rules using Java code...
*/
public void configure() {
// here is a sample which processes the input files
// (leaving them in place - see the 'noop' flag)
// then performs content based routing on the message using XPath
from("file:src/data?noop=true&recusive=true")
.choice()
.when(xpath("/person/city = 'London'"))
.to("log:uk?showall=true")
.to("file:target/messages/uk")
.otherwise()
.to("file:target/messages/others");
}
}

现在只需要将该对象通过CamelContext类的addRoutes方法添加

3-2 创建路由-Java DSL

DSL领域定义语言，用来描述特定领域的特定表达。比如画图从起点到终点；路由中的从A到B。

示例代码如下

public class OrderRouterWithFilterTest extends CamelTestSupport {
@Override
protected CamelContext createCamelContext()
throws Exception {
// create CamelContext
CamelContext camelContext = super.createCamelContext();
// connect to embedded ActiveMQ JMS broker
ConnectionFactory connectionFactory =
new ActiveMQConnectionFactory("vm://localhost");
camelContext.addComponent("jms",
JmsComponent.jmsComponentAutoAcknowledge
(connectionFactory));
return camelContext;
}
@Test
public void testPlacingOrders() throws Exception {
getMockEndpoint("mock:xml").expectedMessageCount(1);
assertMockEndpointsSatisfied();
}
@Override
protected RouteBuilder createRouteBuilder()
throws Exception {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
// load file orders from src/data into the JMS queue
from("file:src/data?noop=true").to
("jms:incomingOrders");
// content-based router
from("jms:incomingOrders")
.choice()
.when(header("CamelFileName").endsWith(".xml"))
.to("jms:xmlOrders")
.when(header("CamelFileName").regex
("^.*(csv|csl)$"))
.to("jms:csvOrders")
.otherwise()
.to("jms:badOrders");
...
}
};
}
}

3-3、创建路由-Spring方式

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:broker="http://activemq.apache.org/schema/core"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">

<broker:broker useJmx="false" persistent="false" brokerName="localhost">
<broker:transportConnectors>
<broker:transportConnector name="tcp" uri="tcp://localhost:61616"/>
</broker:transportConnectors>
</broker:broker>
<bean id="jms" class="org.apache.camel.component.jms.JmsComponent">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://localhost:61616"/>
</bean>
</property>
</bean>
<bean id="downloadLogger" class="camelinaction.DownloadLogger"/>
<camelContext xmlns="http://camel.apache.org/schema/spring">
<route>
<from uri="file:src/data?noop=true"/>
<process ref="downloadLogger"/>
<to uri="jms:incomingOrders"/>
</route>
</camelContext>
</beans>

3-4、EIP路由模式

3-4-1 Content-based router

3-4-2 消息filters

3-4-4、RecipientList 多个接收者

消息动态的路由到一个多个结点，recipientList 会将消息复制多份发送到后面的节点,后面每个节点处理的消息不是同一份，recipientList支持在线程池里面运行。recipientList 效果如同 multicast

示例代码如下

<route autoStartup="true">
<from uri="direct:recipientList" />
<recipientList delimiter="," ignoreInvalidEndpoints="true">
<header>endlist</header>
</recipientList>
</route>

ProducerTemplate pt = context.getBean("camelTemplate", ProducerTemplate.class);
pt.send("direct:routingSlip", new Processor()
{
String endList = "bean:bean1,bean:changeInfo,bean:bean2";
public void process(Exchange exch) throws Exception
{
exch.getIn().setHeader("endlist", endList);
}
});

最后，结果都会路由到bean:bean1,bean:changeInfo,bean:bean2这3个节点上面

3-4-5、wireTap窃听器

4、错误处理

编写应用程序集成异构系统，如何处理意想不到的事件是一个很大的挑战。在单一系统中，由于你有完全的控制权，你可以处理这些事件并恢复它。当时通过网络集成起来的系统有额外的风险：网络连接可能断掉、远程系统可能不会及时响应，或者它可能无缘无故宕机。即使在你的本地服务器,也可能发生意外事件,如服务器的磁盘满了或服务器内存耗尽。不管哪种错误出现,您的应用程序应该准备处理它们。

在这些情况下,日志文件通常是记录意外事件的唯一证据，所以日志非常重要。Camel对日志记录和错误处理有广泛支持，确保您的应用程序可以持续运行。

在本章，你会发现Camel的错误处理是多么灵活,深入和全面，学会如何对它进行定制,以应对大多数情况下的异常。我们将讨论Camel提供开箱即用的所有错误处理程序,以及他们的最佳使用场景,所以你可以选择最适合您的应用程序的处理方式。您还将了解如何配置和掌握消息返回处理，Camel可以使用返回技术尝试从特定的错误中恢复过来。我们还可以看到异常处理策略，这些策略允许你对错误进行区分，只处理特定的错误，以及定义通用错误处理规则，实现路由级别的错误处理。最后我们看下错误处理的细粒度控制

4-1、理解错误处理

在走进Camel错误处理的世界之前，我们需要退一步,看看更普遍的错误。首先，错误一般分为两大类：可恢复的错误和不可恢复的错误。其次我们需要看看何时何地开始错误处理，因为错误的发生是有先决条件的

4-1-1-1、可恢复的错误和不可恢复的错误

当涉及到的错误，我们可以将他们分为可恢复的错误和不可恢复的错误，如下图所示

不可恢复的错误是指一个错误,无论你尝试多少次相同的操作，它仍是一个错误。在集成的项目中,这可能意味着试图访问的数据库表不存在,这将导致JDBC驱动程序抛出SQLException异常。

可恢复的错误,是一个临时错误,在接下来的尝试操作后，可能并不会再次出现这个错误。比如网络连接错误导致的java.io.IOException。在接下来的尝试操作后,网络问题可能已经解决了,您的应用程序可以继续运行。

作为一个Java开发人员在日常生活中,你可能会遇到这样的错误分类：可恢复的错误和不可恢复的错误。一般来说,异常处理代码使用两种分类之一,如下面两个代码片段所示。

第一个代码片段演示了一个常见的错误处理方式，将所有的异常认为是不可恢复的，放弃进一步的尝试，直接向调用者返回异常

public void handleOrder(Order order) throws OrderFailedException {
try {
service.sendOrder(order);
} catch (Exception e) {
throw new OrderFailedException(e);
}
}

下一个片段通过添加一些处理改善了这种情况，在抛出异常之前进行的尝试，尝试5次后，如果仍然失败，抛出异常

public void handleOrder(Order order) throws OrderFailedException {
boolean done = false;
int retries = 5;
while (!done) {
try {
service.sendOrder(order);
done = true;
} catch (Exception e) {
if (--retries == 0) {
throw new OrderFailedException(e);
}
}
}
}

上面的例子都缺少对错误类型的验证，即发生的错误时可恢复的还是不可恢复的？进而采取相应的措施。可恢复的情况下,你可以再试一次,不能恢复的情况下,你可以立即放弃并重新抛出异常。

在Camel中，可恢复的错误由Throwable和Exception代表，可以通过org.apache.camel.Exchange类中的下面两个方法进行存取：

void setException(Throwable cause);
Exception getException();

不可恢复的错误由一个消息代表，此消息有一个错误标识，此标识可以通过org.apache.camel.Exchange来存取。例如，设置”Unknown customer”作为一个错误消息，可以这么做：

Message msg = Exchange.getOut();
msg.setFault(true);
msg.setBody(“Unknown customer”);

错误标识必须使用setFault(true)方法设置。

那么,在Camel中，为什么这两种类型的错误代表不同?原因有两个：第一，Camel API的设计符合JBI(Java业务集成)规范，此规范中有个错误消息的概念。第二，Camel核心包中内置了错误处理，所以，只要Camel中抛出异常，Camel都可以捕获它，并把它设置到Exchange中作为可恢复错误，如下

try {
processor.process(exchange);
} catch (Throwable e) {
exchange.setException(e);
}

使用这种模式允许Came捕获和处理所有异常。Camel的错误处理机制就可以决定如何处理捕获的错误—　重试,传播错误返回给调用者，或者做别的事情。Camel的终端用户可以将不可恢复的错误设置为错误消息，Camel能做出相应的反应，停止路由该消息。

现在你已经了解了可恢复和不可恢复的错误，让我们总结下他们在Camel中是如何表示的：

1、异常(Exceptions)表示为可恢复错误。

2、错误信息表示为不可恢复的错误。

3625 次浏览