Apache Camel 簡介

null-safe 運算符算

Apache Camel中的null安全導航操作

在Apache Camel的Simple語言中，處理可能為null的屬性或嵌套屬性時，有幾種方法可以避免空指針異常：

使用空安全操作符（?.）
Apache Camel的Simple語言支持null安全操作符（?.），可以防止在訪問嵌套屬性時出現空指針異常：

${exchangeProperty.allData?[${exchangeProperty.loopCount}]?.['業務二級產出線ID']?[0]?.['number']}

這種語法確保當任何中間值為null時，表達式會安全地返回null而不是拋出異常。
2. 使用空值合并操作符（??）
當你想在遇到null值時提供默認值時，可以使用null合并操作符（??）：

${exchangeProperty.allData?[${exchangeProperty.loopCount}]?.['業務二級產出線ID']?[0]?.['number'] ?? '默認值'}

這樣如果表達式計算結果為null，則會返回指定的默認值。
3. 使用isEmpty或exists函數
在某些情況下，你可能想先檢查一個值是否存在，然后再嘗試訪問它的屬性：

${header.foo} == null
${header.foo} != null

使用條件表達式
可以使用iif函數來實現條件邏輯：

${iif(exchangeProperty.allData?[${exchangeProperty.loopCount}] != null, exchangeProperty.allData?[${exchangeProperty.loopCount}]?.['業務二級產出線ID']?[0]?.['number'], '默認值')}

處理Map和List類型
對于Map或List類型的數據，Camel支持索引和鍵訪問，同樣可以使用null安全操作符：

${body.address?.lines?[0]}
${header.foo?[bar]}

最佳實踐

盡量使用空安全操作符：在訪問嵌套屬性時，始終使用?.操作符來避免空指針異常。

提供默認值：在可能的情況下，使用??操作符為可能為null的表達式提供默認值。

簡化復雜表達式：對于非常復雜的表達式，可以考慮將其分解為多個步驟，或使用Bean組件在Java代碼中處理復雜的null處理邏輯。

使用變量存儲中間結果：對于復雜的表達式，可以使用變量存儲中間結果，然后在后續步驟中使用這些變量。

在你的具體案例中，可以這樣修改表達式：

${exchangeProperty.allData?[${exchangeProperty.loopCount}]?.['業務二級產出線ID']?[0]?.['number'] ?? ''}

這樣當任何嵌套屬性為null時，表達式會安全地返回空字符串而不是拋出異常。

setVariable、setProperty、setHeader

setVariable

變量范圍：變量（Variable）是當前交換（Exchange）的一部分，但與Header和Property不同，它不會被復制到消息中

生命周期：變量的生命周期僅限于當前Exchange的處理過程

訪問方式：通過${variable.xxx}訪問

用途：適合作為路由內部使用的臨時數據，不會影響到消息本身

setProperty（或setExchangeProperty）

屬性范圍：屬性（Property）是附加到Exchange上的，在整個消息處理過程中都可用

生命周期：屬性在整個Exchange生命周期內持續存在，即使跨多個路由和端點

訪問方式：通過${exchangeProperty.xxx}訪問

用途：適合存儲需要在整個消息處理流程中使用的數據，如跨路由共享的上下文信息

setHeader

作用范圍：Header是消息（Message）級別的，附加在消息上而不是Exchange上

生命周期：Header隨消息一起傳遞，當消息被傳遞到另一個端點時，Header通常也會被傳遞（取決于端點實現）

訪問方式：通過${header.xxx}訪問

端點交互：許多組件（如HTTP、JMS、Kafka等）會讀取特定的Header來控制行為

重要特性：當使用EIP（如split、multicast）時，新創建的Exchange會從原始Exchange中復制Header

特性	setHeader	setVariable	setProperty
作用范圍	消息級別	Exchange級別，但不復制	Exchange級別
生命周期	隨消息傳遞，可能被端點消費	僅當前Exchange處理過程	整個Exchange生命周期
是否傳遞給目標系統	是（通常）	否	否
跨路由可見性	可能會變化（消息轉換時）	僅在當前Exchange中可見	在整個Exchange中可見

