通过 Azure 服务总线在 Azure 逻辑应用中使用顺序保护按顺序发送相关消息Send related messages in order by using a sequential convoy in Azure Logic Apps with Azure Service Bus

如果需要按特定顺序发送相关消息,则可在通过 Azure 服务总线连接器使用 Azure 逻辑应用时按“顺序保护”模式操作。When you need to send correlated messages in a specific order, you can follow the sequential convoy pattern when using Azure Logic Apps by using the Azure Service Bus connector. 相关消息有一个用于定义这些消息之间的关系的属性,例如服务总线中的会话的 ID。Correlated messages have a property that defines the relationship between those messages, such as the ID for the session in Service Bus.

例如,假设你有 10 条消息用于名为“会话 1”的会话,有 5 条消息用于名为“会话 2”的会话,且所有这些消息都发送到同一服务总线队列For example, suppose that you have 10 messages for a session named "Session 1", and you have 5 messages for a session named "Session 2" that are all sent to the same Service Bus queue. 你可以创建一个逻辑应用,用于处理来自队列的消息,使来自“会话 1”的所有消息都由单个触发器运行处理,使来自“会话 2”的所有消息则都由下一个触发器运行处理。You can create a logic app that processes messages from the queue so that all messages from "Session 1" are handled by a single trigger run and all messages from "Session 2" are handled by the next trigger run.

常规顺序保护模式

本文介绍了如何通过“使用服务总线会话按顺序传送相关消息”模板来创建实现此模式的逻辑应用。This article shows how to create a logic app that implements this pattern by using the Correlated in-order delivery using service bus sessions template. 此模板定义了一个逻辑应用工作流,该工作流从服务总线连接器的“队列中收到消息时(扫视锁定)”触发器开始,而该触发器从服务总线队列接收消息。This template defines a logic app workflow that starts with the Service Bus connector's When a message is received in a queue (peek-lock) trigger, which receives messages from a Service Bus queue. 下面是此逻辑应用执行的大致步骤:Here are the high-level steps that this logic app performs:

  • 根据触发器从服务总线队列读取的消息初始化会话。Initialize a session based on a message that the trigger reads from the Service Bus queue.

  • 在当前工作流运行期间,读取并处理来自队列中同一会话的所有消息。Read and process all the messages from the same session in the queue during the current workflow run.

若要查看此模板的 JSON 文件,请参阅 GitHub:service-bus-sessions.jsonTo review this template's JSON file, see GitHub: service-bus-sessions.json.

先决条件Prerequisites

  • Azure 订阅。An Azure subscription. 如果没有订阅,可以注册 Azure 试用帐户If you don't have a subscription, sign up for a trial Azure account.

  • 服务总线命名空间和服务总线队列,后者是将在逻辑应用中使用的消息传送实体。A Service Bus namespace and a Service Bus queue, which is a messaging entity that you'll use in your logic app. 这些项和你的逻辑应用需使用同一 Azure 订阅。These items and your logic app need to use the same Azure subscription. 在创建队列时,请确保选择“启用会话”。Make sure that you select Enable sessions when you create your queue. 如果没有这些项,请了解如何创建服务总线命名空间和队列If you don't have these items, learn how to create your Service Bus namespace and a queue.

    重要

    当你同时选择具有相同连接器类型的触发器和操作并使用它们处理相同的实体(如消息队列或主题订阅)时,请务必谨慎。Use caution when you select both a trigger and action that have the same connector type and use them to work with the same entity, such as a messaging queue or topic subscription. 这种组合可以创建无限循环,从而导致逻辑应用永不结束。This combination can create an infinite loop, which results in a logic app that never ends.

  • 有关如何创建逻辑应用的基础知识。Basic knowledge about how to create logic apps. 如果不熟悉 Azure 逻辑应用,请尝试学习快速入门:创建你的第一个自动化工作流If you're new to Azure Logic Apps, try the quickstart, Create your first automated workflow.

检查对服务总线命名空间的访问权限Check access to Service Bus namespace

