从 Azure 逻辑应用中的工作流连接到 Azure 服务总线

适用对象：Azure 逻辑应用（消费型 + 标准型）

本指南介绍如何使用服务总线连接器从 Azure 逻辑应用中的自动化和集成工作流访问 Azure 服务总线中的服务总线资源。 可以让服务总线事件触发工作流或运行与服务总线项交互的操作，例如：

• 监视消息何时抵达（自动完成），或者在队列、主题和主题订阅中接收（扫视-锁定）。
• 发送消息。
• 创建和删除主题订阅。
• 管理队列和主题订阅中的消息，例如，获取、获取延期消息、完成、延期、丢弃和加入死信。
• 更新队列和主题订阅中消息和会话的锁定。
• 关闭队列和主题中的会话。

可以使用触发器和操作从 Azure 服务总线获取响应，然后将输出提供给工作流中的其他操作。

连接器技术参考

服务总线连接器具有不同的版本，具体取决于逻辑应用工作流类型和主机环境。

Prerequisites

• Azure 帐户和订阅。 获取试用Azure帐户。

• 服务总线命名空间和消息传送实体，例如队列。

  有关详细信息，请参见:

  • 创建服务总线命名空间

  • 创建服务总线命名空间和队列

  • 创建包含订阅的服务总线命名空间和主题

• 要在其中访问服务总线命名空间和消息传送实体的逻辑应用资源和工作流。

  • 若要使用服务总线触发器启动工作流，请创建一个空白工作流。 若要在工作流中使用服务总线操作，请使用最适合方案的任何触发器启动工作流。

  • 如果逻辑应用资源使用托管标识对服务总线命名空间和消息传送实体的访问权限进行身份验证，请在相应的级别分配必要的角色权限。 例如，若要访问队列，必须给托管标识分配一个具有该队列所需权限的角色。

    • 每个逻辑应用资源只能使用一个托管标识，即使逻辑应用工作流访问不同的消息传送实体也是如此。

    • 访问队列或主题订阅的每个托管标识都需要使用自己的服务总线 API 连接。

    • 每个服务总线操作中，与不同消息实体交换消息并需要不同权限的，都需要使用其自己的服务总线 API 连接。

    有关托管标识的详细信息，请参阅在 Azure 逻辑应用中使用托管标识验证对 Azure 资源的访问。

• 默认情况下，服务总线内置连接器操作是 无状态的。 若要在监控状态模式下运行这些操作，请参阅为无状态内置连接器启用监控状态模式。

Azure 服务总线操作注意事项

无限循环

连接器缓存中会话的保存限制

对于每个服务总线消息传送实体（例如 订阅或主题），服务总线连接器一次可以在连接器缓存中节省多达 1,500 个唯一会话。 如果会话计数超过此限制，连接器缓存将删除旧会话。 有关详细信息，请参阅消息会话。

按顺序发送相关消息

如果需要按特定顺序发送相关消息，请使用服务总线连接器和 顺序护送 模式创建工作流。 相关消息有一个用于定义这些消息之间的关系的属性，例如 Azure 服务总线中的会话的 ID。

创建消费逻辑应用工作流时，可以选择“使用服务总线会话的按顺序关联发送”模板，该模板可实现顺序护航设计模式。 有关详细信息，请参阅按顺序发送相关消息。

大消息支持

使用服务总线内置连接器操作时，大消息支持仅适用于标准工作流。 例如，可以分别使用内置触发器和操作来接收和处理大型消息。

服务总线托管连接器的最大消息大小限制为 1 MB，即使你使用高级层服务总线命名空间。

增大接收和发送消息的超时

在使用服务总线内置操作的标准工作流中，可以增大接收和发送消息的超时。 例如，若要增加接收消息的超时，请更改 Azure Functions 扩展中以下代码示例中的设置：

{
   "version": "2.0",
   "extensionBundle": {
      "id": "Microsoft.Azure.Functions.ExtensionBundle.Workflows",
      "version": "[1.*, 2.0.0)"
   },
   "extensions": {
      "serviceBus": {
         "batchOptions": {
            "operationTimeout": "00:15:00"
         }
      }  
   }
}

若要调高发送消息的超时，请添加 ServiceProviders.ServiceBus.MessageSenderOperationTimeout 应用设置。

服务总线托管连接器触发器

对于服务总线托管连接器，所有触发器都使用 长轮询 模式。 此触发器类型处理所有消息，然后等待 30 秒，使更多消息出现在队列或主题订阅中。 如果在 30 秒内未显示任何消息，则跳过触发器的运行。 否则，该触发器将继续读取消息，直到队列或主题订阅为空。 下一次触发器轮询将基于在触发器的属性中指定的重复间隔。

某些触发器（例如“一条或多条消息抵达队列时(自动完成)”触发器）可能会返回一条或多条消息。 这些触发器在触发时返回的消息数在 1 到触发器的最大消息计数属性指定的数量之间。

服务总线内置连接器触发器

