快速入门：使用 .NET 向/从 Azure 事件中心发送/接收事件

将角色分配到 Microsoft Entra 用户帐户

在本地开发时，请确保连接到 Azure 事件中心的用户帐户具有正确的权限。 你需要拥有 Azure 事件中心数据所有者角色才能发送和接收消息。 若要为自己分配此角色，需要具有“用户访问管理员”角色，或者具有包含 Microsoft.Authorization/roleAssignments/write 操作的其他角色。 可使用 Azure 门户、Azure CLI 或 Azure PowerShell 向用户分配 Azure RBAC 角色。 可在范围概述页上详细了解角色分配的可用范围。

以下示例将 Azure Event Hubs Data Owner 角色分配给用户帐户，该角色提供对 Azure 事件中心资源的完全访问权限。 在实际方案中，遵循最小特权原则，仅向用户提供更安全的生产环境所需的最小权限。

Azure 事件中心的内置 Azure 角色

对于 Azure 事件中心，通过 Azure 门户和 Azure 资源管理 API 对命名空间和所有相关资源进行的管理已使用 Azure RBAC 模型进行了保护。 Azure 提供以下 Azure 内置角色，用于授予对事件中心命名空间的访问权限：

• Azure 事件中心数据所有者：允许对事件中心命名空间及其实体（队列、主题、订阅和筛选器）进行数据访问
• Azure 事件中心数据发送者：使用此角色授予对事件中心命名空间及其实体的发送者访问权限。
• Azure 事件中心数据接收者：使用此角色授予对事件中心命名空间及其实体的接收者访问权限。

如果要创建自定义角色，请参阅执行事件中心操作所需的权限。

az servicebus namespace show -g '<your-event-hub-resource-group>' -n '<your-event-hub-name> --query id

az role assignment create --assignee "<user@domain>" \
--role "Azure Event Hubs Data Owner" \
--scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<your-event-hub-resource-group>" -Name "<your-event-hub-name>"

New-AzRoleAssignment -SignInName <user@domain> `
-RoleDefinitionName "Azure Event Hubs Data Owner" `
-Scope <yourStorageAccountId>

启动 Visual Studio 并登录到 Azure

可以使用以下步骤授权对服务总线命名空间的访问：

1. 启动 Visual Studio。 如果看到“入门”窗口，请在右窗格中选择“在不使用代码的情况下继续”链接。

2. 选择 Visual Studio 右上角的“登录”按钮。

   

3. 使用你之前为其分配角色的 Microsoft Entra 帐户登录。

   

获取连接字符串

创建新的命名空间会自动生成一个初始共享访问签名 (SAS) 策略，该策略包含主密钥和辅助密钥以及连接字符串，每个连接字符串都授予对命名空间所有方面的完全控制权。 请参阅事件中心身份验证和授权，了解如何创建规则来对普通发送者和接收者的权限进行更多限制。

客户端可以使用连接字符串连接到事件中心命名空间。 若要复制命名空间的主要连接字符串，请执行以下步骤：

1. 在“事件中心命名空间”页上的左侧菜单中选择“共享访问策略”。

2. 在“共享访问策略”页，选择“RootManageSharedAccessKey” 。

3. 在“策略: RootManageSharedAccessKey”窗口中，选择“主连接字符串”旁边的“复制”按钮，将连接字符串复制到剪贴板供以后使用。 将此值粘贴到记事本或其他某个临时位置。

   

   可使用此页面复制主密钥、辅助密钥、主连接字符串和辅助连接字符串。

1. 在菜单中选择“工具”>“NuGet 包管理器”>“包管理器控制台”。

2. 运行以下命令以安装 Azure.Messaging.EventHubs 和 Azure.Identity NuGet 包。 按 ENTER 运行第二个命令。

   Install-Package Azure.Messaging.EventHubs
   Install-Package Azure.Identity
   

1. 在菜单中选择“工具”>“NuGet 包管理器”>“包管理器控制台”。

