Azure.Messaging.EventHubs 5.11.5

Azure事件中心.NET客户端库

Azure事件中心是一个高度可扩展的发布/订阅服务，每秒可以处理数百万个事件并将它们流式传输到多个消费者。这使您能够处理和分析由您的连接设备和应用程序产生的海量数据。一旦事件中心收集了数据，您就可以使用任何实时分析提供商或批处理/存储适配器检索、转换和存储它。如果您想了解更多关于Azure事件中心的信息，您可以查阅：什么是事件中心。

Azure事件中心客户端库可发布和消费Azure事件中心事件，可用于

• 发布有关您的应用程序的遥测数据，用于商业智能和诊断目的。

• 发布有关您应用程序状态的有关事实，感兴趣的各方可以观察并用作采取行动的触发器。

• 观察您业务或其他生态系统中发生的有趣操作和交互，允许松散耦合的系统在不将它们绑定在一起的情况下进行交互。

• 接收来自一个或多个发布者的事件，将它们转换以满足您生态系统更好的需求，然后将转换后的事件发布到新的流以便消费者观察。

源代码 | 包(NuGet) | API参考文档 | 产品文档 | 迁移指南 | 故障排除指南

入门

先决条件

• Azure订阅：要使用包括Azure事件中心在内的Azure服务，您需要订阅。如果您没有现有的Azure账户，您可以通过注册一个免费试用或使用您Visual Studio订阅的好处来创建一个账户。

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

• C# 8.0：Azure事件中心客户端库利用了C# 8.0中引入的新功能。为了利用C# 8.0语法，建议使用3.0或更高版本的.NET Core SDK编译，并设置语言版本为latest。

  希望充分利用C# 8.0语法的Visual Studio用户需要使用Visual Studio 2019或更高版本。包括免费社区版在内的Visual Studio 2019可以在此处下载。Visual Studio 2017用户可以通过使用Microsoft.Net.Compilers NuGet包并设置语言版本来利用C# 8语法，尽管编辑体验可能不理想。

  您仍然可以使用与以前的C#语言版本兼容的库，但需要手动管理异步可枚举和异步可释放成员，而不是从新语法中受益。您可以为您的.NET Core SDK支持的任何框架版本进行编译，包括.NET Core或.NET框架的早期版本。有关更多信息，请参阅：如何指定目标框架。
  重要说明：为了在不受修改的情况下构建或运行示例和样本，需要C# 11.0。您仍然可以运行样本，如果您决定为其他语言版本进行修改。此操作的一个示例在样本中提供：早期语言版本。

为了快速在Azure中创建一组基本的Event Hubs资源并获取其连接字符串，您可以点击部署我们的示例模板。

安装包

使用NuGet安装Azure Event Hubs客户端库 (.NET)

dotnet add package Azure.Messaging.EventHubs

验证客户端

为了使Event Hubs客户端库与Event Hub交互，它需要了解如何连接和授权。最简单的方法是使用连接字符串，它是在创建Event Hubs命名空间时自动创建的。如果你不熟悉使用Event Hubs的连接字符串，可以遵循以下获取Event Hubs连接字符串的步骤指南。

获取连接字符串后，可以使用它创建任何类型的Event Hubs客户端

var connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using var producer = new EventHubProducerClient(connectionString, eventHubName);

有关使用凭证类型验证Event Hubs客户端的示例，请参阅使用Azure Active Directory (AAD) 主实体或身份和共享访问凭证示例。

有关验证用于ASP.NET Core应用程序的Event Hubs客户端的示例，请参阅使用ASP.NET Core依赖注入进行注册。

关键概念

• Event Hub客户端是开发者与Event Hubs客户端库交互的主要接口。有几种不同的Event Hub客户端，每个客户端都针对Event Hubs的特定用途，如发布或消费事件。

• Event Hub生产者是客户端类型，它作为遥测数据、诊断信息、使用日志或其他日志数据的来源，可以作为嵌入式设备解决方案、移动设备应用程序、在控制台或其他设备上运行的游戏标题、某些基于客户端或服务器的业务解决方案或网站的组成部分。

