借助 Apache Camel 实现企业应用集成（EAI）在日益增长

IT 项目的需求通常会发生变化，这其中就包括与其他系统集成的需求。对于项目的成功来讲，能够快速地响应这样的变化是至关重要的，所以软件和开发过程必须要做到这一点。幸运的是，企业应用集成（ Enterprise Application Integration，EAI）在构建可扩展性的、可维护性的以及可胜任的集成解决方案方面，以一种创造性的方式为我们提供了所有的知识、技术以及最佳实践。

但是，大多数的集成方案会给我们带来一种困境：尽管它们功能完备并且对于苛刻的环境来说富有成效，但是在开始学习、部署和维护系统的时候，需要巨大的预先投资。

基于这个原因，当面对简单集成需求时，临时的解决方案看起来很有吸引力。但是它们会变得难以维护并且集成所要求的效率也会增加。使用 EAI 最佳实践将会解决这个问题，但如果他们自己实现的话，将会需要付出一些努力并且要掌握如何正确做事的知识。起初看起来阻力最小的解决路径最终可能是个死胡同。

那么，当面临简单的和复杂的集成任务时，我们怎样在避免早期巨额投资的同时又能高效完成任务呢？在本文中，我介绍的 Apache Camel 将会提供一个解决方案。我将会论证 Camel 能够解决复杂集成方面的挑战，它能够让你使用 EAI 的最佳实践，并且易于起步和掌握。同时，Camel 能够让你专注于提供业务价值，而不必处理一些框架所带来的复杂性。

我将会通过典型集成所面临挑战的实际例子来展示这一点并了解 Camel 是怎样帮助我们解决这些挑战的。集成方案通常在开始的时候很简单，但是随着时间的推移会出现新的集成需求，这些例子就是以这样的上下文进行呈现的。每次我都会被问到 Camel 是如何满足这些需求的，它的主要关注点在于管理复杂性和保持较高的生产效率。

依我来看，我选择 Apache Camel 是因为它为像 Service Mix 、 Mule ESB 、 OpenESB 以及 JBossESB 这样的完整 ESB 产品提供了杰出的和轻量级的替代方案。它的竞争对手可能是 Spring Integration ，如果你的项目已经使用了 SpringSource 技术，那么后者是一个尤其值得考虑的选项。你将会看到，你也可以同时使用 Camel 和 Spring。Gunnar Hillert在这里进一步讨论了替代方案。

简单的开始

集成的开始通常会很简单。例如，从 FTP 服务器获取一些文件并将其放在本地文件系统中。这种场景下，自己动手解决（do-it-yourself）的方案看起来很有吸引力。但是让我们更为仔细地观察一下。

自己动手解决的方案看起来可能会是这样的：

 
public class FTPFetch {
 
     public static void main(String[] args) {
        FTPClient ftp = new FTPClient();
        try {
              ftp.connect("host"); // try to connect
 
              if (!ftp.login("camel", "apache")) // login to server
              {
                  ftp.disconnect();
                  return;
              }
 
              int reply = ftp.getReplyCode();
 
              if (!FTPReply.isPositiveCompletion(reply)) {
                  ftp.logout();
                  ftp.disconnect();
                  return;
              }
 
               ftp.changeWorkingDirectory("folder");
 
              // get output stream for destination file
              OutputStream output = new FileOutputStream("data/outbox/file.xml");
 
              ftp.retrieveFile("file.xml", output); // transfer the file
              output.close();
 
              ftp.logout();
              ftp.disconnect();
 
        } catch (Exception ex) {
                  ex.printStackTrace();
 
        } finally {
                  if (ftp.isConnected()) {
                     try {
                           ftp.disconnect();
                     } catch (IOException ioException) {
                           ioException.printStackTrace();
                     }
              }
         } 
     }
}

这个方案使用了 Apache Commons 的 FTPClient 类。因为它仅仅是一个客户端并不能做更多的事情，我们需要自己建立 FTP 连接并做错误处理。但是，如果 FTP 服务器上的文件随后发生了变化会怎么样呢？我认为我们需要对其进行调度，使其周期性地运行。