當前時間

${date:now} 如果不指定格式模式，$ {date:now}會返回一個默認格式的日期時間字符串。
默認情況下，Apache Camel會使用java.util.Date對象的toString()方法的格式，即類似于：
Wed Apr 24 10:15:30 CST 2024

${date:now:yyyy-MM-dd}
可以根據需要調整日期時間格式。
例如，如果只需要日期部分，可以使用yyyy-MM-dd格式。

Simple語言中的時間偏移支持的單位：
h：小時
m：分鐘
s：秒

注意事項：
Simple語言的時間偏移語法使用+和-操作符表示時間的增減
可以組合多個時間單位，如now+1h30m（當前時間加1小時30分鐘） ${date:now+1h30m} 在路由中使用時，需要將表達式放在$ {}內部

Apache Camel onException 完整使用指南

概述

在Apache Camel中，onException、onWhen、retryWhile和redeliveryPolicy可以組合使用來實現復雜的異常處理邏輯。本文檔詳細說明如何同時使用這些功能。

核心組件說明

1. onException

作用: 定義異常處理器，捕獲特定類型的異常
語法: <onException><exception>異常類型</exception></onException>

2. onWhen

作用: 條件判斷，決定是否處理特定的異常
語法: <onWhen><simple>條件表達式</simple></onWhen>

3. retryWhile

作用: 定義重試條件，決定是否繼續重試
語法: <retryWhile><simple>重試條件</simple></retryWhile>

4. redeliveryPolicy

作用: 定義重試策略，包括重試次數、延遲時間等
語法: <redeliveryPolicy>重試策略配置</redeliveryPolicy>

組合使用示例

基本語法結構

<onException><!-- 1. 定義異常類型 --><exception>java.lang.Exception</exception><!-- 2. 使用onWhen條件判斷 --><onWhen><simple>${exception.message} contains 'retryable'</simple></onWhen><!-- 3. 使用retryWhile定義重試條件 --><retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 3</simple></retryWhile><!-- 4. 應用重試策略 --><redeliveryPolicy><maximumRedeliveries>3</maximumRedeliveries><redeliveryDelay>1000</redeliveryDelay><backOffMultiplier>2.0</backOffMultiplier><useExponentialBackOff>true</useExponentialBackOff></redeliveryPolicy><!-- 5. 異常處理邏輯 --><setHeader headerName="ErrorType"><simple>${exception.class.simpleName}</simple></setHeader><handled><constant>true</constant></handled>
</onException>

詳細配置說明

1. 異常類型捕獲

<!-- 捕獲單個異常類型 -->
<exception>java.lang.RuntimeException</exception><!-- 捕獲多個異常類型 -->
<exception>java.net.ConnectException</exception>
<exception>java.net.SocketTimeoutException</exception>
<exception>java.net.UnknownHostException</exception>

2. onWhen條件判斷

<!-- 基于異常消息內容判斷 -->
<onWhen><simple>${exception.message} contains 'retryable'</simple>
</onWhen><!-- 基于異常類型判斷 -->
<onWhen><simple>${exception.class.simpleName} == 'BusinessException'</simple>
</onWhen><!-- 復合條件判斷 -->
<onWhen><simple>${exception.message} contains 'business' or ${exception.message} contains 'validation'</simple>
</onWhen><!-- 總是處理 -->
<onWhen><constant>true</constant>
</onWhen>

3. retryWhile重試條件

<!-- 基于重試次數判斷 -->
<retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 3</simple>
</retryWhile><!-- 基于異常消息和重試次數判斷 -->
<retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 5 and ${exception.message} contains 'business'</simple>
</retryWhile><!-- 基于自定義邏輯判斷 -->
<retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 10 and ${exception.message} contains 'connection'</simple>
</retryWhile>

4. redeliveryPolicy重試策略