如果不确定逻辑应用是否有权访问服务总线命名空间,请确认这些权限。If you're not sure whether your logic app has permissions to access your Service Bus namespace, confirm those permissions.

  1. 登录到 Azure 门户Sign in to the Azure portal. 找到并选择你的服务总线命名空间。Find and select your Service Bus namespace.

  2. 在命名空间菜单上的“设置”下,选择“共享访问策略”。 On the namespace menu, under Settings, select Shared access policies. 在“声明”下,检查你是否有该命名空间的“管理”权限。 Under Claims, check that you have Manage permissions for that namespace.

    管理服务总线命名空间的权限

  3. 现在,获取你的服务总线命名空间的连接字符串。Now get the connection string for your Service Bus namespace. 稍后在创建从逻辑应用到命名空间的连接时可以使用此字符串。You can use this string later when you create a connection to the namespace from your logic app.

    1. 在“共享访问策略”窗格中,在“策略”下,选择“RootManageSharedAccessKey” 。On the Shared access policies pane, under Policy, select RootManageSharedAccessKey.

    2. 在主连接字符串旁边选择复制按钮。Next to your primary connection string, select the copy button. 保存连接字符串供以后使用。Save the connection string for later use.

      复制服务总线命名空间连接字符串

    提示

    若要确认连接字符串是与服务总线命名空间关联还是与消息传送实体(例如队列)关联,请在该连接字符串中搜索 EntityPath  参数。To confirm whether your connection string is associated with your Service Bus namespace or a messaging entity, such as a queue, search the connection string for the EntityPath parameter. 如果找到了该参数,则表示连接字符串适用于特定的实体,不是适用于逻辑应用的正确字符串。If you find this parameter, the connection string is for a specific entity, and isn't the correct string to use with your logic app.

创建逻辑应用Create logic app

在此部分中,你将使用“使用服务总线会话按顺序传送相关消息”模板来创建逻辑应用,该模板包括用于实现此工作流模式的触发器和操作。In this section, you create a logic app by using the Correlated in-order delivery using service bus sessions template, which includes the trigger and actions for implementing this workflow pattern. 另外,请创建一个到服务总线命名空间的连接,并指定要使用的服务总线队列的名称。You also create a connection to your Service Bus namespace and specify the name for the Service Bus queue that you want to use.

  1. Azure 门户中,创建一个空白逻辑应用。In the Azure portal, create a blank logic app. 在 Azure 主页中,选择“创建资源” > “集成” > “逻辑应用”。 From the Azure home page, select Create a resource > Integration > Logic App.

  2. 在模板库出现后,滚动经过视频和常用触发器部分。After the template gallery appears, scroll past the video and the common triggers sections. 从“模板”部分中,选择“使用服务总线会话按顺序传送相关消息”模板。From the Templates section, select the template, Correlated in-order delivery using service bus sessions.

    选择“使用服务总线会话按顺序传送相关消息”模板

  3. 在确认框出现后,选择“使用此模板”。When the confirmation box appears, select Use this template.

  4. 在逻辑应用设计器中,在“服务总线”形状中,选择“继续”,然后选择形状中显示的加号 ( + )。On the Logic App Designer, in the Service Bus shape, select Continue, and then select the plus sign (+) that appears in the shape.

    选择“继续”以连接到 Azure 服务总线

  5. 现在,通过选择以下任一选项来创建服务总线连接:Now create a Service Bus connection by choosing either option:

    • 若要使用之前从服务总线命名空间复制的连接字符串,请执行以下步骤:To use the connection string that you copied earlier from your Service Bus namespace, follow these steps:

      1. 选择“手动输入连接信息”。Select Manually enter connection information.

      2. 对于“连接名称”,请为连接提供一个名称。For Connection Name, provide a name for your connection. 对于“连接字符串”,请粘贴你的命名空间连接字符串,然后选择“创建”,例如: For Connection String, paste your namespace connection string, and select Create, for example:

        输入连接名称和服务总线连接字符串

        提示

        如果你没有此连接字符串,请了解如何查找并复制服务总线命名空间连接字符串If you don't have this connection string, learn how to find and copy the Service Bus namespace connection string.

    • 若要从当前 Azure 订阅中选择服务总线命名空间,请执行以下步骤:To select a Service Bus namespace from your current Azure subscription, follow these steps:

      1. 对于“连接名称”,请为连接提供一个名称。For Connection Name, provide a name for your connection. 对于“服务总线命名空间”,请选择你的服务总线命名空间,例如:For Service Bus Namespace, select your Service Bus namespace, for example:

        输入连接名称并选择服务总线命名空间

      2. 当下一个窗格出现时,请选择你的服务总线策略,然后选择“创建”。When the next pane appears, select your Service Bus policy, and select Create.

        选择服务总线策略,然后选择“创建”

  6. 完成后,选择“继续”。When you're done, select Continue.

    逻辑应用设计器现在会显示“使用服务总线会话按顺序传送相关消息”模板,该模板包含一个预先填充了触发器和操作的工作流,其中包括的两个作用域实现了遵循 Try-Catch 模式的错误处理。The Logic App Designer now shows the Correlated in-order delivery using service bus sessions template, which contains a pre-populated workflow with a trigger and actions, including two scopes that implement error handling that follow the Try-Catch pattern.

