使用 TypeScript 实现 Azure Event Hubs 高吞吐量事件流处理

解决大规模实时数据处理难题：通过配置 Azure Event Hubs TypeScript SDK，实现支持分区、检查点（Checkpointing）和批量发送的高性能事件流管道。

为什么需要这个技能

在构建实时监控、日志分析或 IoT 遥测系统时，传统的 API 接口无法支撑每秒数万次的并发写入。Azure Event Hubs 提供了大数据量摄入的平台，而 TypeScript SDK 则允许开发者以类型安全的方式定义生产者（Producer）和消费者（Consumer）。

掌握此技能可以让你在处理海量流数据时，确保数据的顺序性（通过分区键）、处理的可靠性（通过 Blob 检查点存储）以及极高的吞吐性能（通过批处理发送）。

适用场景

• IoT 遥测数据采集：从成千上万个传感器实时接收温度、压力等数据。
• 实时日志聚合：将分布式服务的运行日志实时推送至分析平台。
• 事件驱动架构（EDA）：在微服务之间构建异步的、高可扩展的消息总线。
• 数据管道预处理：在数据进入数据库之前进行实时过滤或简单的转换。

核心工作流

1. 环境初始化与认证

安装核心依赖 @azure/event-hubs 和 @azure/identity，使用 DefaultAzureCredential 实现无密钥的身份验证。

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

2. 高效发送事件（生产者）

不建议单条发送，而应使用 createBatch() 创建批次。可以通过 partitionKey 确保同一设备的数据进入同一分区，从而保证顺序。

const producer = new EventHubProducerClient(fullyQualifiedNamespace, eventHubName, credential);
const batch = await producer.createBatch({ partitionKey: "device-123" });
batch.tryAdd({ body: { temperature: 72.5, deviceId: "sensor-1" } });
await producer.sendBatch(batch);

3. 可靠消费事件（消费者）

在生产环境下，必须结合 @azure/eventhubs-checkpointstore-blob 记录处理进度（Checkpoint）。这样当服务重启时，消费者能从上次停止的位置继续读取，而非重新开始。

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)}`);
    }
    if (events.length > 0) {
      await context.updateCheckpoint(events[events.length - 1]);
    }
  },
  processError: async (err, context) => {
    console.error(`Error: ${err.message}`);
  },
});

下载和安装

下载 azure-eventhub-ts 中文版 Skill ZIP

解压后将目录放入你的 AI 工具 skills 文件夹，重启工具后即可使用。具体路径参考内附的 USAGE.zh.md。

你可能还需要

暂无推荐