譯者 | 李睿

審校 | 重樓

本文將分享一些在Mule4中創(chuàng)建高可靠應(yīng)用的優(yōu)秀實(shí)踐。而用戶對(duì)可靠性的期望是在Mule應(yīng)用程序停止或崩潰之后不會(huì)丟失消息或數(shù)據(jù)。

這里分享的大部分配置細(xì)節(jié)(與可靠性相關(guān))都來自MuleSoft文檔/文章。

1.異步處理—使用持久虛擬機(jī)隊(duì)列| Anypoint MQ使用外部消息代理(基于JMS)

使用持久虛擬機(jī)隊(duì)列

當(dāng)在單個(gè)運(yùn)行時(shí)實(shí)例模式下運(yùn)行Mule應(yīng)用程序時(shí)，持久隊(duì)列通過序列化并將內(nèi)容存儲(chǔ)在磁盤上來工作。但是，當(dāng)在集群運(yùn)行時(shí)實(shí)例模式下運(yùn)行Mule應(yīng)用程序時(shí)，持久隊(duì)列會(huì)備份在內(nèi)存網(wǎng)格中。在單個(gè)運(yùn)行時(shí)實(shí)例模式或集群運(yùn)行時(shí)實(shí)例模式下，當(dāng)使用持久隊(duì)列時(shí)，發(fā)送的數(shù)據(jù)必須是可序列化的。

CloudHub(1.0)部署的應(yīng)用程序可以選擇使用CloudHub持久隊(duì)列。CloudHub持久隊(duì)列是一種云服務(wù)，允許將發(fā)布到虛擬機(jī)（VM）隊(duì)列的消息存儲(chǔ)在應(yīng)用程序的外部。

如何啟用CloudHub持久隊(duì)列

CloudHub持久隊(duì)列可以在每個(gè)應(yīng)用程序的基礎(chǔ)上啟用，該選項(xiàng)可以在運(yùn)行時(shí)管理器→應(yīng)用程序→設(shè)置頁面中找到。如果用戶擁有這一權(quán)限，還可以選擇加密持久隊(duì)列以增加安全性。這一功能僅適用于擁有白金及以上訂閱權(quán)限的用戶。

使用Anypoint MQ

Anypoint MQ是一個(gè)多租戶的云消息傳遞服務(wù)，允許客戶在其應(yīng)用程序之間執(zhí)行高級(jí)異步消息傳遞方案。Anypoint MQ與Anypoint平臺(tái)完全集成，提供基于角色的訪問控制、客戶端管理和連接器。

Anypoint MQ確保安全可靠的消息傳遞。自動(dòng)啟用跨多個(gè)數(shù)據(jù)中心的持久數(shù)據(jù)存儲(chǔ)，以確保消息隊(duì)列架構(gòu)能夠處理數(shù)據(jù)中心中斷并具有完全的災(zāi)難恢復(fù)。對(duì)消息隊(duì)列進(jìn)行加密以確保數(shù)據(jù)處于靜止?fàn)顟B(tài)，或?qū)⑾l(fā)送到死信隊(duì)列以提高可靠性。

如何使用Anypoint MQ

安裝和配置Anypoint MQ

• 使用Enterprise Mule憑據(jù)登錄Anypoint平臺(tái)，然后單擊MQ。
• 單擊“訪問管理”，然后單擊“用戶或角色”，創(chuàng)建Anypoint MQ用戶或角色。
• 從MQ創(chuàng)建隊(duì)列、消息交換或FIFO隊(duì)列。
• 單擊詳細(xì)信息中的隊(duì)列或消息交換器名稱，可以訪問消息發(fā)送器，將消息發(fā)送到隊(duì)列或消息交換器，并使用消息瀏覽器從隊(duì)列中獲取消息。
• 在MQ中，單擊“客戶端應(yīng)用程序”注冊(cè)應(yīng)用程序?？梢圆榭丛搼?yīng)用程序的客戶端應(yīng)用ID和客戶端秘密。

在Anypoint Studio中，

• 使用Anypoint Exchange安裝Anypoint MQ連接器。
• 使用所需的構(gòu)建塊（如HTTP連接器、Anypoint MQ連接器、Set Payload和Logger）創(chuàng)建一個(gè)新的Mule項(xiàng)目。
• 配置Anypoint MQ連接器，并提供應(yīng)用程序的客戶端應(yīng)用程序ID和客戶端密碼。
• 將Anypoint MQ連接器操作設(shè)置為發(fā)布或使用消息，或接受（ACK）或不接受（NACK）消息。

