使用循序群組傳送相互關聯的訊息 - Azure Logic Apps

發行項
01/04/2024

適用於：Azure Logic Apps (使用量)

當您需要以特定順序傳送相互關聯的訊息時，可以透過使用 Azure 服務匯流排連接器，在使用 Azure Logic Apps 時，遵循循序群組模式。相互關聯的訊息具有屬性，可定義這些訊息之間的關聯性，例如服務匯流排中的工作階段識別碼。

例如，假設您有一個名為「工作階段 1」的工作階段有 10 個訊息，而您有 5 個訊息要用於名為「工作階段 2」的工作階段，這些訊息全都傳送至相同的服務匯流排佇列。您可以建立邏輯應用程式來處理來自佇列的訊息，讓來自「工作階段 1」的所有訊息都由單一觸發程序執行來處理，而來自「工作階段 2」的所有訊息均會由下一個觸發程序執行來處理。

General sequential convoy pattern

本文說明如何使用使用服務匯流排工作階段以依序傳遞建立相互關聯範本，來建立可實作此模式的邏輯應用程式。此範本會定義邏輯應用程式工作流程，並從服務匯流排連接器的佇列中收到訊息時 (查看鎖定)觸發程序開始，此觸發程序會接收來自服務匯流排佇列的訊息。以下是此邏輯應用程式執行的高階步驟：

根據觸發程序從服務匯流排佇列讀取的訊息初始化工作階段。
在目前的工作流程執行期間，讀取和處理來自佇列中相同工作階段的所有訊息。

若要檢閱此範本的 JSON 檔案，請參閱 GitHub：service-bus-sessions.json。

如需詳細資訊，請參閱循序群組模式 - Azure 架構雲端設計模式。

必要條件

Azure 訂用帳戶。如果您沒有訂用帳戶，請註冊一個免費的 Azure 帳戶。
服務匯流排命名空間和服務匯流排佇列，這是您將在邏輯應用程式中使用的傳訊實體。這些項目和您的邏輯應用程式需要使用相同的 Azure 訂閱。請確定您在建立佇列時選取 [啟用工作階段]。如果您沒有這些項目，請了解如何建立服務匯流排命名空間和佇列。

重要

當您同時選取具有相同連接器類型的觸發程序和動作，並使用它們來處理相同實體時，例如傳訊佇列或主題訂閱時，請多加注意。此組合會建立無限迴圈，這會導致永遠不會結束的邏輯應用程式。
如何建立邏輯應用程式的基本知識。如果您不熟悉 Azure Logic Apps，請試試快速入門，在多租用戶 Azure Logic Apps 中建立範例使用量邏輯應用程式工作流程。

檢查服務匯流排命名空間的存取

如果您不確定您的邏輯應用程式是否具有存取服務匯流排命名空間的權限，請確認這些存取權限。

登入 Azure 入口網站。選取服務匯流排的 [命名空間]。
在命名空間選單的 [設定] 下方，選取 [共用存取原則]。在 [宣告] 之下，確認您有該命名空間的 [管理] 權限。
現在取得服務匯流排命名空間的連接字串。當您從邏輯應用程式建立命名空間的連線時，可於稍後使用此字串。
1. 在 [共用存取原則] 窗格的 [原則]下方，選取 [RootManageSharedAccessKey]。
2. 選取您主要連接字串旁邊的 [複製] 按鈕。儲存連接字串以供稍後使用。
提示

若要確認您的連接字串，是否與您的服務匯流排命名空間或傳訊實體 (例如佇列) 相關聯，請搜尋連接字串中是否有 EntityPath 參數。如果您發現這個參數，表示此連接字串適用於特定實體，但不是可用於您邏輯應用程式的正確字串。

建立邏輯應用程式

在本節中，您會以使用服務匯流排工作階段以依序傳遞建立相互關聯範本來建立邏輯應用程式，其中包含實作此工作流程模式的觸發程序和動作。您也可以建立與服務匯流排命名空間的連線，並指定您想要使用的服務匯流排佇列名稱。