2. 运行以下命令安装 Azure.Messaging.EventHubs NuGet 包：

   Install-Package Azure.Messaging.EventHubs
   

1. 将 Program.cs 文件中的现有代码替换为以下示例代码。 然后，将 EventHubProducerClient 参数的 <EVENT_HUB_NAMESPACE> 和 <HUB_NAME> 占位符值替换为事件中心命名空间和事件中心的名称。 例如 "spehubns0309.servicebus.chinacloudapi.cn" 和 "spehub"。

   下面是代码中的重要步骤：

   1. 使用命名空间和事件中心名称创建一个 EventHubProducerClient 对象。
   2. 对 EventHubProducerClient 对象调用 CreateBatchAsync 方法，创建一个 EventDataBatch 对象。
   3. 使用 EventDataBatch.TryAdd 方法将事件添加到批处理中。
   4. 使用 EventHubProducerClient.SendAsync 方法将这批消息发送到事件中心。

   using Azure.Identity;
   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Producer;
   using System.Text;
   
   // number of events to be sent to the event hub
   int numOfEvents = 3;
   
   // The Event Hubs client types are safe to cache and use as a singleton for the lifetime
   // of the application, which is best practice when events are being published or read regularly.
   // TODO: Replace the <EVENT_HUB_NAMESPACE> and <HUB_NAME> placeholder values
   EventHubProducerClient producerClient = new EventHubProducerClient(
       "<EVENT_HUB_NAMESPACE>.servicebus.chinacloudapi.cn",
       "<HUB_NAME>",
       new DefaultAzureCredential());
   
   // Create a batch of events 
   using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();
   
   for (int i = 1; i <= numOfEvents; i++)
   {
       if (!eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes($"Event {i}"))))
       {
           // if it is too large for the batch
           throw new Exception($"Event {i} is too large for the batch and cannot be sent.");
       }
   }
   
   try
   {
       // Use the producer client to send the batch of events to the event hub
       await producerClient.SendAsync(eventBatch);
       Console.WriteLine($"A batch of {numOfEvents} events has been published.");
       Console.ReadLine();
   }
   finally
   {
       await producerClient.DisposeAsync();
   }
   

1. 将 Program.cs 文件中的现有代码替换为以下示例代码。 然后，替换 EventHubProducerClient 参数的 <CONNECTION_STRING> 和 <HUB_NAME> 占位符值。

   下面是代码中的重要步骤：

   1. 使用命名空间的主连接字符串和事件中心名称创建 EventHubProducerClient 对象。
   2. 对 EventHubProducerClient 对象调用 CreateBatchAsync 方法，创建一个 EventDataBatch 对象。
   3. 使用 EventDataBatch.TryAdd 方法将事件添加到批处理中。
   4. 使用 EventHubProducerClient.SendAsync 方法将这批消息发送到事件中心。

   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Producer;
   using System.Text;
   
   // number of events to be sent to the event hub
   int numOfEvents = 3;
   
   // The Event Hubs client types are safe to cache and use as a singleton for the lifetime
   // of the application, which is best practice when events are being published or read regularly.
   // TODO: Replace the <CONNECTION_STRING> and <HUB_NAME> placeholder values
   EventHubProducerClient producerClient = new EventHubProducerClient(
       "<CONNECTION_STRING>",
       "<HUB_NAME>");
   
   // Create a batch of events 
   using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();
   
   for (int i = 1; i <= numOfEvents; i++)
   {
       if (!eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes($"Event {i}"))))
       {
           // if it is too large for the batch
           throw new Exception($"Event {i} is too large for the batch and cannot be sent.");
       }
   }
   
   try
   {
       // Use the producer client to send the batch of events to the event hub
       await producerClient.SendAsync(eventBatch);
       Console.WriteLine($"A batch of {numOfEvents} events has been published.");
       Console.ReadLine();
   }
   finally
   {
       await producerClient.DisposeAsync();
   }
   