使用外部消息代理（基于JMS）

用于JMS(Java消息服務(wù))的Anypoint連接器(JMS連接器)支持向?qū)崿F(xiàn)JMS規(guī)范的任何消息服務(wù)的隊(duì)列和主題發(fā)送和接收消息。

如何配置JMS連接器

配置源：

可以配置以下三個(gè)輸入源中的一個(gè)來使用JMS連接器：

1. JMS>新建消息— 通過偵聽傳入消息來啟動(dòng)數(shù)據(jù)流
2.  HTTP>偵聽器— 每當(dāng)它在配置的主機(jī)和端口上接收到請(qǐng)求時(shí)，就啟動(dòng)數(shù)據(jù)流
3. 調(diào)度器— 在滿足基于時(shí)間的條件時(shí)，啟動(dòng)數(shù)據(jù)流

要配置“新建消息”源，執(zhí)行以下步驟：

• 在Mule Palette視圖中，選擇JMS>新建消息。
• 將新消息拖到Studio畫布上。
• 在“新建消息”配置屏幕中，可選擇更改“顯示名稱”字段的值。
• 單擊連接器配置字段旁邊的加號(hào)(+)來配置一個(gè)全局元素，該元素可以被應(yīng)用程序中的所有源實(shí)例使用。

在JMS配置窗口中，為“連接”選擇要提供給此配置的連接類型之一：

• 主動(dòng)MQ連接
• 主動(dòng)MQ連接— 無連接測(cè)試-(已棄用)
• 通用連接
• 在通用選項(xiàng)卡上，指定連接器的連接信息，例如代理所需的庫、JMS規(guī)范、緩存策略、身份驗(yàn)證和連接。
• 在TLS/SSL選項(xiàng)卡上，可以選擇指定TLS配置。
• 在“高級(jí)”選項(xiàng)卡上，可以選擇指定重新連接策略和XA連接池。
• 單擊“確定”關(guān)閉窗口。
• 在“新建消息”配置屏幕的“目的地”中，指定要從中使用消息的目的地的名稱。
• 在“新建消息”配置界面中配置其他可選字段。

添加連接器

要為JMS連接器添加操作，遵循以下步驟：

• 在Mule Palette視圖中，選擇JMS連接器，然后選擇所需的操作。
• 將操作拖到Studio畫布上，并拖動(dòng)到輸入源的右側(cè)。

2.狀態(tài)管理-使用持久對(duì)象存儲(chǔ)|使用持久虛擬機(jī)隊(duì)列|使用外部存儲(chǔ)(DB、FTP等)

使用持久對(duì)象存儲(chǔ)

對(duì)象存儲(chǔ)是一個(gè)存儲(chǔ)鍵值信息的存儲(chǔ)容器。對(duì)象存儲(chǔ)可以是持久的或暫時(shí)的(非持久的)，在應(yīng)用程序重啟的情況下，持久性操作系統(tǒng)不會(huì)丟失任何信息(鍵值)，而非持久性(鍵值)信息丟失。

如何使用/啟用持久對(duì)象存儲(chǔ)

使用默認(rèn)對(duì)象存儲(chǔ)

在默認(rèn)情況下，每個(gè)Mule應(yīng)用程序都有一個(gè)持久的對(duì)象存儲(chǔ)，并且總是可以在不需要任何配置的情況下對(duì)應(yīng)用程序可用。數(shù)據(jù)流可以使用它來持久化和共享數(shù)據(jù)。

如果希望使用默認(rèn)的對(duì)象存儲(chǔ)，可以為對(duì)象存儲(chǔ)指定鍵，而無需為對(duì)象存儲(chǔ)操作選擇或創(chuàng)建對(duì)象存儲(chǔ)引用，也無需在對(duì)象存儲(chǔ)組件的XML元素中指定對(duì)象存儲(chǔ)屬性。

Mule應(yīng)用程序使用運(yùn)行時(shí)管理器部署到CloudHub worker，但是默認(rèn)對(duì)象存儲(chǔ)的內(nèi)容在應(yīng)用程序的應(yīng)用程序數(shù)據(jù)頁面的運(yùn)行時(shí)管理器中是不可見的。

使用自定義對(duì)象存儲(chǔ)