在 Azure 入口網站中，建立空白的邏輯應用程式。在 Azure 首頁中，選取 [建立資源]>[整合]>[邏輯應用程式]。
範本資源庫出現之後，捲動超過影片和常見的觸發程序區段。從 [範本] 區段中，選取 [使用服務匯流排工作階段以依序傳遞建立相互關聯] 範本。
當確認方塊出現時，請選取 [使用此範本]。
在邏輯應用程式設計工具的 [服務匯流排] 圖形中，選取[繼續]，然後選取出現在圖形中的加號 (+)。
現在選擇任一選項來建立服務匯流排連線：
- 若要使用您稍早從服務匯流排命名空間複製的連接字串，請遵循下列步驟：
  1. 選取 [手動輸入連線資訊]。
  2. 針對 [連線名稱]，提供連線的名稱。針對 [連接字串]，貼上命名空間連接字串，然後選取 [建立]，例如：
    
    提示
    
    如果您沒有此連接字串，請了解如何尋找並複製服務匯流排命名空間連接字串。
- 若要從您目前的 Azure 訂閱中選取服務匯流排命名空間，請遵循下列步驟：
  1. 針對 [連線名稱]，提供連線的名稱。針對 [服務匯流排命名空間]，選取服務匯流排命名空間，例如：
  2. 當下一個窗格出現時，請選取您的服務匯流排原則，然後選取 [建立]。
當您完成時，選取 [繼續]。

邏輯應用程式設計工具現在會顯示 [使用服務匯流排工作階段以依序傳遞建立相互關聯] 範本，其中包含預先填入的工作流程以及觸發程序和動作，包括實作遵循 Try-Catch 模式錯誤處理的兩個範圍。

現在您可以深入了解範本中的觸發程序和動作，或直接跳到提供邏輯應用程式範本的值。

範本摘要

以下是折疊詳細資料時，使用服務匯流排工作階段以依序傳遞建立相互關聯範本中的最上層工作流程：

Template's top-level workflow

名稱	描述
`When a message is received in a queue (peek-lock)`	根據指定的週期，此服務匯流排觸發程序會檢查指定的服務匯流排佇列是否有任何訊息。如果佇列中存有訊息，就會引發觸發程序，其會建立和執行工作流程執行個體。「查看鎖定」一詞表示觸發程序會傳送要求，以從佇列擷取訊息。如果訊息存在，觸發程序會擷取並鎖定訊息，以便在鎖定期間到期之前，不會針對訊息進行任何其他處理。如需詳細資訊，請初始化工作階段。
`Init isDone`	此初始化變數動作會建立設定為 `false` 的布林值變數，並指出下列條件何時為 true： - 工作階段中沒有其他訊息可供讀取。 - 工作階段鎖定不再需要更新，因此目前的工作流程執行個體可以完成。如需詳細資訊，請參閱初始化工作階段。
`Try`	此範圍動作包含用來處理訊息的動作。如果 `Try` 範圍內發生問題，後續 `Catch` 範圍動作會處理該問題。如需詳細資訊，請參閱 "Try" 範圍。
`Catch`	此範圍動作包含在上述 `Try` 範圍發生問題時執行的動作。如需詳細資訊，請參閱 "Catch" 範圍。

"Try" 範圍

以下是摺疊詳細資料時，Try 範圍動作中的頂層流程：