• Event Hub消费者是从Event Hub读取信息并将其处理的一种客户端类型。处理可能包括聚合、复杂计算和过滤。处理也可能包括以原始或转换形式分发或存储信息。Event Hub消费者通常是坚固且高扩展性平台基础设施组件，内置分析功能，如Azure Stream Analytics、Apache Spark或Apache Storm。

• 分区是保存Event Hub中的事件的有序序列。分区是关联事件消费者所需并行性的数据组织方法。Azure Event Hubs通过分区的消费者模式提供消息流，其中每个消费者只读取消息流的一个特定子集或分区。随着较新事件的到达，它们被添加到此序列的末尾。分区数量在创建Event Hub时指定，无法更改。

• 消费者组是Event Hub的整个视图。消费者组允许多个消费应用程序各自拥有事件流的不同视图，并以它们自己的速度和位置独立读取流。每个消费者组的分区上最多可以有5个并发读取器；然而，建议每个分区和消费者组配对只有一个活跃的消费者。每个活跃的读取器接收其分区的所有事件；如果在同一分区上有多个读取器，则它们将接收到重复的事件。

有关更多的概念和深入讨论，请参阅：Event Hubs功能。

客户端生命周期

事件中心客户端的每种类型都可以安全地缓存并在应用程序的生命周期内用作单例，这在定期发布或读取事件时是最佳实践。客户端负责高效管理网络、CPU和内存使用，以在非活跃时段保持使用率低。在客户端上调用 CloseAsync 或 DisposeAsync 是必要的，以确保网络资源和其他非托管对象被正确清理。

线程安全

我们保证所有客户端实例方法是线程安全的，并且互不依赖（参考指南）。这确保了重用客户端实例的建议总是安全的，即使在跨线程的情况下。

数据模型类型，例如 EventData 和 EventDataBatch，不是线程安全的。它们不应在线程间共享或与客户端方法并发使用。

其他概念

客户端选项 | 处理故障 | 诊断 | 模拟

示例

检查事件中心

许多事件中心操作都在特定分区的范围内进行。因为分区归事件中心所有，所以它们的名称在创建时被分配。要了解可用的分区，您可以使用事件中心客户端查询事件中心。为了说明，下面的示例将演示 EventHubProducerClient，但这一概念和形式在客户端中是通用的。

var connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using (var producer = new EventHubProducerClient(connectionString, eventHubName))
{
    string[] partitionIds = await producer.GetPartitionIdsAsync();
}

向事件中心发布事件

为了发布事件，您需要创建一个 EventHubProducerClient。生产者以批量的形式发布事件，可以请求特定的分区，或者允许事件中心服务决定将事件发布到哪个分区。在需要高度可用性或希望事件数据在分区之间均匀分布的情况下，建议使用自动路由。我们的示例将利用自动路由。

var connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using (var producer = new EventHubProducerClient(connectionString, eventHubName))
{
    using EventDataBatch eventBatch = await producer.CreateBatchAsync();

    if ((!eventBatch.TryAdd(new EventData("First"))) ||
        (!eventBatch.TryAdd(new EventData("Second"))))
    {
       throw new ApplicationException("Not all events could be added to the batch!");
    }

    await producer.SendAsync(eventBatch);
}

从事件中心读取事件

要从事件中心读取事件，您需要为特定的消费者组创建一个 EventHubConsumerClient。当事件中心创建时，它提供一个默认的消费者组，可以用来开始探索事件中心。在我们的示例中，我们将使用迭代器关注读取事件中心中已发布的所有事件。

注意：请注意，这种消费方法的目的是为了改善探索事件中心客户端库和原型设计时的体验。不建议在生产场景中使用。对于生产使用，我们建议使用事件处理器客户端，因为它提供了一种更健壮和性能更优的体验。

var connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";

string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using (var consumer = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName))
{
    using var cancellationSource = new CancellationTokenSource();
    cancellationSource.CancelAfter(TimeSpan.FromSeconds(45));

    await foreach (PartitionEvent receivedEvent in consumer.ReadEventsAsync(cancellationSource.Token))
    {
        // At this point, the loop will wait for events to be available in the Event Hub.  When an event
        // is available, the loop will iterate with the event that was received.  Because we did not
        // specify a maximum wait time, the loop will wait forever unless cancellation is requested using
        // the cancellation token.
    }
}