<redeliveryPolicy><!-- 最大重試次數 --><maximumRedeliveries>3</maximumRedeliveries><!-- 重試延遲時間（毫秒） --><redeliveryDelay>1000</redeliveryDelay><!-- 退避倍數 --><backOffMultiplier>2.0</backOffMultiplier><!-- 使用指數退避 --><useExponentialBackOff>true</useExponentialBackOff><!-- 日志配置 --><logRetryAttempted>true</logRetryAttempted><logRetryStackTrace>true</logRetryStackTrace><logHandled>true</logHandled><logNewException>true</logNewException><logExhausted>true</logExhausted><logExhaustedMessageHistory>true</logExhaustedMessageHistory>
</redeliveryPolicy>

實際應用場景

1. 全局異常處理

<onException><exception>java.lang.Exception</exception><onWhen><simple>${exception.message} contains 'retryable' or ${exception.message} contains 'temporary'</simple></onWhen><retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 3 and (${exception.message} contains 'retryable' or ${exception.message} contains 'temporary')</simple></retryWhile><redeliveryPolicy><maximumRedeliveries>3</maximumRedeliveries><redeliveryDelay>1000</redeliveryDelay><backOffMultiplier>2.0</backOffMultiplier><useExponentialBackOff>true</useExponentialBackOff></redeliveryPolicy><handled><constant>true</constant></handled>
</onException>

2. 業務異常處理

<onException><exception>java.lang.RuntimeException</exception><onWhen><simple>${exception.message} contains 'business' or ${exception.message} contains 'validation'</simple></onWhen><redeliveryPolicy><maximumRedeliveries>5</maximumRedeliveries><redeliveryDelay>2000</redeliveryDelay><backOffMultiplier>1.5</backOffMultiplier><useExponentialBackOff>true</useExponentialBackOff></redeliveryPolicy><retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 5 and ${exception.message} contains 'business'</simple></retryWhile><to uri="activemq:dead.letter.queue"/><handled><constant>true</constant></handled>
</onException>

3. 網絡異常處理

<onException><exception>java.net.ConnectException</exception><exception>java.net.SocketTimeoutException</exception><exception>java.net.UnknownHostException</exception><onWhen><constant>true</constant></onWhen><redeliveryPolicy><maximumRedeliveries>10</maximumRedeliveries><redeliveryDelay>5000</redeliveryDelay><backOffMultiplier>2.0</backOffMultiplier><useExponentialBackOff>true</useExponentialBackOff></redeliveryPolicy><retryWhile><simple>${header.CamelRedeliveryCounter} &lt; 10</simple></retryWhile><handled><constant>true</constant></handled>
</onException>

重要變量說明

異常相關變量

${exception} - 當前異常對象
${exception.message} - 異常消息
${exception.class.simpleName} - 異常類名

重試相關變量

${header.CamelRedeliveryCounter} - 當前重試次數
${header.CamelRedeliveryMaxCounter} - 最大重試次數
${header.CamelFailureEndpoint} - 失敗的端點

消息相關變量

${body} - 消息體
${headers} - 消息頭
${exchangeId} - 交換ID

最佳實踐

1. 異常處理順序

將具體的異常處理放在前面
將通用的異常處理放在后面

2. 重試策略設計

根據異常類型設置不同的重試策略
使用指數退避避免系統過載
設置合理的最大重試次數

3. 日志記錄

啟用重試日志記錄
記錄異常堆棧信息
記錄重試歷史

4. 死信隊列

對于無法重試的異常，發送到死信隊列
實現死信隊列處理邏輯

測試示例

<route id="testRoute"><from uri="direct:test"/><choice><when><simple>${body} == 'retryable-error'</simple><throwException exceptionType="java.lang.RuntimeException" message="模擬可重試的臨時故障"/></when><when><simple>${body} == 'business-error'</simple><throwException exceptionType="java.lang.RuntimeException" message="模擬業務異常"/></when><when><simple>${body} == 'network-error'</simple><throwException exceptionType="java.net.ConnectException" message="模擬網絡連接異常"/></when><otherwise><setBody><simple>處理成功: ${body}</simple></setBody></otherwise></choice><to uri="mock:result"/>
</route>

總結

通過組合使用onException、onWhen、retryWhile和redeliveryPolicy，可以實現：