名稱	描述
`Send initial message to topic`	您可以將此動作取代為您想要在佇列中的任何動作，用於處理來自工作階段的第一則訊息。工作階段識別碼會指定工作階段。在此範本中，服務匯流排動作會將第一則訊息傳送至服務匯流排主題。如需詳細資訊，請參閱處理初始訊息。
(平行分支)	此平行分支動作會建立兩個路徑： - 分支 #1：繼續處理訊息。如需詳細資訊，請參閱分支 #1：在佇列中完成初始訊息。 - 分支 #2：如果發生問題則放棄訊息，並釋出以供另一個觸發程序執行收取。如需詳細資訊，請參閱分支 #2：放棄來自佇列的初始訊息。這兩個路徑稍後會聯結關閉佇列中的工作階段並成功動作，如下一列所述。
`Close a session in a queue and succeed`	此服務匯流排動作會聯結先前所述的分支，並在下列其中一個事件發生之後關閉佇列中的工作階段： - 工作流程完成佇列中可用訊息的處理。 - 工作流程會放棄初始訊息，因為發生錯誤。如需詳細資訊，請參閱關閉佇列中的工作階段並成功。

分支 #1：完成佇列中的初始訊息

名稱	描述
`Complete initial message in queue`	此服務匯流排動作會將已成功擷取的訊息標示為完成，並從佇列中移除訊息以防止重新處理。如需詳細資訊，請參閱處理初始訊息。
`While there are more messages for the session in the queue`	這個 Until 迴圈會繼續在訊息存在時，或是一小時過後取得訊息。如需此迴圈動作的詳細資訊，請參閱當佇列中的工作階段有更多訊息時。
`Set isDone = true`	當沒有任何訊息存在時，此設定變數動作會將 `isDone` 設定為 `true`。
`Renew session lock until cancelled`	此 Until 迴圈可確保此邏輯應用程式會在有訊息存在時，或是一小時過後保留工作階段鎖定。如需此迴圈動作的詳細資訊，請參閱更新工作階段鎖定直到取消。

分支 #2：放棄來自佇列的初始訊息

如果處理第一個訊息的動作失敗，則服務匯流排動作，即放棄來自佇列的初始訊息，會將訊息釋放給另一個工作流程執行個體執行以進行收取和處理。如需詳細資訊，請參閱處理初始訊息。

"Catch" 範圍

如果 Try 範圍中的動作失敗，邏輯應用程式仍須關閉工作階段。當 Try 範圍動作導致 Failed、Skipped 或 TimedOut 狀態時，Catch 範圍動作就會執行。此範圍會傳回錯誤訊息，其中包含發生問題的工作階段識別碼，並會終止邏輯應用程式。

以下是摺疊詳細資料時，Catch 範圍動作中的頂層流程：

名稱	描述
`Close a session in a queue and fail`	此服務匯流排動作會關閉佇列中的工作階段，讓工作階段鎖定不會保持開啟狀態。如需詳細資訊，請參閱關閉佇列中的工作階段並失敗。
`Find failure msg from 'Try' block`	此篩選陣列動作會根據指定的準則，在 `Try` 範圍內建立來自所有動作之輸入和輸出的陣列。在此情況下，此動作會傳回導致 `Failed` 狀態之動作的輸出。如需詳細資訊，請參閱從 'Try' 區塊尋找失敗訊息。
`Select error details`	此選取動作會根據指定的準則，建立包含 JSON 物件的陣列。這些 JSON 物件是從上一個動作，即 `Find failure msg from 'Try' block` 所建立陣列中的值而建置。在此情況下，此動作會傳回陣列，其中包含從上一個動作傳回之錯誤詳細資料所建立的 JSON 物件。如需詳細資訊，請參閱選取錯誤詳細資料。
`Terminate`	此終止動作會停止工作流程的執行、取消進行中的任何動作、略過任何剩餘動作，並傳回指定的狀態、工作階段識別碼，以及 `Select error details` 動作所產生的錯誤結果。如需詳細資訊，請參閱終止邏輯應用程式。

完成範本

若要為 [使用服務匯流排工作階段以依序傳遞建立相互關聯] 範本中的觸發程序和動作提供值，請遵循下列步驟。您必須先提供所有必要的值，這些值會以星號 (*) 標示，才能儲存您的邏輯應用程式。

初始化工作階段

針對佇列中收到訊息時 (查看鎖定)觸發程序，請提供這項資訊，使範本可以透過使用工作階段識別碼屬性來初始化工作階段，例如：

Service Bus trigger details for

注意

