Microsoft.Azure.WebJobs.Extensions.EventHubs 6.3.5

Azure WebJobs Event Hubs .NET客户端库

此扩展提供从Azure Function访问Azure Event Hubs的功能。

入门

安装包

使用NuGet安装Event Hubs扩展

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

先决条件

• Azure订阅： 要使用Azure服务，包括Azure Event Hubs，您需要一个订阅。如果您没有现有的Azure帐户，您可以在免费试用中注册或创建帐户时使用您的Visual Studio订阅福利。

• 包含事件中心的事件中心命名空间：要与 Azure 事件中心交互，您还需要一个命名空间和事件中心。如果您不熟悉创建 Azure 资源，您可能想遵循使用 Azure 门户创建事件中心的逐步指南。在那里，您还可以找到使用 Azure CLI、Azure PowerShell 或 Azure 资源管理器（ARM）模板创建事件中心的详细说明。

• 具有blob存储的 Azure 存储帐户：为了将检查点作为 Azure 存储中的 blob 持久化，您需要一个具有 blob 的 Azure 存储帐户。如果您不熟悉 Azure 存储帐户，您可能想遵循使用 Azure 门户创建存储帐户的逐步指南。在那里，您还可以找到使用 Azure CLI、Azure PowerShell 或 Azure 资源管理器（ARM）模板创建存储帐户的详细说明。

验证客户端

为了使事件中心客户端库能够与事件中心交互，它需要了解如何连接并授权。这样做最简单的方式是使用连接字符串，该字符串在创建事件中心命名空间时自动创建。如果您不熟悉使用事件中心中的连接字符串，您可能想遵循获取事件中心连接字符串的逐步指南。

EventHubAttribute 和 EventHubTriggerAttribute 的 Connection 属性用于指定存储连接字符串的配置属性。

使用 AzureWebJobsStorage 连接字符串来保留处理检查点信息。

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

{
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true",
    "<connection_name>": "Endpoint=sb://<event_hubs_namespace>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=Jya7Eh76HU92ibsxuk1ITN8CM8Bt76YLKf5ISjU3jZ8="
  }
}

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

基于身份的认证

如果您的环境启用了托管身份，则可以将其用于验证事件中心扩展。在这样做之前，您需要确保权限已按Azure Functions 开发者指南中描述的方式配置。

要使用基于身份的认证，请提供 <连接名称>__fullyQualifiedNamespace 配置设置。

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

或对于已部署的应用程序，请在该设置中应用程序设置中设置相同的设置。

<connection_name>__fullyQualifiedNamespace={event_hubs_namespace}.servicebus.windows.net

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

关键概念

事件中心触发器

事件中心触发器允许当消息被发送到事件中心时，执行一个函数。

请遵循Azure Event Hubs 触发教程来了解更多关于事件中心触发器的信息。

事件中心输出绑定

事件中心输出绑定允许函数发送事件中心事件。

请遵循Azure Event Hubs 输出绑定来了解更多关于事件中心绑定的信息。

支持的类型

以下类型支持触发器和输出绑定

• EventData
• string - 值将使用 UTF8 编码进行编码
• BinaryData
• byte[]
• 自定义模型类型将使用 Newtonsoft.Json 进行 JSON 序列化
• IAsyncCollector<T> 适用于以下任一类型的批量触发器
• EventHubProducerClient 用于输出绑定

示例

发送单个事件

您可以通过在函数的返回值上应用 EventHubAttribute 来将单个事件发送到事件中心。返回值可以是 string 或 EventData 类型。使用返回值时可能未指定分区键；要这样做，您需要绑定到 IAsyncCollector<EventData>，如发送多个事件中所示。

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

发送多个事件

要从单个 Azure Function 调用发送多个事件，您可以对 IAsyncCollector<string> 或 IAsyncCollector<EventData> 参数应用 EventHubAttribute。仅当绑定到 IAsyncCollector<EventData> 时才能使用分区键。

[FunctionName("BindingToCollector")]
public static async Task Run(
    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
    [EventHub("<event_hub_name>", Connection = "<connection_name>")] IAsyncCollector<EventData> collector)
{
    // When no partition key is used, partitions will be assigned per-batch via round-robin.
    await collector.AddAsync(new EventData($"Event 1 added at: {DateTime.Now}"));
    await collector.AddAsync(new EventData($"Event 2 added at: {DateTime.Now}"));

    // Using a partition key will help group events together; events with the same key
    // will always be assigned to the same partition.
    await collector.AddAsync(new EventData($"Event 3 added at: {DateTime.Now}"), "sample-key");
    await collector.AddAsync(new EventData($"Event 4 added at: {DateTime.Now}"), "sample-key");
}

使用强类型模型绑定

要使用与 EventHub 绑定的强类型模型类，请将 EventHubAttribute 应用到模型参数上。

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

使用 EventHubProducerClient 发送多个事件

您还可以直接绑定到 EventHubProducerClient 以实现发送事件的最大控制。

[FunctionName("BindingToProducerClient")]
public static async Task Run(
    [TimerTrigger("0 */5 * * * *")] TimerInfo myTimer,
    [EventHub("<event_hub_name>", Connection = "<connection_name>")] EventHubProducerClient eventHubProducerClient)
{
    // IAsyncCollector allows sending multiple events in a single function invocation
    await eventHubProducerClient.SendAsync(new[]
    {
        new EventData($"Event 1 added at: {DateTime.Now}"),
        new EventData($"Event 2 added at: {DateTime.Now}")
    });
}

事件级别的触发器

要将函数应用于每次向事件中心发送事件，请将 EventHubTriggerAttribute 应用到 string 或 EventData 参数。

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

批量触发器

要为接收到的事件批量运行函数，请将 EventHubTriggerAttribute 应用到 string[] 或 EventData[] 参数。

[FunctionName("TriggerBatch")]
public static void Run(
    [EventHubTrigger("<event_hub_name>", Connection = "<connection_name>")] EventData[] events,
    ILogger logger)
{
    foreach (var e in events)
    {
        logger.LogInformation($"C# function triggered to process a message: {e.EventBody}");
        logger.LogInformation($"EnqueuedTime={e.EnqueuedTime}");
    }
}

故障排除

请参阅监控 Azure Functions 获取故障排除指南。

下一步

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

贡献

有关构建、测试和向此库贡献的详细信息，请参阅CONTRIBUTING.md。

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

本项目采用了Microsoft Open Source Code of Conduct。有关更多信息，请参阅行为准则常见问题解答或联系[email protected]获得任何额外问题和反馈。