在本地开发时，请确保访问 Blob 数据的用户帐户具有正确的权限。 需有“存储 Blob 数据参与者”角色才能读取和写入 Blob 数据。 若要为你自己分配此角色，需要具有“用户访问管理员”角色，或者具有包含 Microsoft.Authorization/roleAssignments/write 操作的其他角色。 可使用 Azure 门户、Azure CLI 或 Azure PowerShell 向用户分配 Azure RBAC 角色。 可以在范围概述页上详细了解角色分配的可用范围。

在此方案中，你将为用户帐户分配权限（范围为存储帐户）以遵循最低权限原则。 这种做法仅为用户提供所需的最低权限，并创建更安全的生产环境。

以下示例将“存储 Blob 数据参与者”角色分配给用户帐户，该角色提供对存储帐户中 Blob 数据的读取和写入访问权限。

az storage account show --resource-group '<your-resource-group-name>' --name '<your-storage-account-name>' --query id

az role assignment create --assignee "<user@domain>" \
    --role "Storage Blob Data Contributor" \
    --scope "<your-resource-id>"

Get-AzResource -ResourceGroupName "<yourResourceGroupname>" -Name "<yourStorageAccountName>"

New-AzRoleAssignment -SignInName <user@domain> `
    -RoleDefinitionName "Storage Blob Data Contributor" `
    -Scope <yourStorageAccountId>

获取存储帐户的连接字符串

请记下该连接字符串和容器名称 。 在代码中使用它们从事件中心接收事件。

1. 在菜单中选择“工具”>“NuGet 包管理器”>“包管理器控制台”。

2. 在“包管理器控制台”窗口中，确认为“默认项目”选择了“EventHubsReceiver” 。 如果不是，请使用下拉列表选择“EventHubsReceiver”。

3. 运行以下命令安装 Azure.Messaging.EventHubs 和 Azure.Identity NuGet 包。 按 ENTER 运行最新命令。

   Install-Package Azure.Messaging.EventHubs
   Install-Package Azure.Messaging.EventHubs.Processor
   Install-Package Azure.Identity
   

1. 在菜单中选择“工具”>“NuGet 包管理器”>“包管理器控制台”。

2. 在“包管理器控制台”窗口中，确认为“默认项目”选择了“EventHubsReceiver” 。 如果不是，请使用下拉列表选择“EventHubsReceiver”。

3. 运行以下命令安装 Azure.Messaging.EventHubs NuGet 包：

   Install-Package Azure.Messaging.EventHubs
   Install-Package Azure.Messaging.EventHubs.Processor
   