一開始，輪詢間隔會設定為三分鐘，讓邏輯應用程式不會比您所預期地更頻繁執行，並產生非預期的帳單費用。在理想情況下，請將間隔和頻率設定為 30 秒，使邏輯應用程式可在訊息送達時立即觸發。

屬性	此案例的必要項目	值	Description
佇列名稱	Yes	<queue-name>	您先前建立的服務匯流排佇列名稱。此範例使用 "Fabrikam-Service-Bus-Queue"。
佇列類型	Yes	主要	您的主要服務匯流排佇列
工作階段識別碼	Yes	下一個可用的值	此選項會根據服務匯流排佇列中訊息的工作階段識別碼，取得每個觸發程序執行的工作階段。工作階段也會鎖定，讓其他邏輯應用程式或其他用戶端無法處理與這個工作階段相關的訊息。工作流程的後續動作會處理與該工作階段相關聯的所有訊息，本文稍後會再描述。以下是其他工作階段識別碼選項的詳細資訊： - 無：預設選項，不會產生任何工作階段，且無法用於實作循序群組模式。 - 輸入自訂值：當您知道要使用的工作階段識別碼，而且想要一律為工作階段識別碼執行該觸發程序時，請使用此選項。注意：服務匯流排連接器一次可以儲存有限數目的唯一工作階段，從 Azure 服務匯流排到連接器快取。如果工作階段計數超過此限制，則會從快取中移除舊的工作階段。如需詳細資訊，請參閱使用 Azure 服務匯流排與 Azure Logic Apps 在雲端中交換訊息。
間隔	Yes	<number-of-intervals>	檢查訊息之前，週期之間的時間單位數。
頻率	Yes	秒、分鐘、小時、天、週或月	檢查訊息時要使用的週期時間單位。提示：若要新增 [時區]或 [開始時間]，請從 [新增參數] 清單中選取這些屬性。

如需更多觸發程序的資訊，請參閱「服務匯流排 - 佇列中收到訊息時 (查看鎖定)」。觸發程序會輸出 ServiceBusMessage。

初始化工作階段之後，工作流程會使用 初始化變數動作來建立一開始設定為 false 的布林值變數，並指出下列條件何時為 true：

工作階段中沒有其他訊息可供讀取。
工作階段鎖定不再需要更新，因此目前的工作流程執行個體可以完成。

接下來，在 Try 區塊中，工作流程會在第一個已讀取的訊息上執行動作。

處理初始訊息

第一個動作是預留位置服務匯流排動作，將初始訊息傳送至主題，其中您可以用任何其他動作來取代您想要在佇列工作階段所處理的第一個訊息。工作階段識別碼會指定來自訊息來源位置的工作階段。

預留位置服務匯流排動作會將第一個訊息傳送至 [工作階段識別碼] 屬性所指定的服務匯流排主題。如此一來，與特定工作階段相關聯的所有訊息都會移至相同主題。此範本中後續動作的所有 [工作階段識別碼] 屬性都會使用相同的工作階段識別碼值。

Service Bus action details for

在服務匯流排動作中，完成佇列中的初始訊息，提供服務匯流排佇列的名稱，並在動作中保留所有其他預設屬性值。
在服務匯流排動作中，放棄來自佇列的初始訊息，提供服務匯流排佇列的名稱，並在動作中保留所有其他預設屬性值。

接下來，您將針對遵循完成佇列中的初始訊息動作提供必要資訊。您將從當佇列中的工作階段有更多訊息時迴圈中的動作開始。

當佇列中的工作階段有更多訊息時

此 Until 迴圈會在訊息存在於佇列時，或是一小時過後執行這些動作。若要變更迴圈的時間限制，請編輯迴圈的 Timeout 屬性值。

當訊息存在時，從佇列取得其他訊息。
檢查剩餘訊息數。如果訊息仍存在，請繼續處理訊息。如果不存在任何訊息，工作流程會將 isDone 變數設定為 true，並結束迴圈。