精確的異常捕獲 - 通過onWhen條件判斷
靈活的重試邏輯 - 通過retryWhile自定義重試條件
可控的重試策略 - 通過redeliveryPolicy配置重試參數
完善的異常處理 - 通過異常處理器處理捕獲的異常

這種組合使用方式能夠滿足復雜業務場景下的異常處理需求，提高系統的可靠性和穩定性。

onWhen的作用機制

onWhen=true時
異常會被onException處理器捕獲
會執行redeliveryPolicy定義的重試策略
會執行retryWhile定義的重試條件判斷
會執行異常處理邏輯（如setHeader、process等）
onWhen=false時
異常不會被onException處理器捕獲
不會執行重試邏輯
不會執行異常處理邏輯
異常會繼續向上傳播，可能被其他異常處理器捕獲或導致路由失敗

maximumRedeliveries 和 retryWhile

maximumRedeliveries=0 - 無論retryWhile如何設置，都不會重試
retryWhile=false - 無論maximumRedeliveries如何設置，都不會重試
優先級關系 - 兩個條件是"與"的關系，不是"或"的關系
兩個條件都必須滿足 - 只有maximumRedeliveries>0且retryWhile=true時才會重試

重試邏輯的優先級

Apache Camel的重試邏輯遵循以下優先級：

onWhen - 首先判斷是否處理此異常
maximumRedeliveries - 檢查是否允許重試
retryWhile - 檢查是否滿足重試條件

這種設計確保了重試行為的可控性和可預測性，避免了意外的重試行為。

Apache Camel異常處理器注冊機制：

處理器注冊時機：onException處理器在路由定義時就會注冊到異常處理鏈中，與它們在路由中的位置無關
匹配順序：Camel會按照onException在路由中定義的順序進行檢查
第一個匹配的處理器生效：找到匹配的處理器后立即執行并停止查找

注意：onException 不能被嵌套，否則會報錯類似如下：

The output must be added as top-level on the route. Try moving OnException[[java.lang.Exception] When[simple{ 
${exchangeProperty.mysql_custom_exception}} -> []] -> [process[ref:errorLoggingProcessor]]] to the top of route

解決辦法：將所有的異常單獨拿出來，緊挨著 from 放到下面，多個異常的，選放業務自定義異常并使用 onWhen 進行控制，全局異常放到所有異常的最后進行兜底，xml示例如下：

<route xmlns="http://camel.apache.org/schema/spring" id="VjCFgbqRdYh4Dc7tSKAzY"><from uri="direct:VjCFgbqRdYh4Dc7tSKAzY"/><!-- 自定義異常 --><onException><exception>java.lang.Exception</exception><onWhen><simple resultType="java.lang.Boolean">${exchangeProperty.VjCFgbqRdYh4Dc7tSKAzYhttp_client_1_custom_exception}</simple></onWhen><log message="===&gt; routeId：VjCFgbqRdYh4Dc7tSKAzY，nodeId：http_client_1，logMessage：執行自定義異常處理"/><process ref="errorLoggingProcessor"/><log message="===&gt; routeId：VjCFgbqRdYh4Dc7tSKAzY，nodeId：http_client_1，logMessage：忽略異常，繼續執行"/><continued><constant>true</constant></continued></onException><!-- 全局兜底異常 --><onException><exception>java.lang.Exception</exception><log message="===&gt; routeId：VjCFgbqRdYh4Dc7tSKAzY，nodeId：VjCFgbqRdYh4Dc7tSKAzY，logMessage：執行全局兜底異常處理"/><process ref="errorLoggingProcessor"/><log message="===&gt; routeId：VjCFgbqRdYh4Dc7tSKAzY，nodeId：VjCFgbqRdYh4Dc7tSKAzY，logMessage：發生異常，終止執行"/><handled><constant>true</constant></handled></onException><!-- 其他業務邏輯 --><log message="===&gt; routeId：VjCFgbqRdYh4Dc7tSKAzY，nodeId：end，logMessage：工作流執行結束"/><onCompletion><process ref="completionProcessor"/></onCompletion>
</route>

Apache Camel onCompletion 使用說明

