发掘Apache Camel的力量

最近几年，ESB软件越来越受欢迎。如果大多数人通常知道什么是ESB，那么他们很少会清楚地了解这种体系结构的不同组件的确切作用。

例如，Apache ServiceMix由三个主要组件组成：Apache Karaf（OSGI容器），Apache ActiveMQ（消息代理）和Apache Camel。顺便问一下，骆驼到底是什么？什么是“ routing and mediation engine ”？有什么用？

我已经与Camel合作了大约一年，我认为-尽管根本不是Camel专家，但我现在有足够的后见之明，可以使用一些非常具体的示例让您发现Camel的兴趣和力量。为了清楚起见，在本文的其余部分中，我将使用Spring DSL –假设读者熟悉Spring语法。

用例

让我们想象一下，我们想使用Camel实现以下场景。产品信息请求将以平面文件（CSV格式）的形式发送到特定文件夹中。该文件的每一行都包含特定客户关于特定汽车型号的单个请求。我们希望向这些客户发送有关他们感兴趣的汽车的电子邮件。为此，我们首先需要调用Web服务以获取其他客户数据（例如，他们的电子邮件）。然后，我们必须从数据库中获取汽车特性（让我们说一个文本）。由于我们希望邮件看起来像样（例如HTML），因此也需要进行小的文本转换。

当然，我们不希望仅对请求进行顺序处理，而是希望引入一些并行性。同样，我们也不想多次将完全相同的邮件发送给不同的客户（而是将相同的唯一邮件发送给多个收件人）。利用我们后端的集群功能来平衡对Web服务的调用也将是一件很不错的事情。最后，在处理请求失败的情况下，我们希望以某种方式跟踪原始请求，以便例如可以通过邮政发送。
一个（可能的）骆驼实现：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://camel.apache.org/schema/springhttp://camel.apache.org/schema/spring/camel-spring.xsd "
><camelContext xmlns="http://camel.apache.org/schema/spring" errorHandlerRef="myDLQ"><!-- 2 redeliveries max before failed message is placed into a DLQ   --><errorHandler id="myDLQ" type="DeadLetterChannel" deadLetterUri="activemq:queue:errors" useOriginalMessage="true"><redeliveryPolicy maximumRedeliveries="2"/></errorHandler><!-- The polling of a specific folder every 30 sec --><route id="route1"><from uri="file:///Users/bli/folderToPoll?delay=30000&delete=true"/><unmarshal><csv/></unmarshal><split><simple>${body}</simple><setHeader headerName="customerId"><simple>${body[1]}</simple></setHeader><setHeader headerName="carModelId"><simple>${body[2]}</simple></setHeader><setBody><simple>${body[0]}</simple></setBody><to uri="activemq:queue:individualRequests?disableReplyTo=true"/></split></route><!-- The consumption of individual (jms) mailing requests --><route id="route2"><from uri="activemq:queue:individualRequests?maxConcurrentConsumers=5"/><pipeline><to uri="direct:getCustomerEmail"/><to uri="direct:sendMail"/></pipeline></route><!-- Obtain customer email by parsing the XML response of a REST web service --><route id="route3"><from uri="direct:getCustomerEmail"/><setBody><constant/></setBody><loadBalance><roundRobin/><to uri="http://backend1.mycompany.com/ws/customers?id={customerId}&authMethod=Basic&authUsername=geek&authPassword=secret"/><to uri="http://backend2.mycompany.com/ws/customers?id={customerId}&authMethod=Basic&authUsername=geek&authPassword=secret"/></loadBalance><setBody><xpath resultType="java.lang.String">/customer/general/email</xpath></setBody></route><!-- Group individual sendings by car model --><route id="route4"><from uri="direct:sendMail"/><aggregate strategyRef="myAggregator" completionSize="10"><correlationExpression><simple>header.carModelId</simple></correlationExpression><completionTimeout><constant>60000</constant></completionTimeout><setHeader headerName="recipients"><simple>${body}</simple></setHeader><pipeline><to uri="direct:prepareMail"/><to uri="direct:sendMailToMany"/></pipeline></aggregate></route><!-- Prepare the mail content --><route id="route5"><from uri="direct:prepareMail"/><setBody><simple>header.carModelId</simple></setBody><pipeline><to uri="sql:SELECT xml_text FROM template WHERE template_id =# ?dataSourceRef=myDS"/><to uri="xslt:META-INF/xsl/email-formatter.xsl"/></pipeline></route><!-- Send a mail to multiple recipients --><route id="route6"><from uri="direct:sendMailToMany"/><to uri="smtp://mail.mycompany.com:25?username=geek&password=secret&from=no-reply@mycompany.com&to={recipients}&subject=Your request&contentType=text/html"/><log message="Mail ${body} successfully sent to ${headers.recipients}"/></route></camelContext><!-- Pure Spring beans referenced in the various Camel routes --><!-- The ActiveMQ broker --><bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent"><property name="brokerURL" value="tcp://localhost:61616"/></bean><!-- A datasource to our database --><bean id="myDS" class="org.apache.commons.dbcp.BasicDataSource"><property name="driverClassName" value="org.h2.Driver"/><property name="url" value="jdbc:h2:file:/Users/bli/db/MyDatabase;AUTO_SERVER=TRUE;TRACE_LEVEL_FILE=0"/><property name="username" value="sa"/><property name="password" value="sa"/></bean><!-- An aggregator implementation --><bean id="myAggregator" class="com.mycompany.camel.ConcatBody"/></beans>