现在,你可以了解有关模板中的触发器和操作的详细信息,也可以直接跳到为逻辑应用模板提供值Now you can either learn more about the trigger and actions in the template, or jump ahead to provide the values for the logic app template.

模板摘要Template summary

下面是折叠详细信息后,“使用服务总线会话按顺序传送相关消息”模板中的大致工作流:Here is the top-level workflow in the Correlated in-order delivery using service bus sessions template when the details are collapsed:

模板的大致工作流

名称Name 说明Description
When a message is received in a queue (peek-lock) 此服务总线触发器会根据指定的重复周期检查指定的服务总线队列中是否有任何消息。Based on the specified recurrence, this Service Bus trigger checks the specified Service Bus queue for any messages. 如果队列中存在消息,则触发器将触发,这将创建并运行一个工作流实例。If a message exists in the queue, the trigger fires, which creates and runs a workflow instance.

“扫视锁定”这一术语是指触发器发送从队列中检索消息的请求。The term peek-lock means that the trigger sends a request to retrieve a message from the queue. 如果有消息存在,则触发器会检索并锁定该消息,防止在锁定期限到期之前对该消息进行其他处理。If a message exists, the trigger retrieves and locks the message so that no other processing happens on that message until the lock period expires. 有关详细信息,请参阅初始化会话For details, Initialize the session.

Init isDone 初始化变量操作创建一个设置为 false 的布尔变量,并指示何时满足以下条件:This Initialize variable action creates a Boolean variable that's set to false and indicates when the following conditions are true:

- 会话中没有剩余可供读取的消息。- No more messages in the session are available to read.
- 不再需要续订会话锁,就能完成当前工作流实例。- The session lock no longer needs to be renewed so that the current workflow instance can finish.

有关详细信息,请参阅初始化会话For details, see Initialize the session.

Try 作用域操作包含处理消息时要运行的操作。This Scope action contains the actions that run to process a message. 如果 Try 作用域中出现问题,则后续的 Catch 作用域操作会处理该问题。If a problem happens in the Try scope, the subsequent Catch Scope action handles that problem. 有关详细信息,请参阅“Try”作用域For more information, see "Try" scope.
Catch 作用域操作包含在上述 Try 作用域中出现问题时将运行的操作。This Scope action contains the actions that run if a problem happens in the preceding Try scope. 有关详细信息,请参阅“Catch”作用域For more information, see "Catch" scope.

“Try”作用域"Try" scope

下面是折叠详细信息后 Try 作用域操作中的大致流:Here is the top-level flow in the Try scope action when the details are collapsed:

“Try”作用域操作工作流

名称Name 说明Description
Send initial message to topic 可以将此操作替换对来自队列中会话的第一条消息进行处理时需要执行的任何操作。You can replace this action with whatever action that you want to handle the first message from the session in the queue. 会话 ID 用于指定会话。The session ID specifies the session.

对于此模板,服务总线操作会将第一条消息发送到一个服务总线主题。For this template, a Service Bus action sends the first message to a Service Bus topic. 有关详细信息,请参阅处理初始消息For details, see Handle the initial message.

(并行分支)(parallel branch) 并行分支操作创建两个路径:This parallel branch action creates two paths:

- 分支 1:继续处理消息。- Branch #1: Continue processing the message. 有关详细信息,请参阅分支 1:完成队列中的初始消息For more information, see Branch #1: Complete initial message in queue.

- 分支 2:如果出现问题,则放弃该消息,将其释放,供另一个触发器运行来拾取。- Branch #2: Abandon the message if something goes wrong, and release for pickup by another trigger run. 有关详细信息,请参阅分支 2:放弃队列中的初始消息For more information, see Branch #2: Abandon initial message from queue.