概述

onCompletion是Apache Camel中用于處理路由完成事件的機制，類似于onException，但處理的是路由正常完成或異常完成后的清理工作。

onCompletion vs onException 對比

相似點

注冊機制：兩者都在路由定義時注冊到處理鏈中
位置無關性：在路由中的定義位置不影響其生效
支持條件判斷：都可以使用onWhen條件
處理器引用：都可以使用<process ref="xxx">引用處理器

不同點

特性	onException	onCompletion
觸發時機	異常發生時	路由完成時（成功或失敗）
處理對象	異常對象	完成事件
重試機制	支持重試策略	不支持重試
條件類型	異常類型匹配	完成狀態匹配
使用場景	異常處理	清理工作、日志記錄

onCompletion 基本語法

<onCompletion><process ref="completionProcessor"/>
</onCompletion><!-- 帶條件的onCompletion -->
<onCompletion><onWhen><simple>${exchangeProperty.CamelRouteStop} == true</simple></onWhen><process ref="stopCompletionProcessor"/>
</onCompletion><!-- 成功完成時的處理 -->
<onCompletion><onCompleteOnly><process ref="successCompletionProcessor"/></onCompleteOnly>
</onCompletion><!-- 失敗完成時的處理 -->
<onCompletion><onFailureOnly><process ref="failureCompletionProcessor"/></onFailureOnly>
</onCompletion>

位置影響分析

關鍵結論：位置不影響功能

與onException類似，onCompletion在路由中的位置不會影響其功能，因為：

注冊時機：在路由啟動時就已經注冊到完成處理鏈中
執行時機：在路由完成時觸發，與定義位置無關
作用范圍：對整個路由生效

示例驗證

<route id="testRoute"><from uri="direct:start"/><!-- 業務邏輯 --><log message="開始處理"/><process ref="businessProcessor"/><!-- onCompletion定義在中間 --><onCompletion><process ref="completionProcessor"/></onCompletion><!-- 更多業務邏輯 --><to uri="mock:result"/><!-- 路由結束<!-- 即使onCompletion定義在中間，仍然會在路由完成時執行 -->
</route>

常用配置模式

1. 基本完成處理

<onCompletion><log message="路由執行完成: ${exchangeId}"/><process ref="cleanupProcessor"/>
</onCompletion>

2. 條件完成處理

<onCompletion><onWhen><simple>${header.ProcessType} == 'batch'</simple></onWhen><log message="批處理完成，執行清理"/><process ref="batchCleanupProcessor"/>
</onCompletion>

3. 成功/失敗分別處理

<!-- 成功完成處理 -->
<onCompletion><onCompleteOnly><log message="路由執行成功"/><setHeader headerName="CompletionStatus"><constant>SUCCESS</constant></setHeader><process ref="successProcessor"/></onCompleteOnly>
</onCompletion><!-- 失敗完成處理 -->
<onCompletion><onFailureOnly><log message="路由執行失敗"/><setHeader headerName="CompletionStatus"><constant>FAILURE</constant></setHeader><process ref="failureProcessor"/></onFailureOnly>
</onCompletion>

4. 資源清理

<onCompletion><log message="開始清理資源"/><!-- 清理臨時文件 --><setHeader headerName="TempFile"><simple>${exchangeProperty.tempFile}</simple></setHeader><process ref="fileCleanupProcessor"/><!-- 清理數據庫連接 --><process ref="dbCleanupProcessor"/><!-- 記錄完成時間 --><setHeader headerName="CompletionTime"><simple>${date:now:yyyy-MM-dd HH:mm:ss}</simple></setHeader><log message="資源清理完成"/>
</onCompletion>

實際應用示例

1. 工作流完成處理

<route id="workflowRoute"><from uri="direct:workflow"/><onCompletion><process ref="workflowCompletionProcessor"/></onCompletion><!-- 工作流邏輯 --><process ref="workflowProcessor"/><to uri="mock:result"/>
</route>

2. 批處理完成處理