1. 将 Program.cs 文件中的现有代码替换为以下示例代码。 然后，替换 BlobContainerClient URI 的 <STORAGE_ACCOUNT_NAME> 和 <BLOB_CONTAINER_NAME> 占位符值。 同时请替换 EventProcessorClient 的 <EVENT_HUB_NAMESPACE> 和 <HUB_NAME> 占位符值。

   下面是代码中的重要步骤：

   1. 使用事件中心命名空间和事件中心名称创建 EventProcessorClient 对象。 需要在先前创建的 Azure 存储中为容器生成 BlobContainerClient 对象。
   2. 为 EventProcessorClient 对象的 ProcessEventAsync 和 ProcessErrorAsync 事件指定处理程序。
   3. 通过对 EventProcessorClient 对象调用 StartProcessingAsync，开始处理事件。
   4. 通过对 EventProcessorClient 对象调用 StopProcessingAsync，在 30 秒后停止处理事件。

   using Azure.Identity;
   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Consumer;
   using Azure.Messaging.EventHubs.Processor;
   using Azure.Storage.Blobs;
   using System.Text;
   
   // Create a blob container client that the event processor will use
   // TODO: Replace <STORAGE_ACCOUNT_NAME> and <BLOB_CONTATINAER_NAME> with actual names
   BlobContainerClient storageClient = new BlobContainerClient(
       new Uri("https://<STORAGE_ACCOUNT_NAME>.blob.core.chinacloudapi.cn/<BLOB_CONTAINER_NAME>"),
       new DefaultAzureCredential());
   
   // Create an event processor client to process events in the event hub
   // TODO: Replace the <EVENT_HUBS_NAMESPACE> and <HUB_NAME> placeholder values
   var processor = new EventProcessorClient(
       storageClient,
       EventHubConsumerClient.DefaultConsumerGroupName,
       "<EVENT_HUB_NAMESPACE>.servicebus.chinacloudapi.cn",
       "<HUB_NAME>",
       new DefaultAzureCredential());
   
   // Register handlers for processing events and handling errors
   processor.ProcessEventAsync += ProcessEventHandler;
   processor.ProcessErrorAsync += ProcessErrorHandler;
   
   // Start the processing
   await processor.StartProcessingAsync();
   
   // Wait for 30 seconds for the events to be processed
   await Task.Delay(TimeSpan.FromSeconds(30));
   
   // Stop the processing
   await processor.StopProcessingAsync();
   
   Task ProcessEventHandler(ProcessEventArgs eventArgs)
   {
       // Write the body of the event to the console window
       Console.WriteLine("\tReceived event: {0}", Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray()));
       Console.ReadLine();
       return Task.CompletedTask;
   }
   
   Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
   {
       // Write details about the error to the console window
       Console.WriteLine($"\tPartition '{eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
       Console.WriteLine(eventArgs.Exception.Message);
       Console.ReadLine();
       return Task.CompletedTask;
   }
   

1. 将 Program.cs 文件中的现有代码替换为以下示例代码。 然后，替换 BlobContainerClient URI 的 <AZURE_STORAGE_CONNECTION_STRING> 和 <BLOB_CONTAINER_NAME> 占位符值。 同时请替换 EventProcessorClient 的 <EVENT_HUB_NAMESPACE_CONNECTION_STRING> 和 <HUB_NAME> 占位符值。

   下面是代码中的重要步骤：

   1. 使用命名空间的主连接字符串和事件中心创建 EventProcessorClient 对象。 需要在先前创建的 Azure 存储中为容器生成 BlobContainerClient 对象。
   2. 为 EventProcessorClient 对象的 ProcessEventAsync 和 ProcessErrorAsync 事件指定处理程序。
   3. 通过对 EventProcessorClient 对象调用 StartProcessingAsync，开始处理事件。
   4. 通过对 EventProcessorClient 对象调用 StopProcessingAsync，在 30 秒后停止处理事件。

   using Azure.Messaging.EventHubs;
   using Azure.Messaging.EventHubs.Consumer;
   using Azure.Messaging.EventHubs.Processor;
   using Azure.Storage.Blobs;
   using System.Text;
   
   // Create a blob container client that the event processor will use 
   BlobContainerClient storageClient = new BlobContainerClient(
       "<AZURE_STORAGE_CONNECTION_STRING>", "<BLOB_CONTAINER_NAME>");
   
   // Create an event processor client to process events in the event hub
   var processor = new EventProcessorClient(
       storageClient,
       EventHubConsumerClient.DefaultConsumerGroupName,
       "<EVENT_HUBS_NAMESPACE_CONNECTION_STRING>",
       "<HUB_NAME>");
   
   // Register handlers for processing events and handling errors
   processor.ProcessEventAsync += ProcessEventHandler;
   processor.ProcessErrorAsync += ProcessErrorHandler;
   
   // Start the processing
   await processor.StartProcessingAsync();
   
   // Wait for 30 seconds for the events to be processed
   await Task.Delay(TimeSpan.FromSeconds(30));
   
   // Stop the processing
   await processor.StopProcessingAsync();
   
   Task ProcessEventHandler(ProcessEventArgs eventArgs)
   {
       // Write the body of the event to the console window
       Console.WriteLine("\tReceived event: {0}", Encoding.UTF8.GetString(eventArgs.Data.Body.ToArray()));
       return Task.CompletedTask;
   }
   
   Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
   {
       // Write details about the error to the console window
       Console.WriteLine($"\tPartition '{eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
       Console.WriteLine(eventArgs.Exception.Message);
       return Task.CompletedTask;
   }