从事件中心分区读取事件

为了从事件中心分区读取事件，您需要为特定的消费者组创建一个 EventHubConsumerClient。当事件中心创建时，它提供一个默认的消费者组，可以用来开始探索事件中心。要读取特定分区，消费者还需要指定从事件流中的哪个位置开始接收事件；在我们的示例中，我们将关注读取事件中心第一个分区的所有已发布事件。

var connectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";

string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using (var consumer = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName))
{
    EventPosition startingPosition = EventPosition.Earliest;
    string partitionId = (await consumer.GetPartitionIdsAsync()).First();

    using var cancellationSource = new CancellationTokenSource();
    cancellationSource.CancelAfter(TimeSpan.FromSeconds(45));

    await foreach (PartitionEvent receivedEvent in consumer.ReadEventsFromPartitionAsync(partitionId, startingPosition, cancellationSource.Token))
    {
        // At this point, the loop will wait for events to be available in the partition.  When an event
        // is available, the loop will iterate with the event that was received.  Because we did not
        // specify a maximum wait time, the loop will wait forever unless cancellation is requested using
        // the cancellation token.
    }
}

使用事件处理器客户端处理事件

对于大多数生产场景，建议使用事件处理器客户端来读取和处理事件。该处理器旨在以高性能和容错的方式处理Event Hub的所有分区的事件，同时提供持久化其状态的手段。事件处理器客户端还可以在消费组的上下文中协同工作，针对特定的Event Hub，它们将自动管理工作实例的分配和平衡。

由于EventProcessorClient依赖于Azure存储blob来持久化其状态，因此您需要为处理器提供BlobContainerClient，该客户端已针对应使用的存储帐户和容器进行配置。

var cancellationSource = new CancellationTokenSource();
cancellationSource.CancelAfter(TimeSpan.FromSeconds(45));

var storageConnectionString = "<< CONNECTION STRING FOR THE STORAGE ACCOUNT >>";
var blobContainerName = "<< NAME OF THE BLOB CONTAINER >>";

var eventHubsConnectionString = "<< CONNECTION STRING FOR THE EVENT HUBS NAMESPACE >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var consumerGroup = "<< NAME OF THE EVENT HUB CONSUMER GROUP >>";

Task processEventHandler(ProcessEventArgs eventArgs) => Task.CompletedTask;
Task processErrorHandler(ProcessErrorEventArgs eventArgs) => Task.CompletedTask;

var storageClient = new BlobContainerClient(storageConnectionString, blobContainerName);
var processor = new EventProcessorClient(storageClient, consumerGroup, eventHubsConnectionString, eventHubName);

processor.ProcessEventAsync += processEventHandler;
processor.ProcessErrorAsync += processErrorHandler;

await processor.StartProcessingAsync();

try
{
    // The processor performs its work in the background; block until cancellation
    // to allow processing to take place.

    await Task.Delay(Timeout.Infinite, cancellationSource.Token);
}
catch (TaskCanceledException)
{
    // This is expected when the delay is canceled.
}

try
{
    await processor.StopProcessingAsync();
}
finally
{
    // To prevent leaks, the handlers should be removed when processing is complete.

    processor.ProcessEventAsync -= processEventHandler;
    processor.ProcessErrorAsync -= processErrorHandler;
}

更多详细信息请参阅事件处理器客户端的README和相关的示例。

使用Active Directory主体与事件处理器客户端

Azure身份库提供Azure Active Directory (AAD)身份验证支持，可用于Azure客户端库，包括事件处理器。

要使用Active Directory主体，在创建事件处理器客户端时，指定从Azure.Identity库中可用的凭证之一。此外，提供完全限定的Event Hub命名空间和所需Event Hub的名称，而不是Event Hub连接字符串。例如，以下示例演示了EventHubProducerClient，但概念和形式对所有客户端都是通用的。

var fullyQualifiedNamespace = "<< FULLY-QUALIFIED EVENT HUBS NAMESPACE (like something.servicebus.windows.net) >>";
var eventHubName = "<< NAME OF THE EVENT HUB >>";
var credential = new DefaultAzureCredential();