这两个路径稍后会在“关闭队列中的会话并成功”操作中汇合,详见下一行中的说明。Both paths join up later in the Close session in a queue and succeed action, described in the next row.

Close a session in a queue and succeed 此服务总线操作将前面所述的分支汇合,并在发生以下事件之一后关闭队列中的会话:This Service Bus action joins the previously described branches and closes the session in the queue after either of the following events happen:

- 工作流完成了对队列中可用消息的处理。- The workflow finishes processing available messages in the queue.
- 工作流因为某个错误放弃了初始消息。- The workflow abandons the initial message because something went wrong.

有关详细信息,请参阅关闭队列中的会话并成功For details, see Close a session in a queue and succeed.

分支 1:完成队列中的初始消息Branch #1: Complete initial message in queue

名称Name 说明Description
Complete initial message in queue 此服务总线操作会将成功检索的消息标记为完成,然后从队列中删除该消息,防止重新处理。This Service Bus action marks a successfully retrieved message as complete and removes the message from the queue to prevent reprocessing. 有关详细信息,请参阅处理初始消息For details, see Handle the initial message.
While there are more messages for the session in the queue Until 循环会在有消息存在时或一小时过后继续获取消息。This Until loop continues to get messages while messages exists or until one hour passes. 有关此循环中的操作的详细信息,请参阅当队列中的会话存在剩余消息时For more information about the actions in this loop, see While there are more messages for the session in the queue.
Set isDone = true 当不存在剩余消息时,此“设置变量”操作会将 isDone 设置为 trueWhen no more messages exist, this Set variable action sets isDone to true.
Renew session lock until cancelled Until 循环可确保此逻辑应用在有消息存在时或一小时过后持有会话锁。This Until loop makes sure that the session lock is held by this logic app while messages exist or until one hour passes. 有关此循环中的操作的详细信息,请参阅续订会话锁直至取消For more information about the actions in this loop, see Renew session lock until cancelled.

分支 2:放弃队列中的初始消息Branch #2: Abandon initial message from the queue

如果处理第一条消息的操作失败,则服务总线操作“放弃队列中的初始消息”会释放此消息,供另一个工作流实例运行来拾取并处理。If the action that handles the first message fails, the Service Bus action, Abandon initial message from the queue, releases the message for another workflow instance run to pick up and process. 有关详细信息,请参阅处理初始消息For details, see Handle the initial message.

“Catch”作用域"Catch" scope

如果 Try 作用域中的操作失败,则逻辑应用仍必须关闭会话。If actions in the Try scope fail, the logic app must still close the session. Try 作用域操作导致 FailedSkippedTimedOut 状态时,Catch 作用域操作会运行。The Catch scope action runs when the Try scope action results in the status, Failed, Skipped, or TimedOut. 此作用域会返回一条错误消息(其中包含发生问题的会话 ID)并终止逻辑应用。The scope returns an error message that includes the session ID where the problem happened, and terminates the logic app.

下面是折叠详细信息后 Catch 作用域操作中的大致流:Here is the top-level flow in the Catch scope action when the details are collapsed:

“Catch”作用域操作工作流

名称Name 说明Description
Close a session in a queue and fail 此服务总线操作会关闭队列中的会话,使会话锁不再保持打开状态。This Service Bus action closes the session in the queue so that the session lock doesn't stay open. 有关详细信息,请参阅关闭队列中的会话并失败For details, see Close a session in a queue and fail.
Find failure msg from 'Try' block “筛选数组”操作按照指定的条件基于 Try 作用域内的所有操作的输入和输出创建一个数组。This Filter Array action creates an array from the inputs and outputs from all the actions inside the Try scope based on the specified criteria. 在这种情况下,此操作会返回导致 Failed 状态的操作的输出。In this case, this action returns the outputs from the actions that resulted in Failed status. 有关详细信息,请参阅从“Try”块查找失败消息For details, see Find failure msg from 'Try' block.
Select error details “选择”操作根据指定的条件创建包含 JSON 对象的数组。This Select action creates an array that contains JSON objects based on the specified criteria. 这些 JSON 对象是根据上一操作(Find failure msg from 'Try' block)创建的数组中的值生成的。These JSON objects are built from the values in the array created by the previous action, Find failure msg from 'Try' block. 在这种情况下,此操作会返回一个数组,其中包含根据从上一操作返回的错误详细信息创建的 JSON 对象。In this case, this action returns an array that contains a JSON object created from the error details returned from the previous action. 有关详细信息,请参阅“选择”错误详细信息For details, see Select error details.
Terminate “终止”操作会停止工作流的运行,取消正在进行的任何操作,跳过任何剩余的操作,并返回指定的状态、会话 ID 以及来自“Select error details”操作的错误结果。This Terminate action stops the run for the workflow, cancels any actions in progress, skips any remaining actions, and returns the specified status, the session ID, and the error result from the Select error details action. 有关详细信息,请参阅终止逻辑应用For details, see Terminate logic app.

