在Mule 4中創建高可靠性應用程序的優秀實踐?
譯文譯者 | 李睿
審校 | 重樓
本文將分享一些在Mule4中創建高可靠應用的優秀實踐。而用戶對可靠性的期望是在Mule應用程序停止或崩潰之后不會丟失消息或數據。
這里分享的大部分配置細節(與可靠性相關)都來自MuleSoft文檔/文章。
1.異步處理—使用持久虛擬機隊列| Anypoint MQ使用外部消息代理(基于JMS)
使用持久虛擬機隊列
當在單個運行時實例模式下運行Mule應用程序時，持久隊列通過序列化并將內容存儲在磁盤上來工作。但是，當在集群運行時實例模式下運行Mule應用程序時，持久隊列會備份在內存網格中。在單個運行時實例模式或集群運行時實例模式下，當使用持久隊列時，發送的數據必須是可序列化的。
CloudHub(1.0)部署的應用程序可以選擇使用CloudHub持久隊列。CloudHub持久隊列是一種云服務，允許將發布到虛擬機（VM）隊列的消息存儲在應用程序的外部。
如何啟用CloudHub持久隊列
CloudHub持久隊列可以在每個應用程序的基礎上啟用，該選項可以在運行時管理器→應用程序→設置頁面中找到。如果用戶擁有這一權限，還可以選擇加密持久隊列以增加安全性。這一功能僅適用于擁有白金及以上訂閱權限的用戶。
使用Anypoint MQ
Anypoint MQ是一個多租戶的云消息傳遞服務，允許客戶在其應用程序之間執行高級異步消息傳遞方案。Anypoint MQ與Anypoint平臺完全集成，提供基于角色的訪問控制、客戶端管理和連接器。
Anypoint MQ確保安全可靠的消息傳遞。自動啟用跨多個數據中心的持久數據存儲，以確保消息隊列架構能夠處理數據中心中斷并具有完全的災難恢復。對消息隊列進行加密以確保數據處于靜止狀態，或將消息發送到死信隊列以提高可靠性。
如何使用Anypoint MQ
安裝和配置Anypoint MQ
- 使用Enterprise Mule憑據登錄Anypoint平臺，然后單擊MQ。
- 單擊“訪問管理”，然后單擊“用戶或角色”，創建Anypoint MQ用戶或角色。
- 從MQ創建隊列、消息交換或FIFO隊列。
- 單擊詳細信息中的隊列或消息交換器名稱，可以訪問消息發送器，將消息發送到隊列或消息交換器，并使用消息瀏覽器從隊列中獲取消息。
- 在MQ中，單擊“客戶端應用程序”注冊應用程序。可以查看該應用程序的客戶端應用ID和客戶端秘密。
在Anypoint Studio中，
- 使用Anypoint Exchange安裝Anypoint MQ連接器。
- 使用所需的構建塊（如HTTP連接器、Anypoint MQ連接器、Set Payload和Logger）創建一個新的Mule項目。
- 配置Anypoint MQ連接器，并提供應用程序的客戶端應用程序ID和客戶端密碼。
- 將Anypoint MQ連接器操作設置為發布或使用消息，或接受（ACK）或不接受（NACK）消息。
使用外部消息代理（基于JMS）
用于JMS(Java消息服務)的Anypoint連接器(JMS連接器)支持向實現JMS規范的任何消息服務的隊列和主題發送和接收消息。
如何配置JMS連接器
配置源：
可以配置以下三個輸入源中的一個來使用JMS連接器：
- JMS>新建消息— 通過偵聽傳入消息來啟動數據流
- HTTP>偵聽器— 每當它在配置的主機和端口上接收到請求時，就啟動數據流
- 調度器— 在滿足基于時間的條件時，啟動數據流
要配置“新建消息”源，執行以下步驟：
- 在Mule Palette視圖中，選擇JMS>新建消息。
- 將新消息拖到Studio畫布上。
- 在“新建消息”配置屏幕中，可選擇更改“顯示名稱”字段的值。
- 單擊連接器配置字段旁邊的加號(+)來配置一個全局元素，該元素可以被應用程序中的所有源實例使用。
在JMS配置窗口中，為“連接”選擇要提供給此配置的連接類型之一：
- 主動MQ連接
- 主動MQ連接— 無連接測試-(已棄用)
- 通用連接
- 在通用選項卡上，指定連接器的連接信息，例如代理所需的庫、JMS規范、緩存策略、身份驗證和連接。
- 在TLS/SSL選項卡上，可以選擇指定TLS配置。
- 在“高級”選項卡上，可以選擇指定重新連接策略和XA連接池。
- 單擊“確定”關閉窗口。
- 在“新建消息”配置屏幕的“目的地”中，指定要從中使用消息的目的地的名稱。
- 在“新建消息”配置界面中配置其他可選字段。
添加連接器
要為JMS連接器添加操作，遵循以下步驟：
- 在Mule Palette視圖中，選擇JMS連接器，然后選擇所需的操作。
- 將操作拖到Studio畫布上，并拖動到輸入源的右側。
2.狀態管理-使用持久對象存儲|使用持久虛擬機隊列|使用外部存儲(DB、FTP等)
使用持久對象存儲
對象存儲是一個存儲鍵值信息的存儲容器。對象存儲可以是持久的或暫時的(非持久的)，在應用程序重啟的情況下，持久性操作系統不會丟失任何信息(鍵值)，而非持久性(鍵值)信息丟失。
如何使用/啟用持久對象存儲
使用默認對象存儲
在默認情況下，每個Mule應用程序都有一個持久的對象存儲，并且總是可以在不需要任何配置的情況下對應用程序可用。數據流可以使用它來持久化和共享數據。
如果希望使用默認的對象存儲，可以為對象存儲指定鍵，而無需為對象存儲操作選擇或創建對象存儲引用，也無需在對象存儲組件的XML元素中指定對象存儲屬性。
Mule應用程序使用運行時管理器部署到CloudHub worker，但是默認對象存儲的內容在應用程序的應用程序數據頁面的運行時管理器中是不可見的。
使用自定義對象存儲
自定義對象存儲必須指定對象存儲屬性。這些對象存儲可以配置為不同于默認對象存儲的行為。例如，可以指出對象存儲是持久的(這樣對象存儲的數據在Mule運行時崩潰時仍然存在)還是瞬態的(在Mule運行時崩潰時數據不存在)。
使用Persistence Gateway
Anypoint Runtime Fabric提供Persistence Gateway。
允許部署到Mule運行時實例的Mule應用程序在應用程序副本和重啟之間存儲和共享數據，從而確保可靠性。
在Anypoint Runtime Fabric中配置了Persistence Gateway后，它就可以用于部署到Mule運行時引擎(4.2.1或更高版本)的Mule應用程序。在配置完成后，用戶可以在使用運行時管理器部署應用程序時選擇“使用持久對象存儲”。
Mule應用程序通過對象存儲連接器使用對象存儲v2 REST API連接到Persistence Gateway。這使得用戶可以同時部署到Anypoint Runtime Fabric和CloudHub，而無需修改Mule應用程序。
如何配置Persistence Gateway
在配置過程中，Persistence Gateway創建所需的數據庫模式。然后，當部署到Runtime Fabric的應用程序被配置為使用持久對象存儲時，Persistence Gateway將必要的行寫入數據庫。
要配置Persistence Gateway，必須創建一個Kubernetes自定義資源，該資源允許集群連接到持久性數據存儲。
Create a Kubernetes Secret
Shell
kubectl create secret generic <SECRET NAME> -n rtf --from-literal=persistence-gateway-creds='postgres://
es://username:pass@host:port/databasename' 為數據存儲創建自定義資源
1.將自定義資源模板從Kubernetes自定義資源模板復制到一個名為custom-resource.yaml的文件中。
2.確保secretRef:name的值與Kubernetes秘密文件中定義的name字段匹配。
3.根據環境的需要修改自定義資源模板的其他字段。
4.運行kubectl apply-f customientresource.yaml。
檢查Persistence Gateway Pod的日志，以確保它可以與數據庫通信
Shell
kubectl get pods -n rtf
尋找具有名稱前綴Persistence Gateway Pod
Shell
kubectl logs -f persistence-gateway-6dfb98949c-7xns9 -nrtf使用持久虛擬機隊列
參考上面的“異步處理”。
使用外部存儲
另一種選擇是將數據持久化在外部存儲系統中，如DB、FTP、外部緩存等。Mule應用程序可以使用連接器連接到這些系統。
不同的外部存儲提供不同的服務質量(QoS)級別，從而確保可靠性：
- 持久性
- 事務
- 復制
- 驅逐策略(最不常用)
- 通過集群實現高可用性
- 通過分區加快數據檢索
- 自動故障切換
3.重新連接策略
當Mule應用程序中的操作無法連接到外部服務器時，默認行為是操作立即失敗并返回連接錯誤。
為了確保不丟失數據，可以通過為該操作配置重連接策略來修改這一默認行為。
如何配置重連接策略
可以通過修改操作屬性或修改操作的全局元素的配置來配置操作的重連接策略。
連接性測試在Mule應用程序啟動時運行，然后在應用程序運行時定期運行。重新連接策略指示當連接失敗時應該做什么。
以下是可用的重連接策略及其行為：
- 從不：默認行為，如果連接嘗試不成功，則立即返回連接錯誤。
- 標準(重新連接)：設置重新連接嘗試的次數以及在返回連接錯誤之前執行這些嘗試的間隔。
- 永遠(永遠重新連接)：嘗試在給定的時間間隔內不斷地重新連接。
XML示例代碼:
XML
<ftp:config name="FTP_Config" doc:name="FTP Config" >
<ftp:connection host="ftp.someftphost.com" port="21" username="myusername" password="mypassword" >
<reconnection failsDeployment="true" >
<reconnect count="5"/>
</reconnection>
</ftp:connection>
</ftp:config>
<flow name="reconnectionsFlow" >
<ftp:listener doc:name="On New or Updated File" config-ref="FTP_Config">
<scheduling-strategy >
<fixed-frequency />
</scheduling-strategy>
</ftp:listener>
</flow>XML
<ftp:connection host="ftp.someftphost.com" port="21" username="myusername" password="mypassword" >
<reconnection>
<reconnect-forever frequency="4000"></reconnect>
</reconnection>
</ftp:connection>在默認情況下，只會記錄一個失敗的連接測試，Mule應用程序無論如何都會啟動或繼續運行而不嘗試重新連接。但是，可以在某些連接器操作上配置重新連接策略，以代替重復嘗試連接。
配置屬性/參數如下:
< reconnection >的屬性
failsDeployment：如果為true，當測試連接失敗時導致部署失敗。默認為false。
<reconnect>的屬性
Blocking：如果為false，重連接策略在一個單獨的非阻塞線程中運行。默認為true。
Frequency：重新連接的頻率(毫秒)。默認為2000。
Count：嘗試重連的次數。默認值為2。
<reconnect-forever>的屬性
Blocking：如果為false，重連接策略在一個單獨的非阻塞線程中運行。默認為true。
Frequency：指定重新連接的頻率(單位：毫秒)。默認為2000。
4.重新傳遞策略
重新傳遞策略（Redelivery Policy）是一個過濾器，通過限制Mule運行時引擎(Mule)執行產生錯誤的消息的次數來幫助節省資源。
當向流的源添加重新傳遞策略時，Mule會在執行流的組件之前評估接收到的數據。如果消息傳遞失敗了指定次數，則重新傳遞策略將阻止流處理接收到的數據，并引發REDELIVERY_EXHAUSTED錯誤。
如何配置重新傳遞策略
在流中的事件源上配置重新發送策略，例如：HTTPListener;關于新建或更新的文件；在JMS連接器的新消息等上指定編號。在引發REDELIVERY_EXHAUSTED錯誤之前，流可以處理由事件源發出的“相同”事件的次數。
以下是配置參數:
- Max Redelivery Count：在引發MULE:Redelivery_EXHAUSTED錯誤之前，可以將消息重新傳遞到流并處理失敗的最大次數。默認值為5。
- 0表示不傳遞
- －1表示無限次重新傳遞
- 是否使用安全哈希：是否使用安全哈希算法來識別重新發送的消息。默認為True。
- 消息摘要算法：用于消息的安全哈希算法。如果消息的有效負載是一個Java對象，Mule將忽略消息摘要算法的值，并返回有效負載的hashCode()返回的值。默認為SHA-256。
- ID表達式：定義一個或多個表達式，用于確定何時重新傳遞消息。只有當“使用安全散列”的值為False時，才能設置此屬性。
- 對象存儲：存儲每條消息的重新傳遞計數器的對象存儲。可以將對象存儲配置為引用或內部元素。
重新傳遞策略如何運作
每次源接收到一個新消息，Mule通過生成它的密鑰來識別該消息。
- 如果處理流導致異常，Mule增加與消息鍵關聯的計數器。當計數器達到的值大于配置的maxRedeliveryCountvalue時，Mule拋出Mule:REDELIVERY_EXHAUSTED錯誤。
- 如果處理流沒有引起異常，則重置其計數器。
5.文件存儲的可重復流策略
Mule4引入了可重復流作為處理流的默認框架。可重復流使用戶能夠：
- 多次讀取流。
- 能夠并發訪問流。
該策略最初使用的內存緩沖區大小為512KB。對于較大的流，該策略在磁盤上創建一個臨時文件來存儲內容，而不會溢出內存。
如果需要處理大文件或小文件，可以改變緩沖區大小(inMemorySize)來優化性能：
- 配置更大的緩沖區大小可以通過避免運行時需要將緩沖區寫入磁盤的次數來提高性能，但它也限制了應用程序可以處理的并發請求的數量。
- 配置較小的緩沖區大小可節省內存負載。
還可以設置緩沖區的度量單位(bufferUnit)。
XML示例代碼：
XML
<file:read path="smallFile.json">
<repeatable-file-store-stream
inMemorySize="10"
bufferUnit="KB"/>
</file:read>6.事務管理
事務是Mule應用程序中的操作，其結果需要保持確定。當流中的一系列步驟必須作為一個單元成功或失敗時，Mule使用事務來劃分該單元。
事務類型
Mule支持單一資源(默認為本地)和擴展架構(XA)事務類型(transactionType)。唯一可以定義事務類型的組件是消息源(例如jms:listener和vm:listener)和Try scope。
單一資源事務(也稱為簡單事務或本地事務)僅使用單一資源發送或接收消息：JMS代理、VM隊列或JDBC連接。
XML示例代碼:
XML
<flow name="asdFlow" doc:id="2a67b1ee-0394-44a8-b6d9-9ce4f94f1ae2" >
<jms:listener config-ref="JMS_Config" destinatinotallow="test.in" transactinotallow="ALWAYS_BEGIN"/>
<!-- Other operations -->
<jms:publish config-ref="JMS_Config" destinatinotallow="test.out" transactinotallow="ALWAYS_JOIN"/>
</flow>Mule只提交成功通過完整流的消息。如果在流中的任何一點上，消息拋出了一個傳播的錯誤（不是由出錯時繼續處理的），Mule將回滾事務。
擴展架構事務(或XA事務)可用于將來自多個事務資源(如VM、JMS或Database)的一系列操作分組到單個可靠的全局事務中。
XA(擴展架構)標準是一個X/Open組標準，它指定了全局事務管理器和本地事務資源管理器之間的接口。XA協議定義了一個兩階段提交協議，可用于跨不同類型的多個服務器可靠地協調和排序一系列原子操作。每個本地XA資源管理器都支持A.C.I.D屬性(原子性、一致性、隔離性和持久性)，這些屬性有助于確保XA資源管理器管理的資源中操作序列的完成。
XML示例代碼：
XML
<flow name="exampleFlow" >
<try transactinotallow="ALWAYS_BEGIN" transactinotallow="XA">
<set-payload value="Hello World"/>
<vm:publish queueName="someVmQueue" config-ref="VM_Config"/>
<jms:consume config-ref="JMS_Config" destinatinotallow="someQueue"/>
<db:insert config-ref="Database_Config">
<db:sql>${insertQuery}</db:sql>
</db:insert>
</try>
<error-handler>
<on-error-propagate enableNotificatinotallow="true" logExceptinotallow="true"/>
</error-handler>
</flow>如果db:insert操作失敗，事務將在錯誤處理程序(on-error-propagate)執行之前回滾(即它不是由on-error-continue處理的)。因此，通過vm:publish發送的消息不會被確認發送，而jms:consume中的消息也不會被實際使用，因此下次可以再次使用它。
下表描述了每種事務類型的特征以及加入事務的操作所需的條件：
Mule4中支持事務的常見連接器操作:
- JMS—發布；消費
- VM—發布；消費
- 數據庫—所有操作
事務操作
事務操作(transactionalAction)定義了操作對事務采取的操作類型。
下表描述了所有可用的事務操作:
如何配置事務
在消息源中，可以從消息源啟動事務。在這種情況下，整個流成為一個事務。
要從消息源發起事務，請配置其事務類型和事務操作：
- 在Anypoint Studio中：打開監聽器的高級選項卡，設置事務類型和事務動作值:
- 在配置XML中：添加transactionaction元素和transactionType元素(如果需要的話)，并設置它們的值:
XML
<?xml versinotallow="1.0" encoding="UTF-8"?>
<mule xmlns:http="http://www.mulesoft.org/schema/mule/http" xmlns:vm="http://www.mulesoft.org/schema/mule/vm"
xmlns="http://www.mulesoft.org/schema/mule/core"
xmlns:doc=http://www.mulesoft.org/schema/mule/documentation xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocatinotallow="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/current/mule-vm.xsd
http://www.mulesoft.org/schema/mule/http http://www.mulesoft.org/schema/mule/http/current/mule-http.xsd">
<vm:config name="VM_Config1" >
<vm:queues >
<vm:queue queueName="input" />
<vm:queue queueName="output" />
</vm:queues>
</vm:config>
<flow name="source-transactionsFlow">
<vm:listener config-ref="VM_Config1" queueName="input" transactinotallow="ALWAYS_BEGIN"/>
<http:request method="GET" url="www.google.com"/>
<vm:publish config-ref="VM_Config1" queueName="output"/>
</flow>
</mule>在Try scope內
Mule流也能夠以非事務性連接器(如HTTP)開始，它需要流中的事務。在這種情況下，可以使用Try scope來設置事務。
可以在Try scope組件中通過設置事務類型和事務操作來設置事務：
?在Anypoint Studio中：打開Try scope的General選項卡，設置事務類型和事務操作值：
?在配置XML中：添加transactionaction元素和transactionType元素(如果需要的話)，并設置它們的值:
XML
<?xml versinotallow="1.0" encoding="UTF-8"?>
<mule xmlns:vm="http://www.mulesoft.org/schema/mule/vm" xmlns:db="http://www.mulesoft.org/schema/mule/db"
xmlns="http://www.mulesoft.org/schema/mule/core"
xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocatinotallow="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/db http://www.mulesoft.org/schema/mule/db/current/mule-db.xsd
http://www.mulesoft.org/schema/mule/vm http://www.mulesoft.org/schema/mule/vm/current/mule-vm.xsd">
<db:config name="Database_Config">
<db:derby-connection database="myDb" create="true" />
</db:config>
<vm:config name="VM_Config">
<vm:queues>
<vm:queue queueName="myQueue" />
</vm:queues>
</vm:config>
<flow name="transactionsFlow">
<try transactinotallow="ALWAYS_BEGIN" transactinotallow="XA">
<db:insert doc:name="Insert" transactinotallow="ALWAYS_JOIN">
<db:sql>
INSERT INTO main_flow_audit (errorType, description) VALUES (:errorType, :description)
</db:sql>
<db:input-parameters><![CDATA[
#[{
'errorType' : 'AUTHENTICATION',
'description' : 'invalid authentication credentials',
}]
]]></db:input-parameters>
</db:insert>
<vm:publish config-ref="VM_Config" queueName="myQueue" transactinotallow="ALWAYS_JOIN"/>
</try>
</flow>
</mule>Bitronix事務管理器
Bitronix是Mule應用程序的XA事務管理器。Bitronix事務管理器允許Mule在重啟時自動恢復中斷的事務。
如何配置Mule應用程序使用Bitronix
要使用Bitronix(在單個應用程序中或在Mule域中的所有應用程序中)，在Mule應用程序中將其聲明為全局配置元素：
XML
<?xml versinotallow="1.0" encoding="UTF-8"?>
<mule xmlns="http://www.mulesoft.org/schema/mule/core"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:bti="http://www.mulesoft.org/schema/mule/ee/bti"
xsi:schemaLocatinotallow="
http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/ee/bti http://www.mulesoft.org/schema/mule/ee/bti/current/mule-bti-ee.xsd">
<bti:transaction-manager/>
...
</mule>可以添加Bitronix
從Studio導入到應用程序或域，執行以下步驟:
- 進入全局元素選項卡。
- 點擊創建按鈕。
- 搜索Bitronix事務管理器。
下表列出了Bitronix的配置屬性：
7.批處理作業
批處理作業組件設計用于對大于內存的數據集進行可靠的異步處理。它自動拆分源數據，并將其存儲到持久隊列中，從而可以處理大型數據集。
該組件可配置如下：
- 過濾要在批處理步驟中處理的記錄：要過濾記錄，批處理步驟支持acceptExpression和acceptPolicy。兩者都是可選的。
- 從批量聚合器執行批量操作：聚合用于將數組中的多條記錄發送到外部服務器。
- 改變記錄塊大小：為了提高性能，Mule運行時以塊的形式隊列和調度批記錄，每個線程最多100條記錄。這種行為減少了I/O請求的數量，提高了操作的負載。
- 在批處理作業實例上設置最大并發限制：最大并發（maxConcurrency）字段限制要同時處理的記錄塊的數量。
8. Until Successful作用域
Until Successful作用域依次執行其中的處理器，直到所有處理器都成功，或者該作用域耗盡了最大重試次數。Until Successful同步運行。如果作用域內的任何處理器未能連接或未能產生成功的結果，則Until Successful將重試其中的所有處理器，包括失敗的處理器，直到所有配置的重試都耗盡。如果重試成功，作用域將繼續到下一個組件。如果最后一次重試不成功，Until Successful將報錯。
如何配置Until Successful作用域
要配置Until Successful作用域，需要在應用程序流中添加<Until-Successful>XML元素。
可以在Until Successful作用域內配置以下屬性：
XML示例代碼:
XML
<!-- FTP Connector config-->
<ftp:config name="FTP_Config" doc:name="FTP Config" >
<ftp:connection workingDir="${ftp.dir}" host="${ftp.host}" />
</ftp:config>
<flow name="untilSuccessfulFlow" >
<!-- Scheduler component to trigger the flow-->
<scheduler doc:name="Scheduler" >
<scheduling-strategy >
<fixed-frequency frequency="15" timeUnit="SECONDS"/>
</scheduling-strategy>
</scheduler>
<!-- Until Successful scope-->
<until-successful maxRetries="5" doc:name="Until Successful" millisBetweenRetries="3000">
<!-- FTP Write operation that executes as part of the Until Successful Scope -->
<ftp:write doc:name="Write" config-ref="FTP_Config" path="/"/>
</until-successful>
<logger level="INFO" doc:name="File upload success" message="File upload success"/>
<!-- Error Handler at flow level-->
<error-handler>
<on-error-continue enableNotificatinotallow="true" logExceptinotallow="true" doc:name="On Error Continue" type="RETRY_EXHAUSTED">
<logger level="INFO" doc:name="File upload failed" message="File upload failed"/>
</on-error-continue>
</error-handler>
</flow>上面的XML示例配置了由調度器組件和執行FTP寫入操作的Until Successful作用域觸發的流。
9. First Successful路由器
First Successful路由器遍歷已配置的處理路由列表，直到其中一條路由成功執行。如果任何處理路由執行失敗(拋出錯誤)，路由器執行下一個配置的路由。如果配置的路由都沒有成功執行，First Successful路由將拋出一個錯誤。
First Successful路由器在成功執行路由后停止執行。
XML示例代碼：
XML
<first-successful doc:name="First Successful" doc:id="6ae009e7-ebe5-47cf-b860-db6d51a31251" >
<route>
<file:read doc:name="Read non existent file" doc:id="199cdb01-cb43-404e-acfd-211fe5a9167e" path="nonExistentFile"/>
<set-variable value="1" doc:name="Set successfulRoute var to route 1" doc:id="c740b39e-a1c4-41d6-8a28-0766ca815ec6" variableName="successfulRoute"/>
</route>
<route>
<set-payload value="#[vars.nonExistentVar!]" doc:name="Set Payload with non existent variable" doc:id="0cc9ac4d-5622-4e10-971c-99073cb58df0" />
<set-variable value="2" doc:name="Set successfulRoute var to route 2" doc:id="88f15c26-d242-4b11-af49-492c35625b84" variableName="successfulRoute" />
</route>
<route>
<set-variable value="3" doc:name="Set successfulRoute var to route 3" doc:id="446afb25-0181-45e5-b04a-68ecb98b57b7" variableName="successfulRoute" />
</route>
<route >
<logger level="INFO" doc:name="Logger" doc:id="b94b905a-3a68-4c88-b753-464bc3d0cfeb" message="This route is never going to be executed"/>
</route>
</first-successful>10.可靠性模式
可靠性模式是一種為應用程序提供可靠消息傳遞的設計，即使應用程序從非事務性連接器接收消息。可靠性模式將可靠的獲取流與應用程序邏輯流耦合在一起。
可靠獲取流(圖的左側)可靠地將消息從沒有實現事務的消息源傳遞到實現事務的連接器的出站操作。操作可以是任何類型的事務端點，比如VM或JMS。如果可靠獲取流不能傳遞消息，它將確保消息不會丟失。
- 對于基于套接字的連接（例如HTTP），這意味著向客戶端返回一個“不成功的請求”響應，以便客戶端可以重試請求。
- 對于基于資源的連接（例如文件或FTP），這意味著不刪除文件，以便可以重新處理。
應用程序邏輯流(圖的右側)將使用事務連接器的消息源中的消息傳遞到應用程序的業務邏輯。
XML示例代碼:
XML
<http:listener-config name="HTTP_Listener_config" doc:name="HTTP Listener config" >
<http:listener-connection host="0.0.0.0" port="8081" />
</http:listener-config>
<vm:config name="VM_Config" doc:name="VM Config" >
<vm:queues >
<vm:queue queueName="toTransactionalVM" queueType="PERSISTENT"/>
</vm:queues>
</vm:config>
<flow name="reliable-data-acquisition">
<http:listener config-ref="HTTP_Listener_config" path="transactionalEndpoint"/>
<vm:publish config-ref="VM_Config" queueName="toTransactionalVM" sendCorrelationId="ALWAYS"/> (1)
</flow>
<!-- This is the application logic flow in the reliability pattern.
It is a wrapper around the sub-flow "business-logic-processing". -->
<flow name="main-flow">
<vm:listener doc:name="Listener" config-ref="VM_Config" queueName="toTransactionalVM"
transactionalAction="ALWAYS_BEGIN"/> (2)
<flow-ref name="business-logic-processing"/>
</flow>
<!-- In this sub-flow, the application starts processing the message. -->
<sub-flow name="business-logic-processing">
<logger level="INFO" doc:name="Logger" />
<!--
This is where the actual business-logic is performed.
-->
</sub-flow>注意事項
在實現可靠性模式時，需要考慮以下幾點：
- 當連接器(消息源)允許時，始終使用事務。
- 當希望在同一個事務中登記多個托管資源時，使用XA事務來橋接消息源。
- JMS的可靠性與MQ實現及其配置方式有關。大多數MQ實現允許配置消息是僅存儲在內存中還是持久化。只有將MQ服務器配置為在轉發消息之前持久存儲消息，才能實現可靠性。否則，可能會在MQ服務器崩潰時丟失消息。
- 可靠性會影響性能。
- 如果可靠獲取流中的出站操作不是事務性的(例如，從文件到FTP的流），請在Try Scope內執行該操作。通過這種實踐，可以確認操作是否成功完成，如果失敗，則記錄錯誤消息。
11.測試
可靠性測試是一項重要的軟件測試技術，由團隊執行，以確保軟件在每種環境條件下以及在指定的時間內始終如一地執行和運行。
該測試包含了功能測試和非功能測試的結果，例如壓力測試、安全測試、功能測試、生產測試等。
可靠性測試的類型
- 特性測試：特性測試的目的是檢查軟件的特性和功能。
- 回歸測試：這是為了檢查應用程序中是否因為修復了以前的錯誤而引入了新的錯誤。這是在每次更改或更新軟件特性和功能之后進行的。
- 負載測試：該測試是為了確認軟件在最高工作量條件下的功能。
12.驗證模式
JSON模塊驗證可用于根據JSON模式驗證JSON。它將顯示JSON有效負載的確切錯誤，因此可以將傳入的JSON錯誤通知客戶端。
類似地，有XML模塊驗證來根據XML模式驗證XML。
因為有適當的驗證，這些模塊可以防止在流中進一步拋出錯誤，從而節省了確保可靠性的額外工作，例如持久化到DB、推入到DLQ等。
13.錯誤處理
應該正確處理上述任何實踐拋出的錯誤，以確保沒有數據丟失。
如何處理錯誤
a. REDELIVERY_EXHAUSTED
當引發錯誤的執行次數大于配置的maxRedeliveryCount值時，從配置Redelivery策略的任何地方拋出。
在這種情況下，在“On Error Continue”作用域上，確保將消息(正在處理的當前消息)傳送/持久化到死信隊列(DLQ)中，這樣它就不會丟失。一旦持久化到DLQ中，任何類型的通知都可以發送到相關團隊。
XML示例代碼：
XML
<on-error-continue type="REDELIVERY_EXHAUSTED">
<vm:publish queuename="errorqueue"/>
</on-error-continue>b. RETRY_EXHAUSTED
當某個執行塊的重試已經耗盡時，從給定操作或從Until Successful作用域拋出。
在這種情況下，在“On Error Continue”作用域內，確保將消息（當前消息正在處理中）推送/持久化到死信隊列（DLQ）中，以使其不會丟失。一旦持久化到DLQ中，就可以向相關團隊發送通知失敗的任何類型的通知。
XML示例代碼：
XML
<on-error-continue type="RETRY_EXHAUSTED">
<vm:publish queuename="errorqueue"/>
</on-error-continue>c.事務中的任何錯誤
當事務期間發生錯誤時，應用程序必須處理錯誤并繼續執行或執行回滾。
On Error Propagate- 如果on-error-propagate錯誤處理程序位于開始事務的組件對應的錯誤處理程序范圍內：在執行on-error-propagate作用域的處理器之前，事務將被回滾。這意味著錯誤處理程序中的處理器不在事務中運行。
- 如果on-error-propagate錯誤處理程序位于沒有啟動事務的元素中：事務不會回滾，on-error-propagate錯誤處理程序中的處理器在事務中運行。
On Error Continue錯誤得到處理，事務保持活動狀態并能夠提交。on-error-continue中的處理器在事務中運行。
在上述任何一種情況下，確保將當前消息推入DLQ或DB等存儲系統。
d.批處理作業錯誤
Mule有三個選項來處理記錄級別的錯誤：
1.Finish processing停止當前作業實例的執行。完成當前正在執行的記錄的執行，但不要從隊列中提取更多記錄，并將作業實例設置為FAILURE狀態。調用OnComplete階段。
2.繼續處理批處理，不考慮任何失敗的記錄，使用acceptExpression和acceptPolicy屬性指導后續批處理步驟如何處理失敗的記錄。
3.繼續處理批處理，不考慮任何失敗的記錄(使用acceptExpression和acceptPolicy屬性指示后續批處理步驟如何處理失敗的記錄)，直到批處理作業積累了失敗記錄的最大數量，此時執行將停止，就像選項1中那樣。
在最后兩種情況下，在ONLY_FAILURES批處理步驟中，將失敗的記錄推入DLQ或DB之類的存儲系統。
e.來自第一個成功路由器的錯誤
當配置的路由都沒有成功執行時，通過將消息推入DLQ或DB之類的存儲系統來處理拋出的錯誤。
結論
這是在一個地方整理各種可靠性問題的解決方案的努力，熱衷于構建高可靠性應用程序的Mule開發人員可以參考一些詳盡的列表。
需要記住的是，其中一些解決方案/實踐的性能可能會受到影響，因此在選擇時需要謹慎。
原文標題：Best Practices for Creating Highly Reliable Applications in Mule 4，作者：Praveen Sundar
