Microsoft.Azure.WebJobs.Extensions.ServiceBus 5.16.4

Azure WebJobs Service Bus客户端库，适用于.NET

此扩展提供了从Azure Function访问Azure Service Bus的功能。

入门

安装包

使用NuGet安装Service Bus扩展 NuGet

dotnet add package Microsoft.Azure.WebJobs.Extensions.ServiceBus

先决条件

• Azure 订阅：要使用 Azure 服务，包括 Azure 服务总线，您需要一个订阅。如果您没有现有的 Azure 账户，您可以考虑注册 免费试用 或在您创建账户时使用您的 Visual Studio 订阅 优惠。

• 服务总线命名空间：为了与 Azure 服务总线交互，您还需要有一个命名空间可用。如果您不熟悉创建 Azure 资源，您可能希望遵循使用 Azure 门户创建服务总线命名空间的分步指南。在那里，您还可以找到使用 Azure CLI、Azure PowerShell 或 Azure 资源管理器 (ARM) 模板创建服务总线实体的详细说明。

为了快速在 Azure 中创建所需的 Service Bus 资源并获取它们的连接字符串，您可以单击以下链接部署我们的示例模板

认证客户端

为了使服务总线客户端库能够与队列或主题交互，它需要了解如何与之连接和授权。最简单的方法是使用连接字符串，该字符串在创建服务总线命名空间时自动生成。如果您不熟悉 Azure 中的共享访问策略，您可能希望遵循获取服务总线连接字符串的分步指南。

ServiceBusAttribute 和 ServiceBusTriggerAttribute 的 Connection 属性用于指定存储连接字符串的配置属性。如果没有指定，则假定期望 AzureWebJobsServiceBus 属性包含连接字符串。

对于本地开发，请使用 local.settings.json 文件来存储连接字符串

{
  "Values": {
    "<connection_name>": "Endpoint=sb://<service_bus_namespace>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<access key>"
  }
}

当部署时，请使用应用程序设置来设置连接字符串。

基于身份的认证

如果您的环境已启用 托管身份，您可以使用它来认证服务总线扩展。在此之前，您将需要确保权限已经如Azure Functions 开发者指南中所描述的那样得到配置。要使用基于身份的认证，请提供 <connection_name>__fullyQualifiedNamespace 配置设置。

{
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "<connection_name>__fullyQualifiedNamespace": "<service_bus_namespace>.servicebus.windows.net"
  }
}

或者，在部署的应用中设置相同的设置在应用程序设置

<connection_name>__fullyQualifiedNamespace=<service_bus_namespace>.servicebus.windows.net

有关配置基于身份的连接的更多详细信息，请参阅此处。

关键概念

Service Bus 触发器

Service Bus 触发器允许当消息发送到 Service Bus 队列或主题时执行函数。

请遵循Azure Service Bus 触发器教程以了解更多关于 Service Bus 触发器的信息。

Service Bus 输出绑定

Service Bus 输出绑定允许函数发送 Service Bus 消息。

请遵循Azure Service Bus 输出绑定以了解更多关于 Service Bus 绑定的信息。

示例

发送单个消息

您可以通过将 ServiceBus 属性应用于函数返回值来将单个消息发送到队列或主题。返回值可以为 string、byte[] 或 ServiceBusMessage 类型。

[FunctionName("BindingToReturnValue")]
[return: ServiceBus("<queue_or_topic_name>", Connection = "<connection_name>")]
public static string BindToReturnValue([TimerTrigger("0 */5 * * * *")] TimerInfo myTimer)
{
    // This value would get stored in Service Bus message body.
    // The string would be UTF8 encoded.
    return $"C# Timer trigger function executed at: {DateTime.Now}";
}

您还可以使用类型为 string、byte[] 或 ServiceBusMessage 的 out 参数。