自定義對(duì)象存儲(chǔ)必須指定對(duì)象存儲(chǔ)屬性。這些對(duì)象存儲(chǔ)可以配置為不同于默認(rèn)對(duì)象存儲(chǔ)的行為。例如，可以指出對(duì)象存儲(chǔ)是持久的(這樣對(duì)象存儲(chǔ)的數(shù)據(jù)在Mule運(yùn)行時(shí)崩潰時(shí)仍然存在)還是瞬態(tài)的(在Mule運(yùn)行時(shí)崩潰時(shí)數(shù)據(jù)不存在)。

使用Persistence Gateway

Anypoint Runtime Fabric提供Persistence Gateway。

允許部署到Mule運(yùn)行時(shí)實(shí)例的Mule應(yīng)用程序在應(yīng)用程序副本和重啟之間存儲(chǔ)和共享數(shù)據(jù)，從而確保可靠性。

在Anypoint Runtime Fabric中配置了Persistence Gateway后，它就可以用于部署到Mule運(yùn)行時(shí)引擎(4.2.1或更高版本)的Mule應(yīng)用程序。在配置完成后，用戶可以在使用運(yùn)行時(shí)管理器部署應(yīng)用程序時(shí)選擇“使用持久對(duì)象存儲(chǔ)”。

Mule應(yīng)用程序通過對(duì)象存儲(chǔ)連接器使用對(duì)象存儲(chǔ)v2 REST API連接到Persistence Gateway。這使得用戶可以同時(shí)部署到Anypoint Runtime Fabric和CloudHub，而無需修改Mule應(yīng)用程序。

如何配置Persistence Gateway

在配置過程中，Persistence Gateway創(chuàng)建所需的數(shù)據(jù)庫模式。然后，當(dāng)部署到Runtime Fabric的應(yīng)用程序被配置為使用持久對(duì)象存儲(chǔ)時(shí)，Persistence Gateway將必要的行寫入數(shù)據(jù)庫。

要配置Persistence Gateway，必須創(chuàng)建一個(gè)Kubernetes自定義資源，該資源允許集群連接到持久性數(shù)據(jù)存儲(chǔ)。

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' 

為數(shù)據(jù)存儲(chǔ)創(chuàng)建自定義資源

1.將自定義資源模板從Kubernetes自定義資源模板復(fù)制到一個(gè)名為custom-resource.yaml的文件中。

2.確保secretRef:name的值與Kubernetes秘密文件中定義的name字段匹配。

3.根據(jù)環(huán)境的需要修改自定義資源模板的其他字段。

4.運(yùn)行kubectl apply-f customientresource.yaml。

檢查Persistence Gateway Pod的日志，以確保它可以與數(shù)據(jù)庫通信

Shell 
 kubectl get pods -n rtf
尋找具有名稱前綴Persistence Gateway Pod 
Shell 
 kubectl logs -f persistence-gateway-6dfb98949c-7xns9 -nrtf

使用持久虛擬機(jī)隊(duì)列

參考上面的“異步處理”。

使用外部存儲(chǔ)

另一種選擇是將數(shù)據(jù)持久化在外部存儲(chǔ)系統(tǒng)中，如DB、FTP、外部緩存等。Mule應(yīng)用程序可以使用連接器連接到這些系統(tǒng)。

不同的外部存儲(chǔ)提供不同的服務(wù)質(zhì)量(QoS)級(jí)別，從而確保可靠性：

• 持久性
• 事務(wù)
• 復(fù)制
• 驅(qū)逐策略(最不常用)
• 通過集群實(shí)現(xiàn)高可用性
• 通過分區(qū)加快數(shù)據(jù)檢索
• 自動(dòng)故障切換

3.重新連接策略

當(dāng)Mule應(yīng)用程序中的操作無法連接到外部服務(wù)器時(shí)，默認(rèn)行為是操作立即失敗并返回連接錯(cuò)誤。

為了確保不丟失數(shù)據(jù)，可以通過為該操作配置重連接策略來修改這一默認(rèn)行為。

如何配置重連接策略

可以通過修改操作屬性或修改操作的全局元素的配置來配置操作的重連接策略。

連接性測(cè)試在Mule應(yīng)用程序啟動(dòng)時(shí)運(yùn)行，然后在應(yīng)用程序運(yùn)行時(shí)定期運(yùn)行。重新連接策略指示當(dāng)連接失敗時(shí)應(yīng)該做什么。

以下是可用的重連接策略及其行為：

