当前置Processor失败时跳过后续Camel Processor的最佳实践

针对你提出的需求——业务Processor优雅失败（不抛异常）时跳过后续业务节点，但必须执行consolidateResponse收尾逻辑，最佳实现方式是通过Exchange属性标记处理状态，结合Camel的路由控制组件（choice/when）实现分支流程，具体方案如下：

核心思路 #

1. 状态标记：每个业务Processor内部，在优雅失败时（如服务调用返回错误、业务校验不通过），向Exchange设置自定义属性（如PROCESSING_FAILED）为true；执行成功时设置为false或清除该属性。
2. 即时判断：每个业务Processor执行后，立即检查状态属性，若标记为失败则直接调用consolidateResponse并终止后续业务流程。
3. 异常兜底：保留原有的onException逻辑，同时在异常处理中添加consolidateResponse，确保意外抛出的异常也能执行收尾操作。

修改后的路由配置 #

<camel:routeContext id="myRouteRouteContext">
    <camel:route id="my-route-route">
    
        <camel:from uri="{{camel.uri.myRoute}}" />
        
        <!-- 执行processor1并检查状态 -->
        <camel:process ref="{{bean.processor.processor1}}" />
        <camel:choice>
            <camel:when>
                <camel:simple>${exchangeProperty.PROCESSING_FAILED} == true</camel:simple>
                <camel:process ref="{{bean.processor.consolidateResponse}}" />
                <camel:stop /> <!-- 终止后续业务Processor执行 -->
            </camel:when>
        </camel:choice>
        
        <!-- 执行processor2并检查状态 -->
        <camel:process ref="{{bean.processor.processor21}}" />
        <camel:choice>
            <camel:when>
                <camel:simple>${exchangeProperty.PROCESSING_FAILED} == true</camel:simple>
                <camel:process ref="{{bean.processor.consolidateResponse}}" />
                <camel:stop />
            </camel:when>
        </camel:choice>
        
        <!-- 执行processor3并检查状态 -->
        <camel:process ref="{{bean.processor.processor3}}" />
        <camel:choice>
            <camel:when>
                <camel:simple>${exchangeProperty.PROCESSING_FAILED} == true</camel:simple>
                <camel:process ref="{{bean.processor.consolidateResponse}}" />
                <camel:stop />
            </camel:when>
        </camel:choice>
        
        <!-- 执行processor4并检查状态 -->
        <camel:process ref="{{bean.processor.processor4}}" />
        <camel:choice>
            <camel:when>
                <camel:simple>${exchangeProperty.PROCESSING_FAILED} == true</camel:simple>
                <camel:process ref="{{bean.processor.consolidateResponse}}" />
                <camel:stop />
            </camel:when>
        </camel:choice>
        
        <!-- 执行processor5并检查状态 -->
        <camel:process ref="{{bean.processor.processor5}}" />
        <camel:choice>
            <camel:when>
                <camel:simple>${exchangeProperty.PROCESSING_FAILED} == true</camel:simple>
                <camel:process ref="{{bean.processor.consolidateResponse}}" />
                <camel:stop />
            </camel:when>
        </camel:choice>
        
        <!-- 所有业务Processor执行成功，执行收尾逻辑 -->
        <camel:process ref="{{bean.processor.consolidateResponse}}" />

        <!-- 异常处理：针对意外抛出的异常场景，确保收尾逻辑执行 -->
        <camel:onException>
            <camel:exception>java.lang.Exception</camel:exception>
            <camel:handled>
                <camel:constant>true</camel:constant>
            </camel:handled>
            <camel:process ref="{{bean.processor.consolidateResponse}}" />
            <camel:to uri="{{camel.uri.error}}" />
        </camel:onException>

    </camel:route>
</camel:routeContext>

关键细节说明 #

• 状态属性设置：在每个业务Processor的Java代码中，需要添加状态标记逻辑：

  // 示例：processor1中的处理逻辑
  @Override
  public void process(Exchange exchange) throws Exception {
      boolean success = callService(); // 业务调用逻辑
      if (!success) {
          exchange.setProperty("PROCESSING_FAILED", Boolean.TRUE);
      } else {
          exchange.setProperty("PROCESSING_FAILED", Boolean.FALSE);
      }
  }
  

• <camel:stop />的作用：终止路由的后续执行，确保失败后不会继续调用剩余的业务Processor。
• 异常场景兼容：原有的onException块中添加consolidateResponse，保证即使Processor意外抛出异常，收尾逻辑也能执行，避免数据不完整。

优化方案（针对Processor较多的场景） #

如果业务Processor数量较多，重复的choice代码会显得冗余，可以将状态检查逻辑封装为自定义Predicate，或者使用intercept拦截器统一处理：

<!-- 定义自定义Predicate -->
<camel:predicate id="processingFailedPredicate">
    <camel:simple>${exchangeProperty.PROCESSING_FAILED} == true</camel:simple>
</camel:predicate>

<!-- 在路由中复用Predicate -->
<camel:process ref="{{bean.processor.processor1}}" />
<camel:choice>
    <camel:when predicateRef="processingFailedPredicate">
        <camel:process ref="{{bean.processor.consolidateResponse}}" />
        <camel:stop />
    </camel:when>
</camel:choice>

这种方式可以减少代码重复，同时保持流程的可读性。

内容的提问来源于stack exchange，提问作者Ethan Miller