完成模板Complete the template

若要为“使用服务总线会话按顺序传送相关消息”模板中的触发器和操作提供值,请执行以下步骤。To provide the values for the trigger and actions in the Correlated in-order delivery using service bus sessions template, follow these steps. 在保存逻辑应用之前,必须提供所有必需的值,这些值标有星号 ( * )。You have to provide all the required values, which are marked by an asterisk (*), before you can save your logic app.

初始化会话Initialize the session

  • 对于“队列中收到消息时(扫视锁定)”触发器,请提供此信息,以便模板可以使用“会话 ID”属性来初始化会话,例如:For the When a message is received in a queue (peek-lock) trigger, provide this information so that the template can initialize a session by using the Session id property, for example:

    服务总线触发器“队列中收到消息时(扫视锁定)”的详细信息

    备注

    轮询间隔一开始设置为三分钟,这样逻辑应用的运行频率就不会超出预期,因此不会产生意外的账单费用。Initially, the polling interval is set to three minutes so that the logic app doesn't run more frequently than you expect and result in unanticipated billing charges. 理想情况下,可将时间间隔和频率设置为 30 秒,这样逻辑应用就会在消息到达时立即触发。Ideally, set the interval and frequency to 30 seconds so that the logic app triggers immediately when a message arrives.

    属性Property 此方案所需Required for this scenario ValueValue 说明Description
    队列名称Queue name Yes <queue-name> 之前创建的服务总线队列的名称。The name for your previously created Service Bus queue. 此示例使用“Fabrikam-Service-Bus-Queue”。This example uses "Fabrikam-Service-Bus-Queue".
    队列类型Queue type Yes 主页Main 你的主服务总线队列Your primary Service Bus queue
    会话 IDSession id Yes 下一个可用Next available 此选项会根据服务总线队列中消息的会话 ID 为每个触发器运行获取一个会话。This option gets a session for each trigger run, based on the session ID from the message in the Service Bus queue. 此会话也被锁定,以防止其他逻辑应用或其他客户端处理与此会话相关的消息。The session is also locked so that no other logic app or other client can process messages that are related to this session. 工作流的后续操作将处理与该会话关联的所有消息,如本文稍后所述。The workflow's subsequent actions process all the messages that are associated with that session, as described later in this article.

    下面是有关其他会话 ID 选项的详细信息:Here is more information about the other Session id options:

    - :默认选项,会导致没有会话,不能用于实现顺序保护模式。- None: The default option, which results in no sessions and can't be used for implementing the sequential convoy pattern.

    - 输入自定义值:如果你知道要使用的会话 ID,并且始终要为该会话 ID 运行触发器,请使用此选项。- Enter custom value: Use this option when you know the session ID that you want to use, and you always want to run the trigger for that session ID.

    注意:服务总线连接器一次可以将 Azure 服务总线的有限数量的唯一会话保存到连接器缓存。Note: The Service Bus connector can save a limited number of unique sessions at a time from Azure Service Bus to the connector cache. 如果会话计数超过此限制,则将从缓存中删除旧会话。If the session count exceeds this limit, old sessions are removed from the cache. 有关详细信息,请参阅使用 Azure 逻辑应用和 Azure 服务总线在云中交换消息For more information, see Exchange messages in the cloud with Azure Logic Apps and Azure Service Bus.

    时间间隔Interval Yes <时间间隔数><number-of-intervals> 在重复检查是否存在消息之前,在两个重复周期之间间隔的时间单位数。The number of time units between recurrences before checking for a message.
    频率Frequency Yes “秒”、“分钟”、“小时”、“天”、“周”或“月” Second, Minute, Hour, Day, Week, or Month 在检查是否存在消息时要使用的重复周期的时间单位。The unit of time for the recurrence to use when checking for a message.

    提示:若要添加“时区”或“开始时间”,请从“添加新参数”列表中选择这些属性。 Tip: To add a Time zone or Start time, select these properties from the Add new parameter list.

    如需更多的触发器信息,请参阅服务总线 - 队列中收到消息时(扫视锁定)For more trigger information, see Service Bus - When a message is received in a queue (peek-lock). 触发器会输出一条 ServiceBusMessageThe trigger outputs a ServiceBusMessage.

