Apache Camel源码研究之Rest

野性酷女 2022-12-24 15:56 252阅读 0赞

> 本文以Camel2.24.3 + SpringBoot2.x 为基础简单解读Camel中的Rest组件的源码级实现逻辑。

### 0. 目录 ###

*   *   *  1. 前言
         *  2. 源码解读
         *   *  2.1 启动时
             *   *  2.1.1 \`RestComponent.createEndpoint()\`
                 *  2.1.2 \`RestEndpoint.createConsumer()\`
                 *  2.1.3 \`ServletConsumer.start()\`
                 *  2.1.4 \`CamelHttpTransportServlet.init()\`
             *  2.2 运行时
             *  2.3 其它细节
         *  3. 总结
         *  4. Links

### 1. 前言 ###

距离上一次写Apache Camel相关的博客已经过去半年有余，最近因为公司业务需要再次将Camel重新捡起来，深感无奈之之余也只能感慨好事多磨——笔者在两年前就开始不断建言Apache Camel对公司业务场景良好的兼容和支撑，公司领导也是相当认可Camel的技术应用场景，奈何业务紧急程度的不同，导致实际的应用投入凤毛麟角，离预想中的大面积推广相去甚远。

感慨到此为止，本文主要关注的内容正如标题所言，解读Apache Camel的Rest组件在SpringBoot2.x基础框架下是如何运作的（Apache Camel版本为2.24.3。是的，我知道Camel已经3.x很久了，奈何事急从权，Camel3.x变动还是比较大）。

### 2. 源码解读 ###

首先依然是本次的测试用例：

Java端配置

// 继承自 org.apache.camel.builder.RouteBuilder
    @Component
    public class Sample extends RouteBuilder { 
    
    	@Override
    	public void configure() throws Exception { 
    		restConfiguration()//
    		      .host("0.0.0.0").component("servlet");
    	
    		// 
    		from("rest:get,post:helloWorld")//
    	      .routeId("http-service") //
    	      .description("hello world interface") //
    	      .process(new Processor() { 
    		      @Override
    		      public void process(Exchange exchange) throws Exception { 
    			      exchange.getOut().setBody(Collections.singletonMap("name", "fuLiZhe"));
    		      }
    	      	})//
    	      .to("stream:out");

配置文件部分：

################################## 相关配置项
    camel:
      component:
        # ServletComponentConfiguration
        # 相关的AutoConfig配置类: ServletComponentAutoConfiguration
        servlet:
          # 这个enabled配置默认就是 true 
          enabled: true
          # ServletMappingConfiguration ; 
          # 相关的AutoConfig配置类: ServletMappingAutoConfiguration
          mapping:
            # 这个enabled配置默认就是 true
            enabled: true
            servletName: CamelServlet
            context-path: /api/*

访问地址：

http://{ ip}:{ port}/api/helloWorld

样例源码和相关配置项大概就是这么多了，接下来就让我们正式开始源码解读之旅。

为了便于读者理解，这次我们不以SpringBoot的AuthConfig作为出发点，转而直接针对上面给出的用例代码进行解读。

#### 2.1 启动时 ####

按照惯例，我们先看看Apache Camel是如何在启动时候将各个组件装配到位并做好接收外部请求的操作的？

依据[Apache Camel源码研究之启动][Apache Camel]中的解读，我们可以猜到本测试用例一定会涉及到`RestComponent` ，这里我们借势从这个类出发，尝试解读整个流程:

##### 2.1.1 `RestComponent.createEndpoint()` #####

提到Apache Camel里的Component组件，核心方法当然是其`createEndpoint()`了（以下源码因为篇幅，删除了大部分细节）：

// RestComponent.createEndpoint()
    @Override
    protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception { 
    	String restConfigurationName = getAndRemoveParameter(parameters, "componentName", String.class, componentName);
    
    	RestEndpoint answer = new RestEndpoint(uri, this);
    	answer.setComponentName(restConfigurationName);
    	answer.setApiDoc(apiDoc);
        
    	RestConfiguration config = new RestConfiguration();
    	// 这里找到的正是我们上面测试用例中使用 restConfiguration() 注册到的配置项
    	mergeConfigurations(config, findGlobalRestConfiguration());
    	mergeConfigurations(config, getCamelContext().getRestConfiguration(restConfigurationName, true));
    
    	......
    
    	return answer;
    }

所以在本方法中，主要做的事情就是将收集到Rest用户配置注入到创建的`RestEndpoint`实例中。

##### 2.1.2 `RestEndpoint.createConsumer()` #####

根据测试用例，我们最终定位到 `RestEndpoint`的`createConsumer()`方法（如果endpoint URI被应用于`from(...)`, 则Camel 将创建一个`Consumer`；如果endpoint URI被应用于`to(...)`, 则Camel 将创建一个`Producer`）。

@Override
    public Consumer createConsumer(Processor processor) throws Exception { 
    	RestConsumerFactory factory = null;
    	String cname = null;
    	if (getComponentName() != null) { 
    		Object comp = getCamelContext().getRegistry().lookupByName(getComponentName());
    		if (comp instanceof RestConsumerFactory) { 
    			factory = (RestConsumerFactory) comp;
    		} else { 
    		    这一步将直接触发 `ServletComponentAutoConfiguration` 定义的 `configureServletComponent()` 方法的执行，进而向Spring容器中注入一个ServletComponent实例
    			comp = getCamelContext().getComponent(getComponentName());
    			if (comp instanceof RestConsumerFactory) { 
    				factory = (RestConsumerFactory) comp;
    			}
    		}
    
    		......
    		cname = getComponentName();
    	}
    
    	// 略, 其中的逻辑就是尽一切努力找出一个RestConsumerFactory的实现类
    	......
    
    	if (factory != null) { 
    		......
    		// 这个 factory的实际类型就是 ServletComponent, 这一步很明显会导致ServletConsumer实例的创建
    		Consumer consumer = factory.createConsumer(getCamelContext(), processor, getMethod(), getPath(),
    				getUriTemplate(), getConsumes(), getProduces(), config, getParameters());
    		configureConsumer(consumer);
    
    		// add to rest registry so we can keep track of them, we will remove from the registry when the consumer is removed
    		// the rest registry will automatic keep track when the consumer is removed,
    		// and un-register the REST service from the registry
    		getCamelContext().getRestRegistry().addRestService(consumer, url, baseUrl, getPath(), getUriTemplate(), getMethod(),
    				getConsumes(), getProduces(), getInType(), getOutType(), getRouteId(), getDescription());
    		return consumer;
    	} else { 
    	    // 如果始终无法找到RestConsumerFactory实现类，那就抛出异常让程序停止
    		throw new IllegalStateException("Cannot find RestConsumerFactory in Registry or as a Component to use");
    	}
    }

##### 2.1.3 `ServletConsumer.start()` #####

按照我们对于Apache Camel生命周期的了解，上面的`ServletConsumer`实例创建之后，一定会有某个时刻，`ServletConsumer.doStart()`方法将被回调。

经过追踪，我们最终获得如下堆栈：  
![在这里插入图片描述][20201224210712471.png]  
经过以上堆栈的执行链之后，最终将在`ServletComponent.connect(HttpConsumer consumer)`中完成将当前的`ServletConsumer`实例注册到`DefaultHttpRegistry`实例（记住它，本轮生产者/消费者模式实现中的缓冲区就是它了）中。

##### 2.1.4 `CamelHttpTransportServlet.init()` #####

上一小节我们跟踪到了Apache Camel将作为用户自定义的rest请求响应的`ServletConsumer`实例全部注册到了`DefaultHttpRegistry`实例中，关键性的链条还剩下最后一环 —— 在用户请求到达时候如何确保相应的处理函数被调用？（关于这个`CamelHttpTransportServlet`的由来，则是由SpringBoot的AutoConfig来完成自动注入和配置的，具体细节就让我们放到最后的小节）

一番追踪之后，我们最终发现如下方法：

// CamelHttpTransportServlet.init()
    @Override
    public void init(ServletConfig config) throws ServletException { 
    	super.init(config);
    
    	// use rest enabled resolver in case we use rest
    	this.setServletResolveConsumerStrategy(new HttpRestServletResolveConsumerStrategy());
    
    	......
        
    	String name = config.getServletName();
    	String contextPath = config.getServletContext().getContextPath();
    
    	if (httpRegistry == null) { 
    	    // 不出意外地，这个httpRegistry实例和我们在上一小节提醒要记住的`DefaultHttpRegistry`实例为同一个
    		httpRegistry = DefaultHttpRegistry.getHttpRegistry(name);
    		CamelServlet existing = httpRegistry.getCamelServlet(name);
    		
    		......
    		
    		// 所以在这整个过程中, DefaultHttpRegistry 既是作为一个HttpRegistry仓库存在的(其拥有静态字段registries)，也是作为HttpConsumer / CamelServlet 仓库存在的。
    		// 而且观察其 register(HttpConsumer consumer)和register(CamelServlet provider) 方法实现就会发现里内部的 provider 和 consumer 字段相关强关联，这两个 register 不论是其中的哪一个被调用都会将双方进行关联.
    		httpRegistry.register(this);
    	}
    
    	LOG.info("Initialized CamelHttpTransportServlet[name={}, contextPath={}]", getServletName(), contextPath);
    }

#### 2.2 运行时 ####

分析完初始化，接下来让我们看看Apache Camel的Rest组件是如何完成请求的接收，回调对应的自定义处理逻辑，返回结果的？

不出意外地，本文的测试用例之下，接收到用户请求时候，以下方法将被调用：

// CamelServlet.service()
    @Override
    protected void service(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException { 
    	if (isAsync()) { 
    		final AsyncContext context = req.startAsync();
    		//run async
    		context.start(() -> doServiceAsync(context));
    	} else { 
    		doService(req, resp);
    	}
    }
    
    // CamelServlet.doService(req, resp)
    protected void doService(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException { 
    	log.trace("Service: {}", request);
    
        // 找出匹配的ServletConsumer实例 
    	// Is there a consumer registered for the request.
    	HttpConsumer consumer = resolve(request);
    	
    	....... 根据HTTP协议, 一系列防御性编程
    	
    	// create exchange and set data on it
    	Exchange exchange = consumer.getEndpoint().createExchange(ExchangePattern.InOut);
    
    	if (consumer.getEndpoint().isBridgeEndpoint()) { 
    		exchange.setProperty(Exchange.SKIP_GZIP_ENCODING, Boolean.TRUE);
    		exchange.setProperty(Exchange.SKIP_WWW_FORM_URLENCODED, Boolean.TRUE);
    	}
    	if (consumer.getEndpoint().isDisableStreamCache()) { 
    		exchange.setProperty(Exchange.DISABLE_HTTP_STREAM_CACHE, Boolean.TRUE);
    	}
    
    	// we override the classloader before building the HttpMessage just in case the binding
    	// does some class resolution
    	ClassLoader oldTccl = overrideTccl(exchange);
    	HttpHelper.setCharsetFromContentType(request.getContentType(), exchange);
    	exchange.setIn(new HttpMessage(exchange, consumer.getEndpoint(), request, response));
    	// set context path as header
    	String contextPath = consumer.getEndpoint().getPath();
    	exchange.getIn().setHeader("CamelServletContextPath", contextPath);
    
    	String httpPath = (String)exchange.getIn().getHeader(Exchange.HTTP_PATH);
    	// here we just remove the CamelServletContextPath part from the HTTP_PATH
    	if (contextPath != null
    		&& httpPath.startsWith(contextPath)) { 
    		exchange.getIn().setHeader(Exchange.HTTP_PATH,
    				httpPath.substring(contextPath.length()));
    	}
    
    	// we want to handle the UoW
    	try { 
    		consumer.createUoW(exchange);
    	} catch (Exception e) { 
    		log.error("Error processing request", e);
    		throw new ServletException(e);
    	}
    
    	try { 
    		......
    		// process the exchange
    		// 回调用户自定义处理业务逻辑
    		consumer.getProcessor().process(exchange);
    	} catch (Exception e) { 
    		exchange.setException(e);
    	}
    
    	try { 
    		// now lets output to the response
    		......
    		consumer.getBinding().writeResponse(exchange, response);
    	} catch (IOException e) { 
    		log.error("Error processing request", e);
    		throw e;
    	} catch (Exception e) { 
    		log.error("Error processing request", e);
    		throw new ServletException(e);
    	} finally { 
    		consumer.doneUoW(exchange);
    		restoreTccl(exchange, oldTccl);
    	}
    }

#### 2.3 其它细节 ####

1.  `CamelHttpTransportServlet`的由来  
    细节在于SpringBoot的AutoConfig类`ServletMappingAutoConfiguration`中。而且该AutoConfig默认是启用的，所以只需要引入相关JAR，该Bean将自动注入。

### 3. 总结 ###

1.  Apache Camel中的Rest功能，在2.x版本中，首先是借助Apache Camel解析功能，将一系列`from(rest:get,post:xxx)`解析为一一对应的`ServletConsumer`实例，并将这些`ServletConsumer`实例注册进单例工厂`DefaultHttpRegistry`中（不太严谨，但能够方便理解）。
2.  以上包含着一系列`ServletConsumer`实例的单例工厂`DefaultHttpRegistry`实例，将被由SpringBoot自动化配置注入的`CamelHttpTransportServlet`实例在初始化时，注入到`CamelHttpTransportServlet`自身中，进而在自身履行作为Servlet职责时候，解析用户请求，调用相应业务逻辑，返回结果。

### 4. Links ###

1.  [Apache Camel使用之集成SpingBoot Actuator2.0][Apache Camel_SpingBoot Actuator2.0]

[Apache Camel]: https://blog.csdn.net/lqzkcx3/article/details/103263042
[20201224210712471.png]: /images/20221120/69bcb8bf15e349cbabaab6d2ae70db78.png
[Apache Camel_SpingBoot Actuator2.0]: https://blog.csdn.net/lqzkcx3/article/details/105894988