对于服务总线内置连接器，非会话触发器遵循连接器完全管理的 连续轮询 触发器模式。 在此模式中，触发器会不断检查队列或主题订阅中的消息。 相比之下，会话触发器遵循 长轮询 触发器模式。 Azure Functions 设置 clientRetryOptions:tryTimeout 用于管理其配置。 逻辑应用的 host.json 文件中定义的 Azure Functions 托管扩展，以及通过设计器或代码视图设置的逻辑应用工作流中的触发器设置，共享服务总线内置触发器的配置。 本部分介绍这两个设置位置。 请注意以下有关服务总线触发器的详细信息：

• 某些内置触发器，例如队列中有可用消息时触发器，可以返回一个或多个消息。 这些触发器激活时，会返回一条或多条消息。

• 内置触发器 “当消息在队列中可用（V1）” 不支持名为“最大消息批大小” 的参数。 如果使用此参数，则需要改用 V2 版本。 若要使用不支持参数的触发器，可以通过将 maxMessageBatchSize 参数添加到 host.json 文件中的触发器定义来控制收到的消息数。 若要查找此文件，请参阅编辑标准逻辑应用的主机和应用设置。

  "extensions": {
     "serviceBus": {
        "maxMessageBatchSize": 25
     }
  }
  

• 可以通过设计器或代码在服务总线触发器上启用并发，例如：

  "runtimeConfiguration": {
     "concurrency": {
          "runs": 50
      }
  }
  

  • 如果使用批处理设置并发，请保留并发运行数大于总体批大小。 这样，已读消息不会进入等待状态，而是始终在已读时被读取。 在某些情况下，触发器可以达到批次大小的最多两倍。

  • 如果在名为 “当消息在队列（V1）中可用”的触发器上启用并发，并且向队列发送了 100 条或更多消息，则触发器会将所有消息路由到 死信队列。

  • 如果启用并发，则解除批处理或 拆分 行为的限制将减少到 100 个项。 此行为适用于所有触发器，而不仅仅是服务总线触发器。 请确保在启用了并发的任何触发器上，指定的批处理大小都小于此限制。

  • 如果启用并发，则默认情况下，批处理读取之间存在 30 秒的延迟。 这个延迟减慢了触发器实现以下目标的速度：

    • 减少为检查需要应用并发处理的运行次数而发送的存储调用数量。

    • 模拟服务总线托管连接器触发器的行为，该触发器在没有找到消息时会轮询 30 秒。

      尽管可以更改此延迟，但请确保仔细测试对默认值所做的任何更改：

      "workflow": {
          "settings": {
             "Runtime.ServiceProviders.FunctionTriggers.DynamicListenerEnableDisableInterval": "00:00:30"
          }
      }
      

  • 在某些情况下，触发器可能会超出并发设置。 Azure 逻辑应用不会让这些运行失败，而是将它们排入等待状态，直到可以启动它们为止。 该maximumWaitingRuns设置控制等待状态中允许的运行数：

    "runtimeConfiguration": {
       "concurrency": {
          "runs": 100,
          "maximumWaitingRuns": 50
       }
    }
    

    使用服务总线触发器时，请确保仔细测试这些更改，使运行的等待时间不超过消息锁定超时。 有关默认值的详细信息，请参阅此处的并发和取消批处理限制。

步骤 1：检查对服务总线命名空间的访问权限

若要确认逻辑应用资源有权访问服务总线命名空间，请执行以下步骤：

1. 在 Azure 门户中，打开你的服务总线命名空间。

2. 在命名空间边栏的 “设置”下，选择“ 共享访问策略”。 在“声明”下，检查你是否有该命名空间的“管理”权限。

   

步骤 2：获取连接身份验证要求

首次添加服务总线触发器或操作时，系统会提示输入连接信息，包括连接身份验证类型。 根据逻辑应用工作流类型、服务总线连接器版本和所选身份验证类型，需要以下各节中所述的项。

托管连接器身份验证（消费和标准工作流）

内置连接器身份验证（仅限标准工作流）

获取服务总线命名空间的连接字符串

若要在添加服务总线触发器或操作时创建连接，需要服务总线命名空间的连接字符串。 连接字符串以 sb:// 前缀开头。

1. 在 Azure 门户中，打开你的服务总线命名空间。

2. 在命名空间边栏的 “设置”下，选择“ 共享访问策略”。

3. 在“共享访问策略”窗格中，选择“RootManageSharedAccessKey”。

4. 在主连接字符串或辅助连接字符串旁边选择复制按钮。

   

   Note

   若要确认字符串适用于命名空间，而不是特定的消息传送实体，请搜索该参数的 EntityPath 连接字符串。 如果找到了该参数，则表示该连接字符串用于特定的实体，而不是用于工作流的正确字符串。

5. 保存连接字符串供以后使用。

获取服务总线命名空间的终结点 URL

如果您使用服务总线托管连接器，无论选择Microsoft Entra 集成还是逻辑应用托管标识的身份验证类型，都需要此终结点 URL。 终结点 URL 以 sb:// 前缀开头。