• 從不：默認(rèn)行為，如果連接嘗試不成功，則立即返回連接錯(cuò)誤。
• 標(biāo)準(zhǔn)(重新連接)：設(shè)置重新連接嘗試的次數(shù)以及在返回連接錯(cuò)誤之前執(zhí)行這些嘗試的間隔。
• 永遠(yuǎn)(永遠(yuǎn)重新連接)：嘗試在給定的時(shí)間間隔內(nèi)不斷地重新連接。

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>

在默認(rèn)情況下，只會(huì)記錄一個(gè)失敗的連接測(cè)試，Mule應(yīng)用程序無論如何都會(huì)啟動(dòng)或繼續(xù)運(yùn)行而不嘗試重新連接。但是，可以在某些連接器操作上配置重新連接策略，以代替重復(fù)嘗試連接。

配置屬性/參數(shù)如下:

< reconnection >的屬性  

failsDeployment：如果為true，當(dāng)測(cè)試連接失敗時(shí)導(dǎo)致部署失敗。默認(rèn)為false。

<reconnect>的屬性

Blocking：如果為false，重連接策略在一個(gè)單獨(dú)的非阻塞線程中運(yùn)行。默認(rèn)為true。

Frequency：重新連接的頻率(毫秒)。默認(rèn)為2000。

Count：嘗試重連的次數(shù)。默認(rèn)值為2。

<reconnect-forever>的屬性

Blocking：如果為false，重連接策略在一個(gè)單獨(dú)的非阻塞線程中運(yùn)行。默認(rèn)為true。

Frequency：指定重新連接的頻率(單位：毫秒)。默認(rèn)為2000。

4.重新傳遞策略

重新傳遞策略（Redelivery Policy）是一個(gè)過濾器，通過限制Mule運(yùn)行時(shí)引擎(Mule)執(zhí)行產(chǎn)生錯(cuò)誤的消息的次數(shù)來幫助節(jié)省資源。

當(dāng)向流的源添加重新傳遞策略時(shí)，Mule會(huì)在執(zhí)行流的組件之前評(píng)估接收到的數(shù)據(jù)。如果消息傳遞失敗了指定次數(shù)，則重新傳遞策略將阻止流處理接收到的數(shù)據(jù)，并引發(fā)REDELIVERY_EXHAUSTED錯(cuò)誤。

如何配置重新傳遞策略

在流中的事件源上配置重新發(fā)送策略，例如：HTTPListener;關(guān)于新建或更新的文件；在JMS連接器的新消息等上指定編號(hào)。在引發(fā)REDELIVERY_EXHAUSTED錯(cuò)誤之前，流可以處理由事件源發(fā)出的“相同”事件的次數(shù)。

以下是配置參數(shù):

• Max Redelivery Count：在引發(fā)MULE:Redelivery_EXHAUSTED錯(cuò)誤之前，可以將消息重新傳遞到流并處理失敗的最大次數(shù)。默認(rèn)值為5。
• 0表示不傳遞
• －1表示無限次重新傳遞
• 是否使用安全哈希：是否使用安全哈希算法來識(shí)別重新發(fā)送的消息。默認(rèn)為True。
• 消息摘要算法：用于消息的安全哈希算法。如果消息的有效負(fù)載是一個(gè)Java對(duì)象，Mule將忽略消息摘要算法的值，并返回有效負(fù)載的hashCode()返回的值。默認(rèn)為SHA-256。
• ID表達(dá)式：定義一個(gè)或多個(gè)表達(dá)式，用于確定何時(shí)重新傳遞消息。只有當(dāng)“使用安全散列”的值為False時(shí)，才能設(shè)置此屬性。
• 對(duì)象存儲(chǔ)：存儲(chǔ)每條消息的重新傳遞計(jì)數(shù)器的對(duì)象存儲(chǔ)?？梢詫?duì)象存儲(chǔ)配置為引用或內(nèi)部元素。

重新傳遞策略如何運(yùn)作

每次源接收到一個(gè)新消息，Mule通過生成它的密鑰來識(shí)別該消息。

1. 如果處理流導(dǎo)致異常，Mule增加與消息鍵關(guān)聯(lián)的計(jì)數(shù)器。當(dāng)計(jì)數(shù)器達(dá)到的值大于配置的maxRedeliveryCountvalue時(shí)，Mule拋出Mule:REDELIVERY_EXHAUSTED錯(cuò)誤。
2. 如果處理流沒有引起異常，則重置其計(jì)數(shù)器。