<route id="batchProcessRoute"><from uri="direct:batch"/><onCompletion><onWhen><simple>${header.BatchSize} > 1000</simple></onWhen><log message="大批量處理完成，執行特殊清理"/><process ref="largeBatchCleanupProcessor"/></onCompletion><!-- 批處理邏輯 --><process ref="batchProcessor"/><to uri="mock:result"/>
</route>

3. 事務完成處理

<route id="transactionRoute"><from uri="direct:transaction"/><onCompletion><onCompleteOnly><log message="事務提交成功"/><process ref="commitProcessor"/></onCompleteOnly></onCompletion><onCompletion><onFailureOnly><log message="事務回滾"/><process ref="rollbackProcessor"/></onFailureOnly></onCompletion><!-- 事務邏輯 --><transacted ref="transactionPolicy"/><process ref="transactionProcessor"/><to uri="mock:result"/>
</route>

注意事項

1. 執行順序

多個onCompletion處理器按定義順序執行
與onException不同，onCompletion不會停止后續處理器的執行

2. 異常處理

onCompletion中的異常不會觸發onException處理器
建議在onCompletion中使用try-catch處理異常

3. 性能考慮

onCompletion會在每次路由完成時執行
避免在onCompletion中執行耗時的操作

4. 資源管理

確保在onCompletion中正確清理資源
避免資源泄漏

總結

onCompletion是Apache Camel中重要的完成處理機制，與onException類似，在路由中的位置不影響其功能。它主要用于：

資源清理：清理臨時文件、數據庫連接等
日志記錄：記錄路由執行結果
狀態更新：更新處理狀態
通知機制：發送完成通知

通過合理使用onCompletion，可以確保路由執行后的清理工作得到正確處理，提高系統的穩定性和可維護性。

特殊符號問題 > 、< 和 &

這三個符號會導致xml解析錯誤，需要進行轉義

換行符 \r 和 \n 問題

如果接口入參或者http請求出參中有換行符，并且使用 simple 設置到引擎內，會導致引擎解析字符串為對象時報錯（使用 constant 設置到引擎不會報錯），需要對\r 和 \n 進行處理

在body出參中：自定義Processor，示例代碼如下：

public class MyBodyTransProcessor implements Processor {private final DefaultCamelContext camelContext;public MyBodyTransProcessor(DefaultCamelContext camelContext) {this.camelContext = camelContext;}@Overridepublic void process(Exchange exchange) throws Exception {String routeId = exchange.getFromRouteId();String body = null;try {body = exchange.getIn().getBody(String.class);} catch (Exception e) {log.error("==>routeId：{}，get body error",routeId, e);}String realBody = "{}";if (StringUtils.isNotBlank(body)) {realBody = body.replace("\\n", "\\\\n");realBody = realBody.replace("\\r", "\\\\r");}log.info("==>routeId：{}，oldBody：{}，realBody：{}",routeId, body, realBody);exchange.getIn().setBody(realBody, String.class);}
}

在用戶設置的入參中：在 from 下面設置兩個setProperty，使用

<setProperty name="_nnn_">
<constant>\n</constant>
</setProperty>
<setProperty name="_rrr_">
<constant>\r</constant>
</setProperty>

然后在set到引擎前進行替換

        real = real.replace("\\n", "${exchangeProperty._nnn_"}");real = real.replace("\\r", "${exchangeProperty._rrr_"}");

開發流程

方式一

如果對 Apache Camel 有一定的了解，熟悉 Apache Camel 相關的 xml 語法，可以使用此方式進行開發。

（非必需）在 processor 中定義相關的處理器，只處理核心的邏輯，并在 CamelConfig 中注冊此處理器。
在 parse 中對前端傳過來的json進行解析，拼接成 Apache Camel 的 xml 代碼，如果 xml 不能直接完成需求，則引用步驟1中定義的處理器。

方式二

如果對 Apache Camel 了解程度有限或需要快速完成需求，可以使用此方式進行開發。

在 processor 中定義相關的處理器，處理所有的邏輯，并在 CamelConfig 中注冊此處理器。
在 parse 中對前端傳過來的json進行解析，只進行核心的校驗，并引用步驟1中定義的處理器。