Until loop - Process messages while in queue

在服務匯流排動作中，從工作階段取得其他訊息，並提供服務匯流排佇列的名稱。否則，請在動作中保留所有其他預設屬性值。

注意

根據預設，訊息數目上限會設定為 175 ，但此限制會受到服務匯流排中的訊息大小和訊息大小上限屬性所影響。如需詳細資訊，請參閱佇列的訊息大小。

接下來，工作流程會分割成這些平行分支：
- 如果在檢查其他訊息時發生錯誤或失敗，請將 isDone 變數設定為 true。
- 如果收到任何訊息則處理訊息條件會檢查剩餘訊息數是否為零。如果存在 false 和更多訊息，請繼續處理。如果為 true 且不存在任何訊息，工作流程會將 isDone 變數設定為 true。
在 [若為 false]區段中，For each 迴圈會以先進先出 (FIFO) 順序處理每個訊息。在迴圈的 [設定]中，並行控制設定會設為 1，因此一次只會處理單一訊息。
針對服務匯流排動作，完成佇列中的訊息，並放棄佇列中的訊息，並提供服務匯流排佇列的名稱。

當佇列中的工作階段有更多訊息時完成之後，工作流程會將 isDone 變數設定為 true。

接下來，您將提供更新工作階段鎖定直到取消迴圈中的動作必要資訊。

更新工作階段鎖定直到取消

此 Until 迴圈可確保此邏輯應用程式會在佇列中有訊息存在時，或是一小時過後透過執行這些動作，保留工作階段鎖定。若要變更迴圈的時間限制，請編輯迴圈的 Timeout 屬性值。

延遲 25 秒或少於所處理佇列鎖定逾時持續時間的時間量。最小鎖定持續時間為 30 秒，因此預設值就以足夠。不過，您可以透過適當地調整迴圈執行的次數，將其最佳化。
檢查 isDone 變數是否設定為 true。
- 如果 isDone 設定為 true ，工作流程仍在處理訊息，因此工作流程會更新佇列中的工作階段鎖定，然後再次檢查迴圈條件。
  
  您必須在服務匯流排動作中提供服務匯流排佇列的名稱，更新佇列中工作階段的鎖定。
- 如果 isDone 設定為 true，則工作流程不會更新佇列中工作階段的鎖定以及結束迴圈。

更新佇列中工作階段的鎖定

此服務匯流排動作會在工作流程仍在處理訊息時，更新佇列中工作階段的鎖定。

在服務匯流排動作中，更新佇列中工作階段的鎖定，並提供服務匯流排佇列的名稱。

接下來，您將提供服務匯流排動作的必要資訊，關閉佇列中的工作階段並成功。

關閉佇列中的工作階段並成功

此服務匯流排動作會在工作流程完成處理佇列中的所有可用訊息之後，關閉佇列中的工作階段，或工作流程會放棄初始訊息。

在服務匯流排動作中，關閉佇列中的工作階段並成功，並提供服務匯流排佇列的名稱。

下列各節描述 Catch 區段中的動作，這些動作會處理工作流程中發生的錯誤和例外狀況。

關閉佇列中的工作階段並失敗

此服務匯流排動作會一律當做在 Catch 範圍中的第一個動作執行，並關閉佇列中的工作階段。

在服務匯流排動作中，關閉佇列中的工作階段並失敗，並提供服務匯流排佇列的名稱。

接下來，工作流程會建立陣列，其中包含 Try 範圍中所有動作的輸入和輸出，使邏輯應用程式可以存取所發生錯誤或失敗的相關資訊。

從 'Try' 區塊尋找失敗訊息

此篩選陣列動作會透過使用 result() 函式，依據指定的準則，在 Try 範圍內建立來自所有動作之輸入和輸出的陣列。在此情況下，此動作會使用 equals() 函式和 item() 函式，傳回具有 Failed 狀態的動作輸出。

Filter array action -

以下是此動作的 JSON 定義：

