azure-eventhub-ts

Azure Event Hubs SDK for TypeScript

高スループットのイベントストリーミングとリアルタイムデータ取り込み。

インストール

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

コンシューマーグループでチェックポイントを使用する場合：

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

環境変数

EVENTHUB_NAMESPACE=<namespace>.servicebus.windows.net
EVENTHUB_NAME=my-eventhub
STORAGE_ACCOUNT_NAME=<storage-account>
STORAGE_CONTAINER_NAME=checkpoints

認証

import { EventHubProducerClient, EventHubConsumerClient } from "@azure/event-hubs";
import { DefaultAzureCredential } from "@azure/identity";

const fullyQualifiedNamespace = process.env.EVENTHUB_NAMESPACE!;
const eventHubName = process.env.EVENTHUB_NAME!;
const credential = new DefaultAzureCredential();

// Producer
const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);

// Consumer
const consumer = new EventHubConsumerClient(
  "$Default", // Consumer group
  fullyQualifiedNamespace,
  eventHubName,
  credential
);

基本的なワークフロー

イベントの送信

const producer = new EventHubProducerClient(namespace, eventHubName, credential);

// Create batch and add events
const batch = await producer.createBatch();
for (const body of [{ temperature: 72.5, deviceId: "sensor-1" }, { temperature: 68.2, deviceId: "sensor-2" }]) {
  if (!batch.tryAdd({ body })) throw new Error("Event does not fit this batch; flush/rebatch or reject oversized event");
}

await producer.sendBatch(batch);
await producer.close();

特定のパーティションへの送信

// By partition ID
const batch = await producer.createBatch({ partitionId: "0" });

// By partition key (consistent hashing)
const batch = await producer.createBatch({ partitionKey: "device-123" });

イベントの受信（シンプル）

const consumer = new EventHubConsumerClient("$Default", namespace, eventHubName, credential);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Partition: ${context.partitionId}, Body: ${JSON.stringify(event.body)}`);
    }
  },
  processError: async (err, context) => {
    console.error(`Error on partition ${context.partitionId}: ${err.message}`);
  },
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await consumer.close();
}, 60000);

チェックポイントを使用した受信（本番環境向け）

import { EventHubConsumerClient } from "@azure/event-hubs";
import { ContainerClient } from "@azure/storage-blob";
import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

const containerClient = new ContainerClient(
  `https://${storageAccount}.blob.core.windows.net/${containerName}`,
  credential
);

const checkpointStore = new BlobCheckpointStore(containerClient);

const consumer = new EventHubConsumerClient(
  "$Default",
  namespace,
  eventHubName,
  credential,
  checkpointStore
);

const subscription = consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Processing: ${JSON.stringify(event.body)}`);
    }
    // Checkpoint after processing batch
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  },
});

特定の位置からの受信

const subscription = consumer.subscribe({
  processEvents: async (events, context) => { /* ... */ },
  processError: async (err, context) => { /* ... */ },
}, {
  startPosition: {
    // Start from beginning
    "0": { offset: "@earliest" },
    // Start from end (new events only)
    "1": { offset: "@latest" },
    // Start from specific offset
    "2": { offset: "12345" },
    // Start from specific time
    "3": { enqueuedOn: new Date("2024-01-01") },
  },
});

Event Hub のプロパティ

// Get hub info
const hubProperties = await producer.getEventHubProperties();
console.log(`Partitions: ${hubProperties.partitionIds}`);

// Get partition info
const partitionProperties = await producer.getPartitionProperties("0");
console.log(`Last sequence: ${partitionProperties.lastEnqueuedSequenceNumber}`);

バッチ処理のオプション

const subscription = consumer.subscribe(
  {
    processEvents: async (events, context) => { /* ... */ },
    processError: async (err, context) => { /* ... */ },
  },
  {
    maxBatchSize: 100,           // Max events per batch
    maxWaitTimeInSeconds: 30,    // Max wait for batch
  }
);

主要な型

import {
  EventHubProducerClient,
  EventHubConsumerClient,
  EventData,
  ReceivedEventData,
  PartitionContext,
  Subscription,
  SubscriptionEventHandlers,
  CreateBatchOptions,
  EventPosition,
} from "@azure/event-hubs";

import { BlobCheckpointStore } from "@azure/eventhubs-checkpointstore-blob";

イベントのプロパティ

// Send with properties
const batch = await producer.createBatch();
batch.tryAdd({
  body: { data: "payload" },
  properties: {
    eventType: "telemetry",
    deviceId: "sensor-1",
  },
  contentType: "application/json",
  correlationId: "request-123",
});

// Access in receiver
consumer.subscribe({
  processEvents: async (events, context) => {
    for (const event of events) {
      console.log(`Type: ${event.properties?.eventType}`);
      console.log(`Sequence: ${event.sequenceNumber}`);
      console.log(`Enqueued: ${event.enqueuedTimeUtc}`);
      console.log(`Offset: ${event.offset}`);
    }
  },
});

エラー処理

consumer.subscribe({
  processEvents: async (events, context) => {
    try {
      for (const event of events) {
        await processEvent(event);
      }
      if (events.length > 0) await context.updateCheckpoint(events[events.length - 1]);
    } catch (error) {
      // Don't checkpoint on error - events will be reprocessed
      console.error("Processing failed:", error);
    }
  },
  processError: async (err, context) => {
    if (err.name === "MessagingError") {
      // Transient error - SDK will retry
      console.warn("Transient error:", err.message);
    } else {
      // Fatal error
      console.error("Fatal error:", err);
    }
  },
});

ベストプラクティス

1. チェックポイントを使用する - 永続的な処理が完了した後にのみチェックポイントを記録してください。配信が繰り返される場合があるため、ハンドラーは冪等にしてください
2. バッチで送信する - 効率的に送信するには createBatch() を使用してください
3. パーティションキー - 関連するイベントの順序を保証するには、パーティションキーを使用してください
4. コンシューマーグループ - 処理パイプラインごとに個別のコンシューマーグループを使用してください
5. エラーを適切に処理する - 処理に失敗した場合はチェックポイントを記録しないでください
6. クライアントを閉じる - 完了したら必ずプロデューサー／コンシューマーを閉じてください
7. 遅延を監視する - lastEnqueuedSequenceNumber と処理済みのシーケンスを追跡してください

使用する状況

このスキルは、概要に記載されたワークフローやアクションを実行する場合に適用できます。

制限事項

• タスクが上記の説明の範囲に明確に該当する場合にのみ、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認してください。

チェックポイントは進捗を記録するものであり、外部への副作用を厳密に一度だけ実行するものではなく、エラーを握りつぶした場合に即時の再試行を保証するものでもありません。安定したイベント ID、永続的な重複排除、明示的な再試行／デッドレターの設計を使用してください。すべての MessagingError を一時的なものとして扱わず、再試行可能かどうかを確認してください。例は断片的なものです。インポートやハンドラーを補い、イベント本文はマスキングしてください。公式 SDK ガイド。