// It is recommended that you cache the Event Hubs clients for the lifetime of your
// application, closing or disposing when application ends.  This example disposes
// after the immediate scope for simplicity.

await using (var producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential))
{
    using EventDataBatch eventBatch = await producer.CreateBatchAsync();

    if ((!eventBatch.TryAdd(new EventData("First"))) ||
        (!eventBatch.TryAdd(new EventData("Second"))))
    {
       throw new ApplicationException("Not all events could be added to the batch!");
    }

    await producer.SendAsync(eventBatch);
}

当使用Azure Active Directory时，您的主体必须分配一个允许访问Event Hub的角色，例如Azure Event Hubs Data Owner角色。有关使用Azure Active Directory对Event Hub进行授权的更多信息，请参阅相关文档。

使用ASP.NET Core依赖注入进行注册

要将事件处理器客户端之一作为依赖项注入到ASP.NET Core应用程序中，请安装Azure客户端库集成ASP.NET Core包。

dotnet add package Microsoft.Extensions.Azure

安装后，请在Startup.ConfigureServices方法中注册所需的Event Hub客户端类型。

public void ConfigureServices(IServiceCollection services)
{
    services.AddAzureClients(builder =>
    {
        builder.AddEventHubProducerClient(Configuration.GetConnectionString("EventHubs"));
    });
  
    services.AddControllers();
}

要使用前面的代码，将其添加到您应用程序的配置中。

{
  "ConnectionStrings": {
    "EventHubs": "<connection_string>"
  }
}

对于更喜欢使用共享Azure.Identity凭证的客户应用程序，注册方式略有不同。

var fullyQualifiedNamespace = "<< FULLY-QUALIFIED EVENT HUBS NAMESPACE (like something.servicebus.windows.net) >>";

public void ConfigureServices(IServiceCollection services)
{
    services.AddAzureClients(builder =>
    {
        // This will register the EventHubProducerClient using the default credential.
        builder.AddEventHubProducerClientWithNamespace(fullyQualifiedNamespace);

        // By default, DefaultAzureCredential is used, which is likely desired for most
        // scenarios. If you need to restrict to a specific credential instance, you could
        // register that instance as the default credential instead.
        builder.UseCredential(new ManagedIdentityCredential());
    });
  
    services.AddControllers();
}

有关更多详细信息，请参见使用.NET Azure SDK进行依赖注入。

故障排除

有关详细的故障排除信息，请参阅Event Hubs故障排除指南。

日志记录和诊断

事件处理器客户端库完全集成Logging信息，使用.NET EventSource的不同详细级别发出信息。对于每个操作都进行日志记录，并遵循标记操作开始点、完成情况以及遇到的任何异常的模式。此外，还记录了可能有助于洞察的附加信息。

Event Hubs 客户端日志可通过选择名为 "Azure-Messaging-EventHubs" 的源或选择具有 "AzureEventSource" 特性的所有源，对任何 EventListener 可用。为了使从 Azure 客户端库捕获日志变得更简单，Event Hubs 所使用的 Azure.Core 库提供了一个 AzureEventSourceListener。更多详细信息请参阅 使用 AzureEventSourceListener 捕获 Event Hubs 日志。

Event Hubs 客户端库还支持使用 Application Insights 或 OpenTelemetry 进行分布式跟踪。更多详细信息请参阅 Azure.Core Diagnostics 示例。

下一步

除了所讨论的入门级场景之外，Azure Event Hubs 客户端库还提供支持其他场景的选项，以帮助充分利用 Azure Event Hubs 服务功能。为了帮助用户探索其中一些场景，Event Hubs 客户端库提供了一个用于演示常见场景的示例项目。请参阅示例的 README 获取详细信息。

贡献

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

在提交拉取请求时，CLA-bot 会自动判断您是否需要提供 CLA，并适当地装饰 PR（例如，标签、注释）。只需按照机器人提供的说明操作。您只需在整个使用我们的 CLA 的所有存储库中这样做一次。

本项目已采用 Microsoft Open Source Code of Conduct。有关更多信息，请参阅 Code of Conduct FAQ 或通过 [email protected] 联系我们以获取任何额外的问题或评论。

请参阅我们的 贡献指南 获取更多信息。