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

适用范围：Azure 逻辑应用（消耗型 + 标准型）

本指南介绍如何使用服务总线连接器从 Azure 逻辑应用中的工作流访问 Azure 服务总线。 然后，你可以创建在由服务总线中的事件触发后运行的自动化工作流，或者运行操作来管理服务总线项，例如：

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

可以使用触发器从 Azure 服务总线获取响应，并使输出可供工作流中的其他操作使用。 还可以让其他操作使用服务总线操作的输出。

连接器技术参考

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

逻辑应用	环境	连接器版本
消耗	多租户 Azure 逻辑应用	托管连接器，该连接器显示在连接器库中“运行时”>“共享”下。 注意：服务总线托管连接器触发器遵循长轮询触发器模式，这意味着触发器会定期检查队列或主题订阅中的消息。 有关详细信息，请查看以下文档： - 服务总线托管连接器参考 - Azure 逻辑应用中的托管连接器
标准	单租户 Azure 逻辑应用和应用服务环境 v3（仅限 Windows 计划）	托管连接器（Azure 托管）显示在连接器库中的“运行时”>“共享”下，而内置连接器显示在连接器库中的“运行时”>“应用内”下，并且是基于服务提供商的。 服务总线托管连接器触发器遵循长轮询触发器模式，这意味着触发器会定期检查队列或主题订阅中的消息。 服务总线内置连接器的非会话触发器遵循完全由连接器管理的连续轮询触发模式。 此模式会让触发器不断检查队列或主题订阅中的消息。 会话触发器遵循长轮询触发器模式，但其配置受名为 clientRetryOptions:tryTimeout 的 Azure Functions 设置的约束。 内置版本通常提供更好的性能、更全面的功能以及更实惠的价格。 有关详细信息，请查看以下文档： - 服务总线托管连接器参考 - 服务总线内置连接器操作 - Azure 逻辑应用中的内置连接器

先决条件

• Azure 帐户和订阅。 如果没有 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，至多为触发器的最大消息计数属性指定的消息数。

  注意

  自动完成触发器会自动完成消息，但只有在下一次调用服务总线时才会完成。 此行为可能会影响工作流设计。 例如，应避免更改自动完成触发器的并发性，因为如果工作流进入受限制状态，此更改可能会导致重复的消息。 更改并发性控制会创建以下条件：

  • 使用 WorkflowRunInProgress 代码跳过受限制的触发器。

  • 完成操作不会运行。

  • 下一个触发器运行发生在轮询间隔之后。

  必须将服务总线锁定持续时间设置为比轮询间隔更长的值。 但即使存在此设置，如果工作流在下一次轮询间隔内仍保持受限制状态，也仍可能无法完成消息传递。

  但是，如果打开服务总线触发器的并发性设置，则 maximumWaitingRuns 属性的默认值为 10。 根据服务总线实体的锁定持续时间设置和工作流的运行持续时间，此默认值可能太大，并可能导致“锁丢失”异常。 若要为方案找到最佳值，请对 maximumWaitingRuns 属性使用值 1 或值 2 开始测试。 若要更改最大等待运行值，请查看更改等待运行限制。

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

对于服务总线内置连接器，非会话触发器遵循完全由连接器管理的连续轮询触发模式。 此模式会让触发器不断检查队列或主题订阅中的消息。 会话触发器遵循长轮询触发器模式，而其配置受名为 clientRetryOptions:tryTimeout 的 Azure Functions 设置的约束。 目前，服务总线内置触发器的配置设置在 Azure Functions 主机扩展和触发器设置之间共享。其中，该扩展在逻辑应用的 host.json 文件中定义，而触发器设置在逻辑应用的工作流中定义，你可通过设计器或代码视图设置这些设置。 本部分介绍这两个设置位置。

• 在标准工作流中，某些触发器（例如，“消息在队列中可用时”触发器）可能会返回一条或多条消息。 这些触发器触发时，会返回 1 到若干条消息。 对于此类型的触发器，在不支持“最大消息计数”参数的情况下，你仍可使用 host.json 文件中的 maxMessageBatchSize 属性控制收到的消息数。 若要查找此文件，请参阅编辑标准逻辑应用的主机和应用设置。

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

• 还可通过设计器或在代码中对服务总线触发器启用并发：

  "runtimeConfiguration": {
      "concurrency": {
          "runs": 100
      }
  }
  

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

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

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

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

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

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

  • 请减少为检查应用并发的运行数而发送的存储调用数量。

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

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

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

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

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

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

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

   

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

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

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

身份验证类型	必需的信息
访问密钥	服务总线命名空间的连接字符串。 有关详细信息，请查看获取服务总线命名空间的连接字符串
Microsoft Entra 已集成	服务总线命名空间的终结点 URL。 有关详细信息，请查看获取服务总线命名空间的终结点 URL。
逻辑应用托管标识	服务总线命名空间的终结点 URL。 有关详细信息，请查看获取服务总线命名空间的终结点 URL。

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

身份验证类型	必需的信息
连接字符串	服务总线命名空间的连接字符串。 有关详细信息，请查看获取服务总线命名空间的连接字符串
Active Directory OAuth	- 服务总线命名空间的完全限定名称，例如 <your-Service-Bus-namespace>.servicebus.chinacloudapi.cn。 有关详细信息，请查看获取服务总线命名空间的完全限定名称。 有关其他属性值，请参阅具有 Microsoft Entra ID 的 OAuth。
托管的标识	服务总线命名空间的完全限定名称，例如 <your-Service-Bus-namespace>.servicebus.chinacloudapi.cn。 有关详细信息，请查看获取服务总线命名空间的完全限定名称。

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

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

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

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

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

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

   

   注意

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

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

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

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

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

2. 在命名空间菜单中的“设置”下，选择“属性”。

3. 在“属性”下的“服务总线终结点”旁边，复制终结点 URL，然后进行保存，供以后在必须提供服务总线终结点 URL 时使用。

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

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

2. 在命名空间菜单中选择“概述”。

3. 在“概述”窗格中找到“主机名”属性，然后复制完全限定名称，该名称类似于 <your-Service-Bus-namespace>.servicebus.chinacloudapi.cn。

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

以下步骤使用 Azure 门户，但利用适当的 Azure 逻辑应用扩展，也可以使用以下工具来创建逻辑应用工作流：

• 消耗型逻辑应用工作流：Visual Studio 或 Visual Studio Code

• 标准逻辑应用工作流：Visual Studio Code

步骤 3：选项 2 - 添加服务总线操作

以下步骤使用 Azure 门户，但利用适当的 Azure 逻辑应用扩展，也可以使用以下工具来创建逻辑应用工作流：

• 消耗型逻辑应用工作流：Visual Studio 或 Visual Studio Code

• 标准逻辑应用工作流：Visual Studio Code

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

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

使用服务总线内置触发器从死信队列中读取消息

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

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

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

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

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

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

只要这种错误只是偶尔发生，就会出现错误。 发生错误时，消息仍保留在服务总线中。 下一个触发器或工作流运行会尝试再次处理该消息。

后续步骤

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