Apache Camel源码研究之Policy

快来打我* 2022-10-06 14:59 170阅读 0赞

> 官方接口说明：`A strategy capable of applying interceptors to a processor`.

### 0. 目录 ###

*   *   *  1. 前言
         *  2. 测试用例
         *  3. 分析
         *   *   *  3.1 \`PolicyDefinition\`
         *  4. 扩展
         *  5. Links

### 1. 前言 ###

Apache Camel经过多年的迭代发展，针对不同的场景提供了繁多的扩展以求最大化满足不同的业务需求，本文尝试介绍一类较少被提及的扩展方式 —— `Policy`接口。

借助`Policy`接口提供的扩展，我们可以：

1.  完全控制业务逻辑的走向。得益于Apache Camel优秀的设计，我们可以通过向Apache Camel路由中植入`Policy`实现类来完全把控"之后业务逻辑的走向 —— 就此停止还是继续？"。相较于插入一个自定义`Processor`实现类并通过显式抛出异常来中止流程，使用`Policy`可以让我们更为优雅地中止流程。
2.  添加额外的，统一业务处理逻辑。相较于自定义`Processor`实现类，`Policy`作为官方提供的扩展约定，更容易形成统一的扩展规范来执行。
3.  相较于 `InterceptStrategy`接口扩展，`Policy`扩展所控制的是所配置流程链路上位于policy之后的部分（而 `InterceptStrategy` 为每个`Process`(例如SetBody, SendTo等)都会生效一次），你可以根据自己的业务场景，选择更为合适的扩展。

### 2. 测试用例 ###

先让我们看看本次探究相关的测试用例：

CamelTestUtil.defaultPrepareTest2(new RouteBuilder() { 
    	@Override
    	public void configure() throws Exception { 
    		from("stream:in?promptMessage=Enter something:")//
    		.policy(new PolicyImpl())//
    		.setBody(constant("FuLiZhe"))// 
    		//.policy(new PolicyImpl())// 
    		.to("stream:err");
    	}
    });
    
    // =============================================
    static class PolicyImpl implements Policy { 
    
    	@Override
    	public void beforeWrap(RouteContext routeContext, ProcessorDefinition<?> definition) { 
    		Console.log("beforWrap");
    		Console.log(definition);
    	}
    
    	@Override
    	public Processor wrap(RouteContext routeContext, Processor processor) { 
    		return new PolicyProcess(processor);
    	}
    
    }
    
    // =============================================
    static class PolicyProcess extends DelegateProcessor { 
    
    	public PolicyProcess(Processor processor) { 
    		super(processor);
    	}
    
    	@Override
    	public void process(Exchange exchange) throws Exception { 
    		Console.log(exchange.getIn().getBody(String.class));
    		super.process(exchange);
    	}
    }

### 3. 分析 ###

本次的测试用例依然秉承"简单，够用"原则。执行该测试用例，我们将得到如下堆栈：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70]

以上堆栈图中我们可以看到Policy中声明的扩展接口被一并调用，其中：

1.  `beforeWrap(RouteContext routeContext, ProcessorDefinition<?> definition)`方法给予使用者修改路由定义的机会。
2.  `wrap(RouteContext routeContext, Processor processor)`方法则可以满足用户"在原有路由定义的基础上，加上统一的自定义业务逻辑"的需求。

##### 3.1 `PolicyDefinition` #####

从以上堆栈图中，我们直观感受到的就是对于`PolicyDefinition`类的应用，这里我们将`PolicyDefinition`类中`createProcessor(RouteContext routeContext)`方法贴出来：

@Override
    public Processor createProcessor(RouteContext routeContext) throws Exception { 
    	// 检索出 Policy 实现类
        Policy policy = resolvePolicy(routeContext);
        ObjectHelper.notNull(policy, "policy", this);
    
    	// 给予使用者机会, 修改Route定义
        // before wrap
        policy.beforeWrap(routeContext, this);
    
    	// 本测试用例中, 下面这一步所创建的应该是一个包含了SetBody和SendTo这两个Processor的Pipeline
        // create processor after the before wrap
        Processor childProcessor = this.createChildProcessor(routeContext, true);
    
    	// 给予使用者机会, 封装统一自定义业务逻辑
        // wrap
        Processor target = policy.wrap(routeContext, childProcessor);
    
    	// 预防性编程, 确保纳入生命周期管理
        if (!(target instanceof Service)) { 
            // wrap the target so it becomes a service and we can manage its lifecycle
            target = new WrapProcessor(target, childProcessor);
        }
        return target;
    }

根据以上实现，我们可以推断：`Policy`的设计思路就是将 `policy(Policy policy)` 方法之后定义的路由处理部分作为整体，交给`Policy`实现者进行封装（利用Camel内置提供的`DelegateAsyncProcessor`（示例参见`ShiroSecurityProcessor`）或者`DelegateProcessor`（示例参见`SpringSecurityAuthorizationPolicy.AuthorizeDelegateProcess`），这样`Policy`实现者就有了对于 `policy(Policy policy)` 之后定义的业务流程的完全控制权，进而达到满足用户自定义扩展的需求。

为了验证上面的猜测，我们以本文的测试用例为样板进行如下测试：

1.  `policy(Policy policy)`置于路由定义的起始端，如下：
    
        from("stream:in?promptMessage=Enter something:")//
        		.policy(new PolicyImpl())//
        		.setBody(constant("FuLiZhe"))// 
        		.to("stream:err");
    
    以上最终得到的运行时Processor结构图为：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70 1]  
    结合我们之前在[Apache Camel源码研究之CamelInternalProcessor  
    ][Apache Camel_CamelInternalProcessor]中的研究，以上测试用例中：`.setBody(constant("FuLiZhe")).to("stream:err")` 这两段逻辑被封装为一个Pipeline，然后被以Police扩展实现（也就是我们的`PolicyImpl`）包裹在`DelegateProcessor`实现中（也就是我们的`PolicyProcess`扩展类）对外提供服务。
2.  `policy(Policy policy)`置于路由定义的中段，如下：
    
        from("stream:in?promptMessage=Enter something:")//
        		.setBody(constant("FuLiZhe"))// 
        		.policy(new PolicyImpl())// 
        		.to("stream:err");
    
    以上最终得到的运行时Processor结构图为：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70 2]  
    对比上一个结构图，我们可以看到：
    
    1.  实际的processor类型由DefaultChannel变成了Pipeline。
    2.  Pipeline结构中所包含的processor元素，作为第一个的SetBody没有变化，但第二个的SendTo节点很明显地被我们所扩展的`PolicyProcess`做了Delegate。
    3.  我们可以通过上述测试用例中的`PolicyProcess`类中的`Console.log(exchange.getIn().getBody(String.class));`来验证：  
        a. 对于第一个用例，控制台输出内容为用户在控制台的输入，而非`.setBody(constant("FuLiZhe"))`。  
        b. 对于第二个用例，控制台输出内容为`.setBody(constant("FuLiZhe"))`。

### 4. 扩展 ###

运行时的源码解析之后，接下来让我们看看在Camel中的`Policy`的继承链设计（这也是笔者会写下这篇文章的主要原因）。  
![在这里插入图片描述][20210614165309826.png]  
从以上继承链中：

1.  `AuthorizationPolicy`和`TransactedPolicy`为标志性接口，其并没有定义自己的方法，Camel应该是为了一种语义上的理解和后期扩展的需要进行了这么一个中间层的设计。
2.  `AuthorizationPolicy`接口的实现类直接对应的就是Shiro和Spring Security的鉴权校验，对于这类需求，Policy的设计理念正好满足。

### 5. Links ###

1.  [Apache Camel源码研究之CamelInternalProcessor  
    ][Apache Camel_CamelInternalProcessor]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70]: /images/20221005/d1fccb48bf3c4ee48fc7f52793fb724d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70 1]: /images/20221005/7be1926b59d74578a06ca93e1dce598f.png
[Apache Camel_CamelInternalProcessor]: https://blog.csdn.net/lqzkcx3/article/details/104126561
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xxemtjeDM_size_16_color_FFFFFF_t_70 2]: /images/20221005/83a32ca2b6dd41b3a9deafe5973261bb.png
[20210614165309826.png]: /images/20221005/34e83fee5a7349698ff6bb31774484c9.png