Apache Camel源码研究之DataFormat

article/2025/1/17 3:10:42

在上一篇博客Apache Camel源码研究之TypeConverter中，我们介绍了Apache Camel实现数据格式转换的一种实现方式，本文中我们将介绍另外一种实现方式 —— DataFormat。

1. 概述

相较于前面博客介绍过的TypeConverter，DataFormat在平时应用中应该更容易被研发人员所感知，毕竟其所支持的json，yaml，xml等转换现在已经成为事实的数据传输格式。本文接下来的部分我们将就其实现原理以及如何进行自定义扩展作出一些笔者自己的理解。

2. 源码解读

首先是本次的测试用例：

@Test
public void marshal() throws Exception {// CamelTestUtil.defaultPrepareTest2(new RouteBuilder() {@Overridepublic void configure() throws Exception {from("stream:in?promptMessage=Enter something:")//.setBody(constant(Collections.singletonList("123")))//.marshal().json(JsonLibrary.Gson) // 将对象转换为JSON字符串//.setBody(constant("{'name':'LQ'}"))////.unmarshal().json(JsonLibrary.Gson)//	将JSON字符串转换为对象			.to("stream:out");}});
}

接下来，我们首先探究的是Camel是如何将这段转换逻辑插入到Route的执行流程中的：

跟随上述Route中定义的.marshal().json(JsonLibrary.Gson)可以看出 Apache Camel是将一个 MarshalDefinition实例添加到RouteDefinition中，注意该MarshalDefinition类型是间接继承自ProcessorDefinition<Type>类的，按照我们之前在博客Apache Camel源码研究之ProcessorDefinition中讨论的内容，则可以很轻松得找到如下内容：

// MarshalDefinition.createProcessor()
@Override
public Processor createProcessor(RouteContext routeContext) {// 专门的类和静态方法来根据用户配置创建相应的DataFormat实例// 这个过程牵扯到的细节较多，我们将在下面以专门的小节进行讨论DataFormat dataFormat = DataFormatDefinition.getDataFormat(routeContext, getDataFormatType(), ref);// 将DataFormat实例以Processor的形式介入到Apache Camel的执行逻辑链条中// === 这里多说一句的是, 与DataFormat类似的是 Camel中的 Expression 概念也是以类似的方式介入到Camel的执行逻辑链条中的，更具体的我们放到下一篇博客中return new MarshalProcessor(dataFormat);
}

接下来让我们看看 MarshalProcessor有哪些需要关注的：

// MarshalProcessor
@Override
protected void doStart() throws Exception {// inject CamelContext on data formatif (dataFormat instanceof CamelContextAware) {((CamelContextAware) dataFormat).setCamelContext(camelContext);}// add dataFormat as service which will also start the service// (false => we and handling the lifecycle of the dataFormat)getCamelContext().addService(dataFormat, false);
}@Override
protected void doStop() throws Exception {ServiceHelper.stopService(dataFormat);getCamelContext().removeService(dataFormat);
}

以上可以看到正是因为DataFormat被Processor封装，在获取并入到Apache Camel执行逻辑链条权利的同时，也通过MarshalProcessor覆写自基类的doStart，doStop方法享受到了共享生命周期的好处。DataFormat可以借此做一些自定义的初始化和销毁操作。

在了解完初始化时的逻辑之后，执行时候的逻辑也就一目了然了：DataFormat接口所定义的marshal，unmarshal两个方法，在执行时序上依然是借助Apache Camel中的灵魂组件Processor 来实现的。通过在用户指定的时机采用同样的方式回调相应的Processor组件，来将原始数据格式转换为目标数据类型，这种组件化的实现方式隐藏了实现细节，增加了系统的弹性。感兴趣的读者可以参见 Apache Camel源码研究之启动。

3. DataFormat扩展机制

在了解了DataFormat的基本载入和执行机制之后，我们来关注一点更加深入的内容 —— Apache Camel是如何实现DataFormat的可扩展性的。相比较之下，org.apache.camel.model.dataformat PACKAGE下DataFormatDefinition的子类，数量远远大于org.apache.camel.impl PACKAGE下DataFormat接口的实现类。而且Camel作为一个久负盛名，被多款商业级ESB平台作为核心组件的框架，将所有的实现类和相应依赖硬编码到自身的core JAR中肯定是不可能的。

经过一番探究，我们找到如下关键类和配置文件：

DataFormatDefinition
- 其构造函数约定了 dataFormatName，Camel将以此为标识符从容器中或者相应的配置目录下查找相应的DataFormat实现类。
- createDataFormat()，作为一个protected修饰级别的方法，绝大部分子类正是借助这个方法的覆写来实现自身DataFormat实现类的构造的。
- getDataFormat(RouteContext routeContext)。正是这个方法回掉了上面的createDataFormat()方法，在保证对外接口统一稳定的前提下，实现了强大的可扩展性。
DefaultDataFormatResolver ，以DataFormatResolver接口形式直接在DefaultCamelContext中实例化。内部定义了查找目录路径"META-INF/services/org/apache/camel/dataformat/"，Apache Camel将该路径加上前面提到的dataFormatName指示的文件名，从中读取到DataFormat接口实现类，
最终堆栈图如下：

camel-core下内置的DataFormat。

4. 自定义扩展

下面让我们来尝试一个自定义DataFormat：

public static class EbcdicDataFormat implements DataFormat {// US EBCDIC 037 code pageprivate String codepage = "CP037";public EbcdicDataFormat() {}public EbcdicDataFormat(String codepage) {this.codepage = codepage;}@Overridepublic void marshal(Exchange exchange, Object graph, OutputStream stream) throws Exception {final String str = ExchangeHelper.convertToMandatoryType(exchange, String.class, graph);stream.write(str.getBytes(codepage));}@Overridepublic Object unmarshal(Exchange exchange, InputStream stream) throws Exception {final byte[] bytes = ExchangeHelper.convertToMandatoryType(exchange, byte[].class,stream);return new String(bytes, codepage);}
}

相应的测试用例：

@Test
public void custom() throws Exception {// CamelTestUtil.defaultPrepareTest2(new RouteBuilder() {@Overridepublic void configure() throws Exception {DataFormat df = new EbcdicDataFormat("CP037");from("stream:in?promptMessage=Enter something:")//// 相较于TypeConverter, DataFormat的使用就必须显式调用了.marshal(df) //.to("stream:err");}});
}

5. DataFormat实现

最后让我们看看DataFormat接口的定义，以构建一个全局的认识和理解：

public interface DataFormat {/*** Marshals the object to the given Stream.*/void marshal(Exchange exchange, Object graph, OutputStream stream) throws Exception;/*** Unmarshals the given stream into an object.*/Object unmarshal(Exchange exchange, InputStream stream) throws Exception;
}