5.文件存儲(chǔ)的可重復(fù)流策略

Mule4引入了可重復(fù)流作為處理流的默認(rèn)框架?？芍貜?fù)流使用戶能夠：

• 多次讀取流。
• 能夠并發(fā)訪問流。

該策略最初使用的內(nèi)存緩沖區(qū)大小為512KB。對(duì)于較大的流，該策略在磁盤上創(chuàng)建一個(gè)臨時(shí)文件來存儲(chǔ)內(nèi)容，而不會(huì)溢出內(nèi)存。

如果需要處理大文件或小文件，可以改變緩沖區(qū)大小(inMemorySize)來優(yōu)化性能：

• 配置更大的緩沖區(qū)大小可以通過避免運(yùn)行時(shí)需要將緩沖區(qū)寫入磁盤的次數(shù)來提高性能，但它也限制了應(yīng)用程序可以處理的并發(fā)請(qǐng)求的數(shù)量。
• 配置較小的緩沖區(qū)大小可節(jié)省內(nèi)存負(fù)載。

還可以設(shè)置緩沖區(qū)的度量單位(bufferUnit)。

XML示例代碼：

XML 
<file:read path="smallFile.json">
 <repeatable-file-store-stream
 inMemorySize="10"
 bufferUnit="KB"/>
 </file:read>

6.事務(wù)管理

事務(wù)是Mule應(yīng)用程序中的操作，其結(jié)果需要保持確定。當(dāng)流中的一系列步驟必須作為一個(gè)單元成功或失敗時(shí)，Mule使用事務(wù)來劃分該單元。

事務(wù)類型

Mule支持單一資源(默認(rèn)為本地)和擴(kuò)展架構(gòu)(XA)事務(wù)類型(transactionType)。唯一可以定義事務(wù)類型的組件是消息源(例如jms:listener和vm:listener)和Try scope。

單一資源事務(wù)(也稱為簡(jiǎn)單事務(wù)或本地事務(wù))僅使用單一資源發(fā)送或接收消息：JMS代理、VM隊(duì)列或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只提交成功通過完整流的消息。如果在流中的任何一點(diǎn)上，消息拋出了一個(gè)傳播的錯(cuò)誤（不是由出錯(cuò)時(shí)繼續(xù)處理的），Mule將回滾事務(wù)。

擴(kuò)展架構(gòu)事務(wù)(或XA事務(wù))可用于將來自多個(gè)事務(wù)資源(如VM、JMS或Database)的一系列操作分組到單個(gè)可靠的全局事務(wù)中。

XA(擴(kuò)展架構(gòu))標(biāo)準(zhǔn)是一個(gè)X/Open組標(biāo)準(zhǔn)，它指定了全局事務(wù)管理器和本地事務(wù)資源管理器之間的接口。XA協(xié)議定義了一個(gè)兩階段提交協(xié)議，可用于跨不同類型的多個(gè)服務(wù)器可靠地協(xié)調(diào)和排序一系列原子操作。每個(gè)本地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操作失敗，事務(wù)將在錯(cuò)誤處理程序(on-error-propagate)執(zhí)行之前回滾(即它不是由on-error-continue處理的)。因此，通過vm:publish發(fā)送的消息不會(huì)被確認(rèn)發(fā)送，而jms:consume中的消息也不會(huì)被實(shí)際使用，因此下次可以再次使用它。

下表描述了每種事務(wù)類型的特征以及加入事務(wù)的操作所需的條件：

Mule4中支持事務(wù)的常見連接器操作:

• JMS—發(fā)布；消費(fèi)
• VM—發(fā)布；消費(fèi)
• 數(shù)據(jù)庫—所有操作

事務(wù)操作

事務(wù)操作(transactionalAction)定義了操作對(duì)事務(wù)采取的操作類型。

下表描述了所有可用的事務(wù)操作:

如何配置事務(wù)

在消息源中，可以從消息源啟動(dòng)事務(wù)。在這種情況下，整個(gè)流成為一個(gè)事務(wù)。

要從消息源發(fā)起事務(wù)，請(qǐng)配置其事務(wù)類型和事務(wù)操作：

• 在Anypoint Studio中：打開監(jiān)聽器的高級(jí)選項(xiàng)卡，設(shè)置事務(wù)類型和事務(wù)動(dòng)作值:

• 在配置XML中：添加transactionaction元素和transactionType元素(如果需要的話)，并設(shè)置它們的值:

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內(nèi)

Mule流也能夠以非事務(wù)性連接器(如HTTP)開始，它需要流中的事務(wù)。在這種情況下，可以使用Try scope來設(shè)置事務(wù)。

可以在Try scope組件中通過設(shè)置事務(wù)類型和事務(wù)操作來設(shè)置事務(wù)：

?在Anypoint Studio中：打開Try scope的General選項(xiàng)卡，設(shè)置事務(wù)類型和事務(wù)操作值：

?在配置XML中：添加transactionaction元素和transactionType元素(如果需要的話)，并設(shè)置它們的值:

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事務(wù)管理器

Bitronix是Mule應(yīng)用程序的XA事務(wù)管理器。Bitronix事務(wù)管理器允許Mule在重啟時(shí)自動(dòng)恢復(fù)中斷的事務(wù)。

如何配置Mule應(yīng)用程序使用Bitronix

要使用Bitronix(在單個(gè)應(yīng)用程序中或在Mule域中的所有應(yīng)用程序中)，在Mule應(yīng)用程序中將其聲明為全局配置元素：

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導(dǎo)入到應(yīng)用程序或域，執(zhí)行以下步驟:

1. 進(jìn)入全局元素選項(xiàng)卡。
2. 點(diǎn)擊創(chuàng)建按鈕。
3. 搜索Bitronix事務(wù)管理器。

下表列出了Bitronix的配置屬性：

7.批處理作業(yè)

批處理作業(yè)組件設(shè)計(jì)用于對(duì)大于內(nèi)存的數(shù)據(jù)集進(jìn)行可靠的異步處理。它自動(dòng)拆分源數(shù)據(jù)，并將其存儲(chǔ)到持久隊(duì)列中，從而可以處理大型數(shù)據(jù)集。

該組件可配置如下：

• 過濾要在批處理步驟中處理的記錄：要過濾記錄，批處理步驟支持acceptExpression和acceptPolicy。兩者都是可選的。
• 從批量聚合器執(zhí)行批量操作：聚合用于將數(shù)組中的多條記錄發(fā)送到外部服務(wù)器。
• 改變記錄塊大小：為了提高性能，Mule運(yùn)行時(shí)以塊的形式隊(duì)列和調(diào)度批記錄，每個(gè)線程最多100條記錄。這種行為減少了I/O請(qǐng)求的數(shù)量，提高了操作的負(fù)載。
• 在批處理作業(yè)實(shí)例上設(shè)置最大并發(fā)限制：最大并發(fā)（maxConcurrency）字段限制要同時(shí)處理的記錄塊的數(shù)量。

8. Until Successful作用域

Until Successful作用域依次執(zhí)行其中的處理器，直到所有處理器都成功，或者該作用域耗盡了最大重試次數(shù)。Until Successful同步運(yùn)行。如果作用域內(nèi)的任何處理器未能連接或未能產(chǎn)生成功的結(jié)果，則Until Successful將重試其中的所有處理器，包括失敗的處理器，直到所有配置的重試都耗盡。如果重試成功，作用域?qū)⒗^續(xù)到下一個(gè)組件。如果最后一次重試不成功，Until Successful將報(bào)錯(cuò)。

如何配置Until Successful作用域

要配置Until Successful作用域，需要在應(yīng)用程序流中添加<Until-Successful>XML元素。

可以在Until Successful作用域內(nèi)配置以下屬性：

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示例配置了由調(diào)度器組件和執(zhí)行FTP寫入操作的Until Successful作用域觸發(fā)的流。

9. First Successful路由器

First Successful路由器遍歷已配置的處理路由列表，直到其中一條路由成功執(zhí)行。如果任何處理路由執(zhí)行失敗(拋出錯(cuò)誤)，路由器執(zhí)行下一個(gè)配置的路由。如果配置的路由都沒有成功執(zhí)行，First Successful路由將拋出一個(gè)錯(cuò)誤。

First Successful路由器在成功執(zhí)行路由后停止執(zhí)行。

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.可靠性模式

可靠性模式是一種為應(yīng)用程序提供可靠消息傳遞的設(shè)計(jì)，即使應(yīng)用程序從非事務(wù)性連接器接收消息?？煽啃阅Ｊ綄⒖煽康墨@取流與應(yīng)用程序邏輯流耦合在一起。