现在，让我们看一下 Apache Camel。Apache Camel 是一个集成框架，它通过遵循 EAI 最佳实践来解决这种问题。Camel 可以视为包含现成集成组件的工具箱，同时也可以视为能够针对特定环境进行自定义的运行时，这是通过组合使用集成组件实现的。借助 Camel，我们可以这样解决上面提到的问题：

 
public class CamelRunner{
     public static void main(String args[]) throws Exception {
         Main camelMain = new Main();
         camelMain.enableHangupSupport(); //ctrl-c shutdown
         camelMain.addRouteBuilder(new RouteBuilder() {
                public void configure() {
 
from(
"ftp://host/folder?username=camel&password=apache&fileName=file.xml&delay=360000" )
       .to("file:data/outbox");
 
                }
        });
 
        camelMain.run(); //Camel will keep running indefinitely
     }
}

请注意 from 和 to 方法。Camel 将其称之为“路由”：数据从来源到目的地的路径。另外，数据不是以原始格式进行交换的而是封装在消息中：实际数据的容器。这类似于 SOAP 的信封，它包含主体区、附件以及头部。

消息的来源和目的地被称之为“端点”，正是通过它们 Camel 接收和发送数据。端点通过 URI 格式的字符串来指定，正如 from 和 to 方法的参数所示。所以，我们告知 Camel 做什么的方法就是声明式地创建端点之间的路由，然后使用 Camel 来注册这些路由。

剩下的就是样板式的代码了，当添加多个路由的时候可以进行重用，与直接和 FTP 服务器交互对比，这简单了很多。Camel 将会处理繁琐的 FTP 细节并定期询问服务器文件是否发生了变化，因为它已经被设置为一直运行下去。

清晰和紧凑的代码源于 Camel DSL，这是一个领域专用语言（Domain Specific Language），在这里领域（domain）指的就是 EAI。这意味着，不像其他的解决方案，从 EAI 问题域到 Camel 应用域并不会有转换：这两者实际上是相同的。这也有助于保持学习曲线平滑，相对来说入门门槛较低：一旦理解了你的 EAI 问题，再迈一小步就能使用 Camel 来实现它。

并不仅仅是你所编写的代码很简单：要使它运行起来，所需要只有 camel-core.jar 和 camel-ftp.jar 以及它们的依赖，加在一起只有几兆大小。这个主类可以通过命令行来运行。不需要复杂的应用服务器。实际上，因为 Camel 是如此轻量级，它可以嵌入到任何地方。选择自己动手解决的唯一基础就是框架会带来一些无效的依赖：Camel 易于理解、易于使用并易于运行。

增加复杂性

现在，让我们介绍一下越来越多的集成需求。我们不仅希望能够有更多的集成，还想保持它的可维护性。Camel 怎样应对这一点呢？

随着要做出的连接越来越多，我们只需要添加更多的路由到 Camel 中。这些新的路由可能会通过其他的端点来进行连接如 HTTP、JMS 以及 SMTP 等等。幸好，Camel 所支持的端点列表是可扩展的。很棒的一点是这些端点都了提供可重用的代码，这不需要你去编写。

当然，迟早你会需要不在这个列表中的事情。那么问题就变成了这样：我将自己的代码加入到 Camel 中的难度如何？在这种场景下，我们可以使用 Camel 中被称为组件（Component）的东西。组件定义了一个协议，在实现它的时候会让你的代码可用，就像通过 DSL 调用其他端点一样。

现在，我们知道可以添加越来越多的路由，几乎可以使用任何类型的协议来进行连接，不管这是不是 Camel 内置提供的。但有时，路由的数量会变得很庞大并且会发现你是在重复自己。我们想重用一些路由，甚至可能是将整体的方案拆分为独立且粗粒度的部分。

Camel 的重用策略基于一些特定的、内部的端点，这些端点只能由 Camel 使用。假设你需要重用一个已存在的路由，可以将这个路由重构为两个，通过内部的端点来连接。请看如下的代码：

