使用 JavaScript 向事件中心发送事件或从事件中心接收事件

本快速入门介绍如何使用 @azure/event-hubs npm 包向事件中心发送事件以及从事件中心接收事件。

先决条件

如果不熟悉 Azure 事件中心，请在阅读本快速入门之前参阅事件中心概述。

若要完成本快速入门，需要具备以下先决条件：

• Azure 订阅。 若要使用 Azure 服务（包括 Azure 事件中心），需要一个订阅。 如果没有现有的 Azure 帐户，可以注册试用订阅帐户，或者在创建帐户时使用 MSDN 订阅者权益。
• Node.js LTS。 下载最新的长期支持 (LTS) 版本。
• Visual Studio Code（推荐）或任何其他集成开发环境 (IDE)。
• 创建事件中心命名空间和事件中心。 第一步是使用 Azure 门户创建事件中心类型的命名空间，并获取应用程序与事件中心进行通信所需的管理凭据。 要创建命名空间和事件中心，请按照此文中的步骤操作。

安装 npm 包以发送事件

若要安装适用于事件中心的 Node 包管理器 (npm) 包，请打开路径中包含 npm 的命令提示符，将目录切换到用于保存示例的文件夹。

npm install @azure/event-hubs
npm install @azure/identity

npm install @azure/event-hubs

向 Azure 验证应用

本快速入门介绍连接到 Azure 事件中心的两种方法：无密码方法和连接字符串方法。 第一个选项展示了如何使用 Microsoft Entra ID 中的安全主体和基于角色的访问控制 (RBAC) 连接到事件中心命名空间。 无需担心代码、配置文件或安全存储（如 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>

发送事件

在本部分中，你将创建一个可向事件中心发送事件的 JavaScript 应用程序。

1. 打开偏好的编辑器，例如 Visual Studio Code。

2. 创建名为 send.js 的文件，然后将以下代码粘贴到其中：

   • 无密码（推荐）
   • 连接字符串

   在代码中，使用实际值替换以下占位符：

   • EVENT HUBS NAMESPACE NAME
   • EVENT HUB NAME

   const { EventHubProducerClient } = require("@azure/event-hubs");
   const { DefaultAzureCredential } = require("@azure/identity");
   
   // Event hubs 
   const eventHubsResourceName = "EVENT HUBS NAMESPACE NAME";
   const fullyQualifiedNamespace = `${eventHubsResourceName}.servicebus.chinacloudapi.cn`; 
   const eventHubName = "EVENT HUB NAME";
   
   // Azure Identity - passwordless authentication
   const credential = new DefaultAzureCredential();
   
   async function main() {
   
     // Create a producer client to send messages to the event hub.
     const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);
   
     // Prepare a batch of three events.
     const batch = await producer.createBatch();
     batch.tryAdd({ body: "passwordless First event" });
     batch.tryAdd({ body: "passwordless Second event" });
     batch.tryAdd({ body: "passwordless Third event" });    
   
     // Send the batch to the event hub.
     await producer.sendBatch(batch);
   
     // Close the producer client.
     await producer.close();
   
     console.log("A batch of three events have been sent to the event hub");
   }
   
   main().catch((err) => {
     console.log("Error occurred: ", err);
   });
   

3. 运行 node send.js 以执行此文件。 此命令将包含三个事件的批发送到事件中心。

4. 在 Azure 门户中，验证事件中心是否已收到消息。 刷新页面以更新图表。 可能需要在几秒钟后才会显示已收到消息。

   

   注意

   有关完整源代码（包括附加的参考注释），请参阅 GitHub sendEvents.js 页。

   现已向事件中心发送事件。

接收事件

在本部分中，你将在 JavaScript 应用程序中使用 Azure Blob 存储检查点存储从事件中心接收事件。 该应用程序将在 Azure 存储 Blob 中定期针对收到的消息执行元数据检查点。 使用此方式可以很容易地在以后的某个时间从退出的位置继续接收消息。

使用 Azure Blob 存储作为检查点存储时，请遵循以下建议：

• 对每个使用者组使用单独的容器。 可以使用同一存储帐户，但每个组使用一个容器。
• 请勿将容器用于任何其他用途，也不要将存储帐户用于任何其他用途。
• 存储帐户应位于部署的应用程序所在的同一区域中。 如果应用程序位于本地，请尝试选择最近的区域。

在 Azure 门户的“存储帐户”页上的“Blob 服务”部分，确保禁用以下设置。

• 分层命名空间
• Blob 软删除
• 版本控制