和（仅！）Java类的代码：

public class ConcatBody implements AggregationStrategy {public static final String SEPARATOR = ", ";public Exchange aggregate(Exchange aggregate, Exchange newExchange) {if (aggregate == null) {// The aggregation for the very exchange item is the exchange itselfreturn newExchange;} else {// Otherwise, we augment the body of current aggregate with new incoming exchangeString originalBody = aggregate.getIn().getBody(String.class);String bodyToAdd = newExchange.getIn().getBody(String.class);aggregate.getIn().setBody(originalBody + SEPARATOR + bodyToAdd);return aggregate;}	}}

一些解释

“ route1 ”处理传入的平面文件。文件内容首先被解组（使用CSV格式），然后分成行/记录。每行都将变成一个单独的通知，该通知将发送到JMS队列。
“ route2 ”正在使用这些通知。基本上，完成一个请求意味着依次执行两件事（“管道”）：获取客户电子邮件（route3）并向他发送邮件（route4）。请注意“ maxConcurrentConsumers”参数，该参数用于轻松满足我们的并行性要求。
“ route3 ”对如何获取客户电子邮件进行建模：只需通过解析（使用XPath）在两个后端节点上可用的（安全的）REST Web服务的XML响应即可。
“ route4 ”包含发送大量邮件的逻辑。每次收集到10个类似的发送请求（在我们的示例中，是对同一辆汽车的10个请求）（并且我们不准备等待超过1分钟），我们希望整个过程以新消息继续进行（或“骆驼语”中的“交换”是10条组合消息的串联。继续该过程意味着：首先准备邮件正文（路由5），然后将其发送到组（路由6）。
在“ route5 ”中，发出SQL查询，以便根据汽车型号获得适当的文本。在该结果上，我们应用了一个小的XSL-T转换（它将用xsl转换的输出替换当前交换主体）。
当输入“ route6 ”时，交换包含我们所需的一切。我们有收件人列表（作为标头），也有（正文中）要发送的html文本。因此，我们现在可以继续使用SMTP协议进行实际发送。
如果出现错误（例如临时网络问题）–在整个过程中的任何地方，Camel都会在放弃之前最多进行两次其他尝试。在后一种情况下，始发消息将由Camel自动放置到JMS死信队列中。