初始化会话以后,工作流会使用“初始化变量”操作创建一个最初设置为 false 的布尔变量,并指示何时满足以下条件:After initializing the session, the workflow uses the Initialize variable action to create a Boolean variable that initially set to false and indicates when the following conditions are true:

  • 会话中没有剩余可供读取的消息。No more messages in the session are available to read.

  • 不再需要续订会话锁,就能完成当前工作流实例。The session lock no longer needs to be renewed so that the current workflow instance can finish.

与“Init isDone”相对应的“初始化变量”操作的详细信息

接下来,在 Try 块中,工作流对读取的第一条消息执行操作。Next, in the Try block, the workflow performs actions on the first message that's read.

处理初始消息Handle the initial message

第一个操作是占位符服务总线操作“将初始消息发送到主题”,你可以将其替换为对队列中会话的第一条消息进行处理时需要执行的任何操作。The first action is a placeholder Service Bus action, Send initial message to topic, which you can replace with any other action that you want to handle the first message from the session in the queue. 会话 ID 指定产生消息的会话。The session ID specifies the session from where the message originated.

此占位符服务总线操作会将第一条消息发送到由“会话 ID”属性指定的服务总线主题。The placeholder Service Bus action sends the first message to a Service Bus topic that's specified by the Session Id property. 这样一来,与特定会话关联的所有消息都将移至同一主题。That way, all the messages that are associated with a specific session go to the same topic. 此模板中的后续操作的所有“会话 ID”属性均使用同一会话 ID 值。All Session Id properties for subsequent actions in this template use the same session ID value.

服务总线操作“将初始消息发送到主题”的详细信息

  1. 在服务总线操作“完成队列中的初始消息”中,提供你的服务总线队列的名称,并保留操作中的所有其他默认属性值。In the Service Bus action, Complete initial message in queue, provide the name for your Service Bus queue, and keep all the other default property values in the action.

    服务总线操作“完成队列中的初始消息”的详细信息

  2. 在服务总线操作“放弃队列中的初始消息”中,提供你的服务总线队列的名称,并保留操作中的所有其他默认属性值。In the Service Bus action, Abandon initial message from the queue, provide the name for your Service Bus queue, and keep all the other default property values in the action.

    服务总线操作“放弃队列中的初始消息”的详细信息

接下来,你将为“完成队列中的初始消息”操作后面的操作提供必要信息。Next, you'll provide the necessary information for the actions that follow the Complete initial message in queue action. 你将从“当队列中的会话存在剩余消息时”循环中的操作开始。You'll start with the actions in the While there are more messages for the session in the queue loop.

当队列中的会话存在剩余消息时While there are more messages for the session in the queue

Until 循环会在队列中有消息存在时或一小时过后运行这些操作。This Until loop runs these actions while messages exist in the queue or until one hour passes. 若要更改循环的时间限制,请编辑循环的“超时”属性值。To change the loop's time limit, edit the loop's Timeout property value.

  • 当有消息存在时从队列中获取其他消息。Get additional messages from the queue while messages exist.

  • 检查剩余消息的数目。Check the number of remaining messages. 如果仍然有消息存在,则继续处理消息。If messages still exist, continue processing messages. 如果不存在剩余消息,则工作流会将 isDone 变量设置为 true,并退出循环。If no more messages exist, the workflow sets the isDone variable to true, and exits the loop.

