發掘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死信隊列中。