创建 Azure 存储帐户和 Blob 容器

若要创建 Azure 存储帐户并在其中创建 Blob 容器，请执行以下操作：

1. 创建 Azure 存储帐户
2. 在存储帐户中创建 Blob 容器
3. 对 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>

安装 npm 包以接收事件

对于接收端，需要额外安装两个包。 本快速入门将使用 Azure Blob 存储来保存检查点，使程序不会读取已读取的事件。 它在 Blob 中按固定的时间间隔对收到的消息执行元数据检查点。 使用此方式可以很容易地在以后的某个时间从退出的位置继续接收消息。

npm install @azure/storage-blob
npm install @azure/eventhubs-checkpointstore-blob
npm install @azure/identity

npm install @azure/storage-blob
npm install @azure/eventhubs-checkpointstore-blob

编写用于接收事件的代码

1. 打开偏好的编辑器，例如 Visual Studio Code。

2. 创建名为 receive.js 的文件，然后将以下代码粘贴到其中：

   • 无密码（推荐）
   • 连接字符串

   在代码中，使用实际值替换以下占位符：

   • EVENT HUBS NAMESPACE NAME
   • EVENT HUB NAME
   • STORAGE ACCOUNT NAME
   • STORAGE CONTAINER NAME

   const { DefaultAzureCredential } = require("@azure/identity");
   const { EventHubConsumerClient, earliestEventPosition  } = require("@azure/event-hubs");
   const { ContainerClient } = require("@azure/storage-blob");    
   const { BlobCheckpointStore } = require("@azure/eventhubs-checkpointstore-blob");
   
   // Event hubs 
   const eventHubsResourceName = "EVENT HUBS NAMESPACE NAME";
   const fullyQualifiedNamespace = `${eventHubsResourceName}.servicebus.chinacloudapi.cn`; 
   const eventHubName = "EVENT HUB NAME";
   const consumerGroup = "$Default"; // name of the default consumer group
   
   // Azure Storage 
   const storageAccountName = "STORAGE ACCOUNT NAME";
   const storageContainerName = "STORAGE CONTAINER NAME";
   const baseUrl = `https://${storageAccountName}.blob.core.chinacloudapi.cn`;
   
   // Azure Identity - passwordless authentication
   const credential = new DefaultAzureCredential();
   
   async function main() {
   
     // Create a blob container client and a blob checkpoint store using the client.
     const containerClient = new ContainerClient(
       `${baseUrl}/${storageContainerName}`,
       credential
     );  
     const checkpointStore = new BlobCheckpointStore(containerClient);
   
     // Create a consumer client for the event hub by specifying the checkpoint store.
     const consumerClient = new EventHubConsumerClient(consumerGroup, fullyQualifiedNamespace, eventHubName, credential, checkpointStore);
   
     // Subscribe to the events, and specify handlers for processing the events and errors.
     const subscription = consumerClient.subscribe({
         processEvents: async (events, context) => {
           if (events.length === 0) {
             console.log(`No events received within wait time. Waiting for next interval`);
             return;
           }
   
           for (const event of events) {
             console.log(`Received event: '${event.body}' from partition: '${context.partitionId}' and consumer group: '${context.consumerGroup}'`);
           }
           // Update the checkpoint.
           await context.updateCheckpoint(events[events.length - 1]);
         },
   
         processError: async (err, context) => {
           console.log(`Error : ${err}`);
         }
       },
       { startPosition: earliestEventPosition }
     );
   
     // After 30 seconds, stop processing.
     await new Promise((resolve) => {
       setTimeout(async () => {
         await subscription.close();
         await consumerClient.close();
         resolve();
       }, 30000);
     });
   }
   
   main().catch((err) => {
     console.log("Error occurred: ", err);
   });
   

3. 在命令提示符下运行 node receive.js 以执行此文件。 窗口中应会显示有关已收到事件的消息。

   C:\Self Study\Event Hubs\JavaScript>node receive.js
   Received event: 'First event' from partition: '0' and consumer group: '$Default'
   Received event: 'Second event' from partition: '0' and consumer group: '$Default'
   Received event: 'Third event' from partition: '0' and consumer group: '$Default'
   

   注意

   有关完整源代码（包括附加的参考注释），请参阅 GitHub receiveEventsUsingCheckpointStore.js 页。

现已从事件中心收到事件。 接收器程序将从事件中心内默认使用者组的所有分区接收事件。

后续步骤

查看 GitHub 中的以下示例：

• JavaScript 示例
• TypeScript 示例