可靠獲取流(圖的左側(cè))可靠地將消息從沒有實(shí)現(xiàn)事務(wù)的消息源傳遞到實(shí)現(xiàn)事務(wù)的連接器的出站操作。操作可以是任何類型的事務(wù)端點(diǎn)，比如VM或JMS。如果可靠獲取流不能傳遞消息，它將確保消息不會(huì)丟失。

• 對(duì)于基于套接字的連接（例如HTTP），這意味著向客戶端返回一個(gè)“不成功的請(qǐng)求”響應(yīng)，以便客戶端可以重試請(qǐng)求。
• 對(duì)于基于資源的連接（例如文件或FTP），這意味著不刪除文件，以便可以重新處理。

應(yīng)用程序邏輯流(圖的右側(cè))將使用事務(wù)連接器的消息源中的消息傳遞到應(yīng)用程序的業(yè)務(wù)邏輯。

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>

注意事項(xiàng)

在實(shí)現(xiàn)可靠性模式時(shí)，需要考慮以下幾點(diǎn)：

• 當(dāng)連接器(消息源)允許時(shí)，始終使用事務(wù)。
• 當(dāng)希望在同一個(gè)事務(wù)中登記多個(gè)托管資源時(shí)，使用XA事務(wù)來橋接消息源。
• JMS的可靠性與MQ實(shí)現(xiàn)及其配置方式有關(guān)。大多數(shù)MQ實(shí)現(xiàn)允許配置消息是僅存儲(chǔ)在內(nèi)存中還是持久化。只有將MQ服務(wù)器配置為在轉(zhuǎn)發(fā)消息之前持久存儲(chǔ)消息，才能實(shí)現(xiàn)可靠性。否則，可能會(huì)在MQ服務(wù)器崩潰時(shí)丟失消息。
• 可靠性會(huì)影響性能。
• 如果可靠獲取流中的出站操作不是事務(wù)性的(例如，從文件到FTP的流），請(qǐng)?jiān)赥ry Scope內(nèi)執(zhí)行該操作。通過這種實(shí)踐，可以確認(rèn)操作是否成功完成，如果失敗，則記錄錯(cuò)誤消息。

11.測(cè)試

可靠性測(cè)試是一項(xiàng)重要的軟件測(cè)試技術(shù)，由團(tuán)隊(duì)執(zhí)行，以確保軟件在每種環(huán)境條件下以及在指定的時(shí)間內(nèi)始終如一地執(zhí)行和運(yùn)行。

該測(cè)試包含了功能測(cè)試和非功能測(cè)試的結(jié)果，例如壓力測(cè)試、安全測(cè)試、功能測(cè)試、生產(chǎn)測(cè)試等。

可靠性測(cè)試的類型

• 特性測(cè)試：特性測(cè)試的目的是檢查軟件的特性和功能。
• 回歸測(cè)試：這是為了檢查應(yīng)用程序中是否因?yàn)樾迯?fù)了以前的錯(cuò)誤而引入了新的錯(cuò)誤。這是在每次更改或更新軟件特性和功能之后進(jìn)行的。
• 負(fù)載測(cè)試：該測(cè)試是為了確認(rèn)軟件在最高工作量條件下的功能。

12.驗(yàn)證模式

JSON模塊驗(yàn)證可用于根據(jù)JSON模式驗(yàn)證JSON。它將顯示JSON有效負(fù)載的確切錯(cuò)誤，因此可以將傳入的JSON錯(cuò)誤通知客戶端。

類似地，有XML模塊驗(yàn)證來根據(jù)XML模式驗(yàn)證XML。

因?yàn)?/span>有適當(dāng)?shù)尿?yàn)證，這些模塊可以防止在流中進(jìn)一步拋出錯(cuò)誤，從而節(jié)省了確保可靠性的額外工作，例如持久化到DB、推入到DLQ等。

13.錯(cuò)誤處理

應(yīng)該正確處理上述任何實(shí)踐拋出的錯(cuò)誤，以確保沒有數(shù)據(jù)丟失。

如何處理錯(cuò)誤

a. REDELIVERY_EXHAUSTED

當(dāng)引發(fā)錯(cuò)誤的執(zhí)行次數(shù)大于配置的maxRedeliveryCount值時(shí)，從配置Redelivery策略的任何地方拋出。

