mcpField
A
azure-eventhub-ts

azure-eventhub-ts

github.com/sickn33数据与数据库48.7k66 次安装今天更新
代码仓库

关于 azure-eventhub-ts

这是一项使用 Azure Event Hubs SDK 构建事件流与实时数据摄取工作流的 TypeScript 技能，涵盖发送、接收、分区、批处理和检查点管理。

主要功能

  1. 01批量发布事件
  2. 02按分区 ID 或键路由
  3. 03订阅者消费事件
  4. 04通过检查点记录进度
  5. 05读取 Event Hub 与分区信息

适用场景

  1. 01批量发布遥测数据
  2. 02按分区键路由相关事件
  3. 03使用检查点消费事件
  4. 04检查分区序列位置

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/sickn33/agentic-awesome-skills --skill azure-eventhub-ts

安装后在 Claude Code 中输入/azure-eventhub-ts确认已加载。

详细介绍

Azure Event Hubs TypeScript SDK

高吞吐量事件流和实时数据摄取。

安装

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 指南。

评论

支持 Markdown