Until 循环 - 处理队列中的消息

  1. 在服务总线操作“从会话中获取其他消息”中,提供你的服务总线队列的名称。In the Service Bus action, Get additional messages from session, provide the name for your Service Bus queue. 另外,请保留操作中的所有其他默认属性值。Otherwise, keep all the other default property values in the action.

    备注

    默认情况下,消息的最大数目设置为 175,但此限制受服务总线中的消息大小和最大消息大小属性的影响。By default, the maximum number of messages is set to 175, but this limit is affected by the message size and maximum message size property in Service Bus. 有关详细信息,请参阅队列的消息大小For more information, see Message size for a queue.

    服务总线操作 -“从会话中获取其他消息”

    接下来,工作流会拆分为以下并行分支:Next, the workflow splits into these parallel branches:

    • 如果检查是否存在其他消息时发生错误或失败,则将 isDone 变量设置为 trueIf an error or failure happens while checking for additional messages, set the isDone variable to true.

    • “在获得消息的情况下处理消息”条件检查剩余消息数是否为零。The Process messages if we got any condition checks whether the number of remaining messages is zero. 如果不为零并且存在剩余消息,则继续处理消息。If false and more messages exist, continue processing. 如果为零并且不存在剩余消息,则工作流会将 isDone 变量设置为 trueIf true and no more messages exist, the workflow sets the isDone variable to true.

    条件 - 在有消息的情况下处理消息

    If false 部分中,每个 For each 循环按先进先出顺序 (FIFO) 处理每条消息。In the If false section, a For each loop processes each message in first-in, first-out order (FIFO). 在循环的“设置”中,“并发控制”设置设定为 1,因此一次只处理一条消息。 。In the loop's Settings, the Concurrency Control setting is set to 1, so only a single message is processed at a time.

    “For each”循环 - 一次处理一条消息

  2. 对于服务总线操作“完成队列中的消息”和“放弃队列中的消息” ,请提供你的服务总线队列的名称。For the Service Bus actions, Complete the message in a queue and Abandon the message in a queue, provide the name for your Service Bus queue.

    服务总线操作 -“完成队列中的消息”和“放弃队列中的消息”

    “当队列中的会话存在剩余消息时”循环完成后,工作流会将 isDone 变量设置为 trueAfter the While there are more messages for the session in the queue is done, the workflow sets the isDone variable to true.

接下来,你将为“续订会话锁直至取消”循环中的操作提供必要信息。Next, you'll provide the necessary information for the actions in the Renew session lock until cancelled loop.

续订会话锁直至取消Renew session lock until cancelled

Until 循环可通过运行这些操作确保此逻辑应用在有消息存在于队列中时或一小时过后持有会话锁。This Until loop makes sure that the session lock is held by this logic app while messages exist in the queue or until one hour passes by running these actions. 若要更改循环的时间限制,请编辑循环的“超时”属性值。To change the loop's time limit, edit the loop's Timeout property value.

  • 延迟 25 秒或延迟比正在处理的队列的锁超时持续时间小的某个时间。Delay for 25 seconds or an amount of time that's less than the lock timeout duration for the queue that's being processed. 最小的锁持续时间为 30 秒,因此默认值就足够了。The smallest lock duration is 30 seconds, so the default value is enough. 不过,可以通过适当调整来优化循环的运行次数。However, you can optimize the number of times that the loop runs by adjusting appropriately.

  • 检查 isDone 变量是否已设置为 trueCheck whether the isDone variable is set to true.

    • 如果 isDone 未设置为 true,则表明工作流仍在处理消息,因此工作流会续订队列中的会话锁,并再次检查循环条件。If isDone is not set to true, the workflow is still processing messages, so the workflow renews the lock on the session in the queue, and checks the loop condition again.

      你需要在服务总线操作续订队列中的会话锁中提供你的服务总线队列的名称。You need to provide the name for your Service Bus queue in the Service Bus action, Renew lock on the session in a queue.

    • 如果 isDone 设置为 true,则工作流不会续订队列中的会话锁,而是退出循环。If isDone is set to true, the workflow doesn't renew the lock on the session in the queue, and exits the loop.

    Until 循环 -“续订会话锁直至取消”

续订队列中的会话锁Renew lock on the session in a queue

当工作流仍在处理消息时,此服务总线操作会续订队列中的会话锁。This Service Bus action renews the lock on the session in the queue while the workflow is still processing messages.

  • 在服务总线操作“续订队列中的会话锁”中,提供你的服务总线队列的名称。In the Service Bus action, Renew lock on the session in a queue, provide the name for your Service Bus queue.

    服务总线操作 -“续订队列中的会话锁”

