クイック スタート: JavaScript を使用してイベント ハブとの間でイベントを送信または受信する

このクイック スタートでは、 @azure/event-hubs npm パッケージを使用してイベント ハブとの間でイベントを送受信する方法について説明します。

Azure Event Hubs を初めて使用する場合は、開始する前に Event Hubs の概要 を参照してください。

[前提条件]

• Microsoft Azure サブスクリプション。 Azure Event Hubs を含む Azure サービスを使用するには、サブスクリプションが必要です。 Azure アカウントをお持ちでない場合は、 無料試用版にサインアップしてください。
• Node.js LTS。 最新の 長期サポート (LTS) バージョンをダウンロードします。
• Visual Studio Code (推奨) または他の任意の統合開発環境 (IDE)。
• Event Hubs 名前空間とイベント ハブを作成します。 Azure portal を使用して、Event Hubs 型の名前空間を作成します。アプリケーションがイベント ハブと通信するために必要な管理資格情報を取得します。 詳細については、「 Azure portal を使用してイベント ハブを作成する」を参照してください。

イベントを送信する npm パッケージをインストールする

Event Hubs の Node Package Manager (npm) パッケージをインストールするには、パスにnpmがあるコマンド プロンプト ウィンドウを開きます。 サンプルを保持するフォルダーにディレクトリを変更します。

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

npm install @azure/event-hubs

Azure に対してアプリを認証する

このクイック スタートでは、Azure Event Hubs に接続する 2 つの方法について説明します。

• パスワードレス。 Microsoft Entra ID とロールベースのアクセス制御 (RBAC) のセキュリティ プリンシパルを使用して、Event Hubs 名前空間に接続します。 コード、構成ファイル、または Azure Key Vault などのセキュリティで保護されたストレージに、ハードコーディングされた接続文字列を含める心配はありません。
• 接続文字列。 接続文字列を使用して Event Hubs 名前空間に接続します。 Azure を初めて使用する場合は、接続文字列オプションの方が理解しやすいかもしれません。

実際のアプリケーションと運用環境では、パスワードレス オプションを使用することをお勧めします。 詳細については、Azure サービスの Service Bus の認証と承認とパスワードレス接続に関するページを参照してください。

az eventhubs 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 <yourResourceId>

イベントを送信する

このセクションでは、イベント ハブにイベントを送信する 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.windows.net`; 
   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
   

   このコマンドは、3 つのイベントのバッチをイベント ハブに送信します。

   パスワードレス (Microsoft Entra ID ロールベースのアクセス制御 (RBAC) 認証を使用している場合は、Azure Event Hubs データ所有者ロールに追加したアカウントを使用して Azure にサインインすることが必要になる場合があります。 az login コマンドを使用します。

4. Azure portal で、イベント ハブがメッセージを受信したことを確認します。 グラフを更新するには、ページを更新します。 メッセージが受信されたことが示されるまでに数秒かかることがあります。

   

   注

   詳細と完全なソース コードについては、 GitHub sendEvents.js ページを参照してください。

受信イベント

このセクションでは、JavaScript アプリケーションの Azure Blob Storage チェックポイント ストアを使用して、イベント ハブからイベントを受信します。 Azure Storage BLOB で受信したメッセージに対して一定の間隔でメタデータ チェックポイントを実行します。 この手法によって、後で前回終了したところからメッセージを継続して受信しやすくなります。

チェックポイント ストアとして Azure Blob Storage を使用する場合は、次の推奨事項に従います。

• コンシューマー グループごとに個別のコンテナーを使用します。 同じストレージ アカウントを使用できますが、各グループごとに 1 つのコンテナーを使用します。
• ストレージ アカウントを他の目的で使用しないでください。
• コンテナーを他の目的で使用しないでください。
• デプロイされたアプリケーションと同じリージョンにストレージ アカウントを作成します。 アプリケーションがオンプレミスの場合は、可能な中で最も近いリージョンを選択することを試みてください。

Azure portal の [ストレージ アカウント] ページの [Blob service] セクションで、次の設定が無効になっていることを確認してください。

• 階層型名前空間
• BLOB の論理的な削除
• バージョン管理

Azure Storage アカウントと BLOB コンテナーを作成する

BLOB コンテナーを使用して Azure ストレージ アカウントを作成するには:

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 パッケージをインストールする

受信側では、さらに 2 つのパッケージをインストールする必要があります。 このクイックスタートでは、Azure Blob Storage を使用してチェックポイントを保持し、プログラムが既に読み取ったイベントを読み取らないようにします。 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.windows.net`; 
   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.windows.net`;
   
   // 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'
   

   注

   情報コメントを含む完全なソース コードについては、 receiveEventsUsingCheckpointStore.jsを参照してください。

   受信側プログラムは、イベント ハブ内の既定のコンシューマー グループのすべてのパーティションからイベントを受信します。

リソースをクリーンアップする

Event Hubs 名前空間を持つリソース グループを削除するか、リソース グループを保持する場合は名前空間のみを削除します。

関連コンテンツ

• JavaScript のサンプル
• TypeScript のサンプル