1. 在 Azure 门户中，打开你的服务总线命名空间。

2. 在命名空间边栏上，展开 “设置”，然后选择“ 属性”。

3. 在 “概要”下，在 ID 旁边复制并保存终结点 URL 供以后使用。

获取服务总线命名空间的完全限定名称

1. 在 Azure 门户中，打开你的服务总线命名空间。

2. 在命名空间边栏上，选择“ 概述”。

3. 在 “概述 ”页上，展开 “概要”，并找到 “主机名 ”属性。

4. 复制并保存完全限定名称，其格式如下：<your-Service-Bus-namespace>.servicebus.chinacloudapi.cn，以备后用。

步骤 3：选项 1 - 添加服务总线触发器

以下步骤使用 Azure 门户，但通过使用相应的 Azure 逻辑应用扩展，可以使用 Visual Studio Code 创建逻辑应用工作流：

• 消耗工作流： 使用 Visual Studio Code 创建消耗逻辑应用工作流
• 标准工作流： 使用 Visual Studio Code 创建标准逻辑应用工作流

步骤 3：选项 2 - 添加 Service Bus 操作

以下步骤使用 Azure 门户，但通过使用相应的 Azure 逻辑应用扩展，可以使用 Visual Studio Code 生成逻辑应用工作流：

• 消耗工作流： 使用 Visual Studio Code 创建消耗逻辑应用工作流
• 标准工作流： 使用 Visual Studio Code 创建标准逻辑应用工作流

服务总线内置连接器应用设置

在标准逻辑应用资源中，服务总线内置连接器包括控制各种阈值的应用设置。 例如，这些设置控制发送消息的超时以及消息池中每个处理器核心的消息发送者数。 有关详细信息，请参阅 应用设置参考 - local.settings.json。

通过使用 Service Bus 内置触发器来从死信队列中读取消息

在标准工作流中，若要从队列或主题订阅中的死信队列读取消息，请使用指定的触发器执行以下步骤：

1. 在空白工作流中，根据你的场景，添加名为“当消息在队列中可用”或“当消息在主题订阅中可用时（窥视锁）”的服务总线内置连接器触发器。

2. 在触发器中，设置以下参数值以指定您的队列或主题订阅的默认死信队列，您可以像访问其他队列一样，访问此死信队列。

   • 当消息在队列触发器中可用时 ：将 队列名称 参数设置为 queuename/$deadletterqueue。

   • 当在主题订阅中（窥视锁定）有消息可用时 触发器：将 主题名称 参数设置为 topicname/Subscriptions/subscriptionname/$deadletterqueue。

   有关详细信息，请参阅服务总线死信队列概述。

故障排查

对工作流的更新延迟生效

如果服务总线触发器的轮询间隔很短（例如 10 秒），则对工作流的更新可能在长达 10 分钟时间内都不会生效。 若要解决此问题，请禁用逻辑应用资源，进行更改，然后重新启用逻辑应用资源。

没有可用的会话或另一个接收方锁定它

有时，诸如完成消息或续订会话之类的操作会在管理连接器中产生以下错误：

{
  "status": 400,
  "error": {
    "message": "No session available to complete the message with the lock token 'ce440818-f26f-4a04-aca8-555555555555'."
  }
}

有时，基于会话的触发器可能会失败，托管连接器中出现以下错误：

{
  "status": 400,
  "error": {
    "message": "Communication with the Service Bus namespace 'xxxx' and 'yyyy' entity failed. The requested session 'zzzz' cannot be accepted. It may be locked by another receiver."
  }
}

有时，诸如完成消息或更新会话的操作会在内置连接器中生成以下错误：

{
  "code": "ServiceProviderActionFailed",
  "message": "The service provider action failed with error code 'ServiceOperationFailed' and error message 'The Service Bus session was not found to perform operation 'getMessagesFromQueueSession' on session id '11115555'.'."
}

服务总线连接器使用内存中缓存来支持与会话关联的所有操作。 服务总线消息接收器缓存在接收消息的角色实例（虚拟机）的内存中。 为了处理所有请求，对连接的所有调用都路由到同一角色实例。 此行为是必需的，因为会话中的所有服务总线操作都需要接收特定会话消息的同一接收器。

由于基础结构更新、连接器部署等原因，请求可能无法路由到同一角色实例。 如果发生此事件，请求会失败，原因包括以下几种：

• 为请求提供服务的角色实例中不包含在会话中执行操作的接收器。

• 新角色实例尝试获取会话，该会话在旧角色实例中已超时或未关闭。

此行为可以在托管连接器和内置连接器中发生。 发生错误时，消息仍保留在服务总线中。 下一个触发器或工作流运行会尝试再次处理该消息。 只要这种错误只是偶尔发生，这是可以预期的。

相关内容

• Azure Logic Apps 中的托管连接器
• Azure Logic Apps 中的内置连接器
• 什么是 Azure 逻辑应用中的连接器