接下来,为服务总线操作“关闭队列中的会话并成功”提供必要的信息。Next, you'll provide the necessary information for the Service Bus action, Close a session in a queue and succeed.

关闭队列中的会话并成功Close a session in a queue and succeed

此服务总线操作会在工作流处理完队列中的所有可用消息或放弃初始消息后关闭队列中的会话。This Service Bus action closes the session in the queue after either the workflow finishes processing all the available messages in the queue, or the workflow abandons the initial message.

  • 在服务总线操作“关闭队列中的会话并成功”中,提供你的服务总线队列的名称。In the Service Bus action, Close a session in a queue and succeed, provide the name for your Service Bus queue.

    服务总线操作 -“关闭队列中的会话并成功”

下面的各部分介绍了 Catch 部分中的操作,这些操作处理工作流中发生的错误和异常。The following sections describe the actions in the Catch section, which handle errors and exceptions that happen in your workflow.

关闭队列中的会话并失败Close a session in a queue and fail

此服务总线操作始终作为 Catch 作用域中的第一个操作运行,并会关闭队列中的会话。This Service Bus action always runs as the first action in the Catch scope and closes the session in the queue.

  • 在服务总线操作“关闭队列中的会话并失败”中,提供你的服务总线队列的名称。In the Service Bus action, Close a session in a queue and fail, provide the name for your Service Bus queue.

    服务总线操作 -“关闭队列中的会话并失败”

接下来,工作流会创建一个数组,该数组具有 Try 作用域内所有操作的输入和输出,使逻辑应用可以访问已发生的错误或失败的相关信息。Next, the workflow creates an array that has the inputs and outputs from all the actions in the Try scope so that the logic app can access information about the error or failure that happened.

从“Try”块查找失败消息Find failure msg from 'Try' block

“筛选数组”操作会按照 result() 函数指定的条件创建一个数组,该数组包含 Try 作用域内所有操作的输入和输出。This Filter Array action creates an array that has the inputs and outputs from all the actions inside the Try scope based on the specified criteria by using the result() function. 在这种情况下,此操作会通过使用 equals() 函数item() 函数返回状态为 Failed 的操作的输出。In this case, this action returns the outputs from the actions that have Failed status by using the equals() function and item() function.

“筛选数组”操作 -“从‘Try’块查找失败消息”

下面是此操作的 JSON 定义:Here's the JSON definition for this action:

"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”操作返回的数组中的错误信息。Next, the workflow creates an array with a JSON object that contains the error information in the array returned from the Find failure msg from 'Try' block action.

“选择”错误详细信息Select error details

“选择”操作根据输入数组(上面的操作“Find failure msg from 'Try' block”的输出)创建包含 JSON 对象的数组。This Select action creates an array that contains JSON objects based on the input array that's output from the previous action, Find failure msg from 'Try' block. 具体说来,此操作会返回一个数组,该数组只包含输入数组中每个对象的指定属性。Specifically, this action returns an array that has only the specified properties for each object in the array. 在此示例中,数组包含操作名称和错误结果属性。In this case, the array contains the action name and error result properties.

“选择”操作 -“选择”错误详细信息

下面是此操作的 JSON 定义:Here's the JSON definition for this action:

"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"
},

接下来,工作流会停止逻辑应用运行,并返回运行状态以及已发生的错误或失败的详细信息。Next, the workflow stops the logic app run and returns the run status along with more information about the error or failure that happened.

终止逻辑应用运行Terminate logic app run

“终止”操作会停止逻辑应用运行,并返回 Failed(作为逻辑应用的运行状态)以及会话 ID 和来自“Select error details”操作的错误结果。This Terminate action stops the logic app run and returns Failed as the status for the logic app's run along with the session ID and the error result from the Select error details action.

用于停止逻辑应用运行的“终止”操作

下面是此操作的 JSON 定义:Here's the JSON definition for this action:

"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"
   }
},

保存并运行逻辑应用Save and run logic app

完成模板后,可以保存你的逻辑应用。After you complete the template, you can now save your logic app. 在设计器工具栏上选择“保存”。On the designer toolbar, select Save.

若要测试逻辑应用,请向服务总线队列发送消息。To test your logic app, send messages to your Service Bus queue.

后续步骤Next steps