"Find_failure_msg_from_'Try'_block": {
   "inputs": {
      "from": "@Result('Try')",
      "where": "@equals(item()['status'], 'Failed')"
   },
   "runAfter": {
      "Close_the_session_in_the_queue_and_fail": [
         "Succeeded"
      ]
   },
   "type": "Query"
},

接下來，工作流程會使用 JSON 物件建立陣列，其中包含從 Find failure msg from 'Try' block 動作傳回之陣列中的錯誤資訊。

選取錯誤詳細資料

此選取動作會根據上一個動作 Find failure msg from 'Try' block 所輸出的輸入陣列，建立包含 JSON 物件的陣列。具體來說，此動作會傳回陣列，而該陣列在陣列中只有每個物件的指定屬性。在此情況下，陣列包含動作名稱和錯誤結果屬性。

Select action -

以下是此動作的 JSON 定義：

"Select_error_details": {
   "inputs": {
      "from": "@body('Find_failure_msg_from_''Try''_block')[0]['outputs']",
      "select": {
         "action": "@item()['name']",
         "errorResult": "@item()"
      }
   },
   "runAfter": {
      "Find_failure_msg_from_'Try'_block": [
         "Succeeded"
      ]
   },
   "type": "Select"
},

接下來，工作流程會停止邏輯應用程式執行，並傳回執行狀態，以及發生錯誤或失敗的詳細資訊。

終止邏輯應用程式執行

此終止動作會停止邏輯應用程式執行，並傳回 Failed 作為邏輯應用程式執行的狀態，以及工作階段識別碼和來自 Select error details 動作所產生的錯誤結果。

Terminate action to stop logic app run

以下是此動作的 JSON 定義：

"Terminate": {
   "description": "This Failure Termination only runs if the Close Session upon Failure action runs - otherwise the LA will be terminated as Success",
   "inputs": {
      "runError": {
         "code": "",
         "message": "There was an error processing messages for Session ID @{triggerBody()?['SessionId']}. The following error(s) occurred: @{body('Select_error_details')['errorResult']}"
         },
         "runStatus": "Failed"
      },
      "runAfter": {
         "Select_error_details": [
            "Succeeded"
         ]
      },
      "type": "Terminate"
   }
},

儲存並執行邏輯應用程式

完成範本之後，您現在可以儲存邏輯應用程式。在設計師工具列上選取儲存。

若要測試邏輯應用程式，請將訊息傳送至您的服務匯流排佇列。

下一步

深入了解服務匯流排連接器的觸發程序和動作

共用方式為

必要條件

檢查服務匯流排命名空間的存取

建立邏輯應用程式

範本摘要

"Try" 範圍

分支 #1：完成佇列中的初始訊息

分支 #2：放棄來自佇列的初始訊息

"Catch" 範圍

完成範本

初始化工作階段

處理初始訊息

當佇列中的工作階段有更多訊息時

更新工作階段鎖定直到取消

更新佇列中工作階段的鎖定

關閉佇列中的工作階段並成功

關閉佇列中的工作階段並失敗

從 'Try' 區塊尋找失敗訊息

選取錯誤詳細資料

終止邏輯應用程式執行

儲存並執行邏輯應用程式

下一步

其他資源

共用方式為

在具 Azure 服務匯流排的 Azure Logic Apps 中使用循序群組傳送相關訊息

必要條件

檢查服務匯流排命名空間的存取

建立邏輯應用程式

範本摘要

"Try" 範圍

分支 #1：完成佇列中的初始訊息

分支 #2：放棄來自佇列的初始訊息

"Catch" 範圍

完成範本

初始化工作階段

處理初始訊息

當佇列中的工作階段有更多訊息時

更新工作階段鎖定直到取消

更新佇列中工作階段的鎖定

關閉佇列中的工作階段並成功

關閉佇列中的工作階段並失敗

從 'Try' 區塊尋找失敗訊息

選取錯誤詳細資料

終止邏輯應用程式執行

儲存並執行邏輯應用程式

下一步

其他資源