初始：

 
//original
from(“<a href="ftp://server/path">ftp://server/path</a>”).
     to(“xslt:transform.xsl”).
         to(“http://server2/path”);

重构后：

 
//receiving from internal endpoint d1
from(“direct:d1”).
     to(“xslt:transform.xsl”).
         to(“<a href="http://server2/path">http://server2/path</a>”);
 
//sending to d1 
from(“<a href="ftp://server/path">ftp://server/path</a>”).
     to(“direct:d1”);
 
//also sending to d1
from(“file://path”).
     to(“xslt:other-transformation.xsl”).
         to(“direct:d1”);

这个连接端点是“direct”类型的。这种类型的端点只能在同一个 Camel 上下文中进行寻址。另一个有趣的端点类型是 VM。VM 端点可以在另外一个 Camel 上下文中进行寻址，只要这两个上下文运行在同一个 JVM 实例中即可。

Camel 上下文类似于路由的容器。每次当你运行 Camel 时，它会初始化一个上下文并查找里面的路由。所以当你运行 Camel 时，我们实际上运行的是一个上下文实例。

借助于 VM，能够在另外一个 Camel 上下文中对路由寻址是相当有用的。它提供了一种可能性，那就是将你的完整应用拆分为互相连接的模块，相对于 JMS 来说，这是一种更为轻量级的方式。

以下的图片展现了各种路由，它们能够在不同的 Camel 实例中传播，每个都运行在相同的 JVM 实例中并且通过 VM 端点进行相互寻址：

（点击图片放大）

我们已经将解决方案拆分为多个模块。现在，我们可以开发、部署和运行其他的模块，这些模块会发送到“Consumer Context”中，与“Producer Context1”和“Producer Context2”相独立。为了使最大的解决方案也能保持可管理性，这是关键的一点。

此时，使用应用服务器似乎是顺理成章的，因为它能够充分地支持模块化。或许你已经使用了某一个这样的产品。接下来一个常见的方式就是将Camel 打包成WAR 文件并部署到Tomcat 中。但是你也可以将其部署到一个完整的Java EE 应用服务器中，像JBoss、WebLogic 或WebSphere。其他可选的方案包括OSGi 容器甚至Google App Engine。

管理复杂性

数量庞大并不是应用增长的唯一方式。路由也可能在复杂性方面不断增长：信息可能会是各种数量的并且任意数量组合的传输类型、过滤、增强以及路由等。为了介绍Camel 在这个方面怎样提供帮助，首先让我们考虑怎样处理复杂的问题。

复杂的问题会在任何领域都出现，但是解决它们的总体策略通常是一样的：分而治之。我们会将问题拆分为更容易解决的子问题。然后这些方案再按照与分解相反的方式组合在一起形成整体的解决方案。

通过观察会发现这样的问题是经常发生的；借助于经验，能够识别出最优的方案。我所讨论的就是模式。EAI 模式已经由Gregor Hohpe 和Bobby Woolf 进行了分类和在线的总结

EAI 模式在本质上可以非常简单，通常表现为基本的操作如传输或过滤。最为重要的是，它们可以组合起来形成复杂的解决方案。这种组合本身也可以是模式。这种能力来源于所有的 EAI 模式具有相同的“接口（interface）”：信息可以进出模式。这样，模式就可以组合起来，这是通过接受一个模式的输出，并将这个输出作为另一个模块的输入来实现的。

广义地来说，这说明 EAI 问题就是模式的组合问题。这意味着解决 EAI 问题，即便是很复杂的问题，将会简化成寻找满足需求的组合。实现自己的模式当前也会有大量的复杂性，但是这已经进行了隔离并且是可管理的。

让我们考虑一个实际的模式作为例子。这个模式称为“复合消息处理器（Composed Message Processor）”，它实际上多个更基本模式的组合。当相同信息的各部分要由多个不同的组件进行处理时，要使用这个模式。Camel 并没有直接实现这个模式，但是实现了它的子模式。所以这是展示如何借助 Camel DSL 将模式组合起来的很好的例子。

以下是模式图，“Splitter”会将传入的信息拆分为各个部分，而“Router”将会决定要将它们发送到哪个系统：要么是“Widget Inventory”，要么是“Gadget Inventory”。这些系统可以认为会做一些业务相关的处理，然后返回处理后的信息。“Aggregator”将会把结果再次组合为一个输出信息。

（点击图片放大）

以下为 Camel 实现：

 
from("some:input")
     .setHeader("msgId") //give each message a unique id based on timestamp
         .simple("${date:now:S}")
     .split(xpath("//item")) //split the message into parts (msgId is preserved)
         .choice() //let each part be processed by the appropriate bean
             .when( xpath("/item[@type='widget']") )
                 .to("bean:WidgetInventory")
             .otherwise()
                 .to("bean:GadgetInventory")
         .end()
     .aggregate(new MyAggregationStrategy()) //collect the parts and reassemble
         .header("msgId") //msgId tells us which parts belong together
         .completionTimeout(1000L)
.to("some:output"); //send the result along

在这个实现中，“bean”实际上是基于 Bean 名字注册的 POJO，例如通过 JNDI。按照这种方式，我们可以在路由中执行自定义的逻辑。MyAggregationStrategy 也是自定义的代码，它指明了怎样组合处理过的各部分信息。

注意 split、choice 以及 aggregate 方法，它们直接对应于“Splitter”、“Router”以及“Aggregator”模式。Camel 对“复合消息处理器”模式的实现在本质上就是对上面图片的文本展现。所以，大多数情况下不必关心“Camel”的术语，只关心 EAI 的术语就可以了。结果就是 Camel 可以位于一个相对并不特别值得关注的地方，你可以把关注的重点放在理解问题以及识别合适的模式。这有助于提高解决方案的整体质量。

但是，并不是事事如此顺心。Camel 确实也有“它自己处理问题的方式”，也就是它自己背后的逻辑。当发生预料之外的事情时，你也会一头雾水不知所措。但是这些不足应该与 Camel 实际所节约的时间对应着看：其他的框架会有更为陡峭的学习曲线以及独特的技巧，而自己动手做则意味着你不能重用 Camel 所提供的伟大特性而是重复发明轮子。

毫无疑问，对管理复杂性和软件进化的探讨如果不涉及单元测试将是不完整的。Camel 可以嵌入到任何的类中，所以它也可以运行在单元测试之中。

关于集成测试，Camel 也解决了一个最为棘手的问题：为了运行测试，你必须要建立 FTP 或 HTTP 服务器。它基本上避免了这样做，因为它可以在运行时修改已有的路由。以下是一个例子：

 
public class BasicTest extends CamelTestSupport {
 
     // This is the route we want to test. Setup with anonymous class for
     // educational purposes, normally this would be a separate class.
     @Override
     protected RouteBuilder createRouteBuilder() throws Exception {
         return new RouteBuilder() {
               @Override
               public void configure() throws Exception {
                      from("<a href="ftp://host/data/inbox">ftp://host/data/inbox</a>").
                             routeId("main").
                                  to("file:data/outbox");
               }
         };
     }
 
     @Override
     public boolean isUseAdviceWith() {
         // Indicates we are using advice with, which allows us to advise the route
         // before Camel is started
         return true;
     }
 
     @Test
     public void TestMe() throws Exception {
         // alter the original route
         context.getRouteDefinition("main").adviceWith(context,
                new AdviceWithRouteBuilder() {
                             @Override
                             public void configure() throws Exception {
                                  replaceFromWith("direct:input");
                                  interceptSendToEndpoint("file:data/outbox")
                                         .skipSendToOriginalEndpoint()
                                                      .to("mock:done");
                             }
                });
         context.start();
 
         // write unit test following AAA (Arrange, Act, Assert)
         String bodyContents = "Hello world";
         MockEndpoint endpoint = getMockEndpoint("mock:done");
         endpoint.expectedMessageCount(1);
         endpoint.expectedBodiesReceived(bodyContents);
 
         template.sendBody("direct:input", bodyContents);
 
         assertMockEndpointsSatisfied();
     }
}

AdviceWithRouteBuilder 允许在 configure 方法中通过代码修改已有的路由，而不必改变已有的代码。在这里，我们使用一个 DIRECT 类型的端点来替换原有的源端点，并确保绕过最初的目的地，而是到达一个 mock 的端点。通过这种方式，为了测试路由，我们不必运行实际的 FTP 服务器，尽管它的编程方式是从 FTP 中获取信息。MockEndpoint 类提供了便利的 API 从而支持以声明式的方式来建立单元测试，这类似于 jMock 。另外一个很好的特性就是在测试的时候，我们可以使用模板来更加简单地往路由上发送信息。

可依赖的 Camel

集成解决方案中有一个很重要的特性，因为它们是连接所有其他系统的桥梁，所以基于它们本身的性质，就会有单点的故障。随着越来越多的系统被连接在了一起以及更为重要的系统数据发生失效，即便总量在增加，数据丢失和性能下降也变得更加难以容忍。

尽管本文是关于 Camel 的，但是解决所有挑战的解决方案超出了 Camel 本身的范围。但是，Camel 会位于这种解决方案的中心，因为它包含转运数据相关的所有逻辑。所以，要知道即便是在这些苛刻的条件下，它依然能够完成其职责。

让我们考虑一个例子来描述这些需求一般是如何得到满足的。在这个例子中，有一个输入的 JMS 队列，消息由外部系统来提供。Camel 的工作就是接受这些消息、做一些处理然后将它们分发到一个输出 JMS 队列。JMS 队列可以进行持久化以满足各自的高可用性，所以我们将会关注 Camel 并假设外部系统“总是”能够将信息放到输入队列中。直到装满为止，如果 Camel 不能足够快地获取和处理信息的时候，这就会发生。

我们的目标就是使 Camel 保持对系统故障的弹性并增加它的性能，通过将其部署到多个服务器上我们做到了这一点，每个服务器运行一个 Camel 实例，这些实例连接到同一个端点之上。如下图所示：

（点击图片放大）

这实际上是另一个 EAI 模式“竞争消费者（Competing Consumers）”的实现。这个模式会有两个好处：首先，来自队列的消息被分发到了多个实例中并且进行并行的处理，这会提高性能。其次，如果有一个服务器坏掉的话，其他会继续运行并接受信息，所以信息处理会自动地进行并不需要任何干预，这会增加故障恢复的能力。

当一个 Camel 实例获取到信息后，其他的实例就无法获取了。这能保证信息只会被处理一次。因为每个服务器都在接受消息，所以工作负载被分布到了多个服务器上：更快的服务器会得到更多的信息从而比更慢的服务器自动承担更多的负担。以这种方式，我们可以在 Camel 实例之间实现必要的协调和工作负载分布。

但是，我们忽视了一件事情：如果一台服务器正在处理信息的时候发生了故障，其他的服务器必须接手它的工作，否则这条信息就丢失了。类似的，如果所有的节点都发生了故障，正在处理中的信息不应该丢失。

如果这种事情发生的话，我们需要事务。借助于事务，JMS 队列会在真正丢弃消息之前，将会等待实例确认获取了信息。如果服务器在处理信息的过程中发生了错误，确认就不会抵达，最终将会进行一个回滚，而信息将会再次出现在队列中，并且剩下的正在运行的服务器依然可以获取它。如果没有运行的服务器了，那么信息将会一直呆在队列中直到有服务器处于在线状态。

对于 Camel 来说，这意味着路由必须是事务性的。Camel 本身并没有提供事务，而是要借助第三方的解决方案。这使得 Camel 比较简单并且能够重用已经得到证明的技术，也使得很容易切换实现成为可能。

作为一个示例，我们将会在 Spring 容器中配置具备事务的 Camel 上下文。注意的是，当我们运行在 Spring 中的时候，更为切实可行的方案是使用 Spring XML 版本的 Camel DSL 而不是 Java 版本的，尽管后者对于起步来说是相当不错的。

当然，在项目的过程中更换 DSL 意味着会有返工，所以很重要的一点就是在合适的时间进行比较明智的迁移。幸好，Spring DSL 也支持单元测试，所以单元测试有助于保证转移是安全的，不管使用的是什么类型的 DSL 路由都能正常工作。

 
<beans //namespace declarations omitted >
 
     //setup connection to jms server
     <jee:jndi-lookup id="jmsConnectionFactory" jndi-name="ConnectionFactory">
         <jee:environment>
            java.naming.factory.initial=org.jnp.interfaces.NamingContextFactory
            java.naming.factory.url.pkgs=org.jboss.naming.client
            java.naming.provider.url=jnp://localhost:1099
         </jee:environment>
     </jee:jndi-lookup>
 
    //configuration for the jms client, including transaction behavior
    <bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
         <property name="connectionFactory" ref="jmsConnectionFactory"/>
         <property name="transactionManager" ref="jmsTransactionManager"/>
         <property name="transacted" value="true"/>
         <property name="acknowledgementModeName" value="TRANSACTED"/>
         <property name="cacheLevelName" value="CACHE_NONE"/>
         <property name="transactionTimeout" value="5"/>
    </bean>
 
   //register camel jms component bean
<bean id="jboss" class="org.apache.camel.component.jms.JmsComponent">
      <property name="configuration" ref="jmsConfig" />
   </bean>
 
   //register spring transactionmanager bean
   <bean id="jmsTransactionManager" 
class="org.springframework.jms.connection.JmsTransactionManager">
         <property name="connectionFactory" ref="jmsConnectionFactory"/>
   </bean>
 
     <camelContext xmlns="http://camel.apache.org/schema/spring">
          <route>
               <from uri="jboss:queue:incoming"/>
               <transacted/>
               <log loggingLevel="INFO" message="processing started." />
               <!-- complex processing -->
               <to uri="jboss:queue:outgoing?exchangePattern=InOnly" />
          </route>
     </camelContext>
</beans>

借助于标签，将路由标识为事务性的，所以 Camel 将会通过对应路由的事务管理器保证资源位于事务之中。一旦在处理中发生错误，事务管理器将会确保事务回滚并且消息重新出现在输入队列中。

但是，并不是每一个路由都能标识为事务性的，因为有些端点，比如 FTP，并不支持事务。幸好，即便没有事务的情况下，Camel 还有错误处理。其中比较有意思的就是 DeadLetterChannel，它是实现了“死文字通道（Dead Letter Channel）”模式的错误处理。这个模式可以描述为有些信息不能够或者不应该被发送到初始预期的目的地，那么这些信息应该放到一个单独的位置，以避免使系统变得混乱。消息系统随后会确定怎样处理这样的信息。

例如，假设发送到指定端点失败了，这个端点可能是FTP 的位置。如果在路由上进行了配置，那么DeadLetterChannel 首先会尝试几次重新发送。如果依然发送失败的话，那么这个信息被称之为“poison”，意味着不能对其做任何有用的处理，它应该被清除出系统。默认情况下，Camel 将会记录日志并丢弃这条信息。很自然的，这种机制也可以进行自定义：例如你可以指定Camel 最多执行三次重新连接的尝试，如果三次之后依然失败就将其存储在JMS 队列中。是的，DeadLetterChannel 可以与事务结合，使两者都发挥最佳作用。

结论

难以管理的集成往往在开始的时候只是简单的集成需求，但是这些需求通过临时方案来进行满足。这种方式不能扩展至更为严苛的需求，使得这样做本身也是耗资巨大的。在早期对特定的EAI 中间件进行大笔的投资是有很高风险的，因为它通常会带来复杂性，很可能会得不偿失。

在本文中，我介绍了第三种选择：借助于Camel，在开始的时候比较简单，同时依然能够满足随后更高的要求。在这方面，我相信Camel 已经展现出了自身的能力：它有很容易的学习曲线，使用和部署起来很轻量级，所以初期的投资会很小。即便是在简单的场景下，学习Camel 比自己动手做的方案也要更快捷。因此，使用Camel 进入EAI 的门槛是很低的。

我还认为，对于更大的需求来讲，Camel 也是很好的选择，此时它会置于集成解决方案之中。在生产效率方面，Camel 支持可扩展性和重用以及对DSL 的集成。鉴于此，在使用Camel 时没有过多的复杂性，所以你可以关注实际的问题。当你发现Camel 内置的功能到达极限无法满足需求时，它有一种对组件和POJO 的插件基础设施，这样你就可以自己来解决问题。

Camel 对单元测试的支持也是非常重要的。Camel 证明自身也可以作为高可用解决方案的一部分。

总体而言，几乎对于任何规模和复杂性的集成来说，Camel 都是很好的可选方案：你可以在开始的时候以很小的前期投资获得比较小规模和简单的功能，同时相信如果集成需求变得更加复杂的话，Camel 依然也能满足。在从这个成熟且完整的集成框架受益的同时，你还能保持高生产率。

关于作者

Frans van der Lek是一个软件工程师，在 web、mobile 和 EAI 解决方案方面经验丰富。目前，他受雇于荷兰的 Capgemini，担任很多项目的设计师、开发人员以及规范制定者。在不编写和思考软件的时候，他会拿上一本书、一杯很棒的咖啡，享受和家人在一起的时光。

原文英文链接： Growing EAI with Apache Camel