在這種情況下，在“On Error Continue”作用域上，確保將消息(正在處理的當(dāng)前消息)傳送/持久化到死信隊(duì)列(DLQ)中，這樣它就不會(huì)丟失。一旦持久化到DLQ中，任何類型的通知都可以發(fā)送到相關(guān)團(tuán)隊(duì)。

XML示例代碼：

XML 
<on-error-continue type="REDELIVERY_EXHAUSTED">
 <vm:publish queuename="errorqueue"/>
 </on-error-continue>

b. RETRY_EXHAUSTED

當(dāng)某個(gè)執(zhí)行塊的重試已經(jīng)耗盡時(shí)，從給定操作或從Until Successful作用域拋出。

在這種情況下，在“On Error Continue”作用域內(nèi)，確保將消息（當(dāng)前消息正在處理中）推送/持久化到死信隊(duì)列（DLQ）中，以使其不會(huì)丟失。一旦持久化到DLQ中，就可以向相關(guān)團(tuán)隊(duì)發(fā)送通知失敗的任何類型的通知。

XML示例代碼：

XML 
 <on-error-continue type="RETRY_EXHAUSTED">
 <vm:publish queuename="errorqueue"/>
 </on-error-continue>

c.事務(wù)中的任何錯(cuò)誤

當(dāng)事務(wù)期間發(fā)生錯(cuò)誤時(shí)，應(yīng)用程序必須處理錯(cuò)誤并繼續(xù)執(zhí)行或執(zhí)行回滾。

On Error Propagate

• 如果on-error-propagate錯(cuò)誤處理程序位于開始事務(wù)的組件對(duì)應(yīng)的錯(cuò)誤處理程序范圍內(nèi)：在執(zhí)行on-error-propagate作用域的處理器之前，事務(wù)將被回滾。這意味著錯(cuò)誤處理程序中的處理器不在事務(wù)中運(yùn)行。
• 如果on-error-propagate錯(cuò)誤處理程序位于沒有啟動(dòng)事務(wù)的元素中：事務(wù)不會(huì)回滾，on-error-propagate錯(cuò)誤處理程序中的處理器在事務(wù)中運(yùn)行。

On Error Continue

錯(cuò)誤得到處理，事務(wù)保持活動(dòng)狀態(tài)并能夠提交。on-error-continue中的處理器在事務(wù)中運(yùn)行。

在上述任何一種情況下，確保將當(dāng)前消息推入DLQ或DB等存儲(chǔ)系統(tǒng)。

d.批處理作業(yè)錯(cuò)誤

Mule有三個(gè)選項(xiàng)來處理記錄級(jí)別的錯(cuò)誤：

1.Finish processing停止當(dāng)前作業(yè)實(shí)例的執(zhí)行。完成當(dāng)前正在執(zhí)行的記錄的執(zhí)行，但不要從隊(duì)列中提取更多記錄，并將作業(yè)實(shí)例設(shè)置為FAILURE狀態(tài)。調(diào)用OnComplete階段。

2.繼續(xù)處理批處理，不考慮任何失敗的記錄，使用acceptExpression和acceptPolicy屬性指導(dǎo)后續(xù)批處理步驟如何處理失敗的記錄。

3.繼續(xù)處理批處理，不考慮任何失敗的記錄(使用acceptExpression和acceptPolicy屬性指示后續(xù)批處理步驟如何處理失敗的記錄)，直到批處理作業(yè)積累了失敗記錄的最大數(shù)量，此時(shí)執(zhí)行將停止，就像選項(xiàng)1中那樣。

在最后兩種情況下，在ONLY_FAILURES批處理步驟中，將失敗的記錄推入DLQ或DB之類的存儲(chǔ)系統(tǒng)。

e.來自第一個(gè)成功路由器的錯(cuò)誤

當(dāng)配置的路由都沒有成功執(zhí)行時(shí)，通過將消息推入DLQ或DB之類的存儲(chǔ)系統(tǒng)來處理拋出的錯(cuò)誤。

結(jié)論

這是在一個(gè)地方整理各種可靠性問題的解決方案的努力，熱衷于構(gòu)建高可靠性應(yīng)用程序的Mule開發(fā)人員可以參考一些詳盡的列表。

需要記住的是，其中一些解決方案/實(shí)踐的性能可能會(huì)受到影響，因此在選擇時(shí)需要謹(jǐn)慎。

原文標(biāo)題：Best Practices for Creating Highly Reliable Applications in Mule 4，作者：Praveen Sundar