[FunctionName("BindingToOutputParameter")]
public static void Run(
[TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
[ServiceBus("<queue_or_topic_name>", Connection = "<connection_name>")] out ServiceBusMessage message)
{
    message = new ServiceBusMessage($"C# Timer trigger function executed at: {DateTime.Now}");
}

发送多条消息

要从单个 Azure Function 调用中发送多条消息，可以将 ServiceBus 属性应用到 IAsyncCollector<string> 或 IAsyncCollector<ServiceBusReceivedMessage> 参数。

[FunctionName("BindingToCollector")]
public static async Task Run(
    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
    [ServiceBus("<queue_or_topic_name>", Connection = "<connection_name>")] IAsyncCollector<ServiceBusMessage> collector)
{
    // IAsyncCollector allows sending multiple messages in a single function invocation
    await collector.AddAsync(new ServiceBusMessage(new BinaryData($"Message 1 added at: {DateTime.Now}")));
    await collector.AddAsync(new ServiceBusMessage(new BinaryData($"Message 2 added at: {DateTime.Now}")));
}

使用强类型模型进行绑定

要使用与 ServiceBus 绑定的强类型模型类，请将 ServiceBus 属性应用到模型参数。这样做将尝试将 ServiceBusMessage.Body 反序列化为强类型模型。

[FunctionName("TriggerSingleModel")]
public static void Run(
    [ServiceBusTrigger("<queue_name>", Connection = "<connection_name>")] Dog dog,
    ILogger logger)
{
    logger.LogInformation($"Who's a good dog? {dog.Name} is!");
}

使用 ServiceBusSender 发送多条消息

您还可以直接绑定到 ServiceBusSender，以拥有对消息发送的最大控制权。

[FunctionName("BindingToSender")]
public static async Task Run(
    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
    [ServiceBus("<queue_or_topic_name>", Connection = "<connection_name>")] ServiceBusSender sender)
{
    await sender.SendMessagesAsync(new[]
    {
        new ServiceBusMessage(new BinaryData($"Message 1 added at: {DateTime.Now}")),
        new ServiceBusMessage(new BinaryData($"Message 2 added at: {DateTime.Now}"))
    });
}

每条消息触发器

要为每次将消息发送到 Service Bus 队列或订阅而运行函数，请将 ServiceBusTrigger 属性应用到 string、byte[] 或 ServiceBusReceivedMessage 参数。

[FunctionName("TriggerSingle")]
public static void Run(
    [ServiceBusTrigger("<queue_name>", Connection = "<connection_name>")] string messageBodyAsString,
    ILogger logger)
{
    logger.LogInformation($"C# function triggered to process a message: {messageBodyAsString}");
}

批处理触发器

要为一批接收到的消息运行函数，请将 ServiceBusTrigger 属性应用到 string[] 或 ServiceBusReceivedMessage[] 参数。

[FunctionName("TriggerBatch")]
public static void Run(
    [ServiceBusTrigger("<queue_name>", Connection = "<connection_name>")] ServiceBusReceivedMessage[] messages,
    ILogger logger)
{
    foreach (ServiceBusReceivedMessage message in messages)
    {
        logger.LogInformation($"C# function triggered to process a message: {message.Body}");
        logger.LogInformation($"EnqueuedTime={message.EnqueuedTime}");
    }
}

消息结算

您可以使用 ServiceBusOptions 来配置消息在您的函数执行后自动完成。如果您想对消息结算有更多控制权，可以使用与每条消息和批处理触发器绑定的 MessageActions。

[FunctionName("BindingToMessageActions")]
public static async Task Run(
    [ServiceBusTrigger("<queue_name>", Connection = "<connection_name>")]
    ServiceBusReceivedMessage[] messages,
    ServiceBusMessageActions messageActions)
{
    foreach (ServiceBusReceivedMessage message in messages)
    {
        if (message.MessageId == "1")
        {
            await messageActions.DeadLetterMessageAsync(message);
        }
        else
        {
            await messageActions.CompleteMessageAsync(message);
        }
    }
}

会话触发器

要从启用会话的队列或主题接收消息，您可以在 ServiceBusTrigger 属性上设置 IsSessionsEnabled 属性。在与会话一起工作时，您可以绑定到 SessionMessageActions 以访问消息结算方法，以及会话特定的功能。

[FunctionName("BindingToSessionMessageActions")]
public static async Task Run(
    [ServiceBusTrigger("<queue_name>", Connection = "<connection_name>", IsSessionsEnabled = true)]
    ServiceBusReceivedMessage[] messages,
    ServiceBusSessionMessageActions sessionActions)
{
    foreach (ServiceBusReceivedMessage message in messages)
    {
        if (message.MessageId == "1")
        {
            await sessionActions.DeadLetterMessageAsync(message);
        }
        else
        {
            await sessionActions.CompleteMessageAsync(message);
        }
    }

    // We can also perform session-specific operations using the actions, such as setting state that is specific to this session.
    await sessionActions.SetSessionStateAsync(new BinaryData("<session state>"));
}

绑定到 ReceiveActions

您可以在您的函数调用中接收额外的消息。这在您需要根据传送到您函数的初始消息的某些特征来控制函数调用中要处理的更多消息时可能会有所帮助。您收到的任何附加消息都将受到与初始传送到您的函数的消息相同的 AutoCompleteMessages 和 MaxAutoLockRenewalDuration 配置的影响。您还可以查看消息。已查看的消息不受 AutoCompleteMessages 和 MaxAutoLockRenewalDuration 配置的限制，因为这些消息未被锁定，因此不能完成。

[FunctionName("BindingToReceiveActions")]
public static async Task Run(
    [ServiceBusTrigger("<queue_name>", Connection = "<connection_name>", IsSessionsEnabled = true)]
    ServiceBusReceivedMessage message,
    ServiceBusMessageActions messageActions,
    ServiceBusReceiveActions receiveActions)
{
    if (message.MessageId == "1")
    {
        await messageActions.DeadLetterMessageAsync(message);
    }
    else
    {
        await messageActions.CompleteMessageAsync(message);

        // attempt to receive additional messages in this session
        var receivedMessages = await receiveActions.ReceiveMessagesAsync(maxMessages: 10);

        // you can also use the receive actions to peek messages
        var peekedMessages = await receiveActions.PeekMessagesAsync(maxMessages: 10);
    }
}

绑定到 ServiceBusClient

在某些情况下，您可能希望绑定到与触发器使用相同的 ServiceBusClient。这在您需要根据接收到的消息动态创建发送者时很有用。

[FunctionName("BindingToClient")]
public static async Task Run(
    [ServiceBus("<queue_or_topic_name>", Connection = "<connection_name>")]
    ServiceBusReceivedMessage message,
    ServiceBusClient client)
{
    ServiceBusSender sender = client.CreateSender(message.To);
    await sender.SendMessageAsync(new ServiceBusMessage(message));
}

故障排除

如果您的函数触发了未处理的异常，并且您尚未解决消息，扩展将尝试弃用消息，以便它立即再次可用于接收。

有关更多故障排除指导，请参阅 监控 Azure Functions。

下一步

阅读 Azure Functions 简介 或 创建 Azure Function 指南。

贡献

有关构建、测试和对该库做出贡献的详细信息，请参阅我们的 CONTRIBUTING.md。

本项目欢迎贡献和建议。大多数贡献都需要您同意一项贡献者许可协议 (CLA)，声明您有权，并且确实授予我们使用您贡献的权利。有关详细信息，请访问 cla.microsoft.com。

本项目采用 Microsoft 开源行为准则。有关更多信息，请参阅 行为准则常见问题解答 或联系 [email protected] 就任何附加问题或评论。