在 TypeScript 中使用 Kafka：KafkaJS 示例

使用 KafkaJS 在 TypeScript 中操作 Kafka

Apache Kafka 是一个分布式流数据处理平台，广泛应用于微服务架构、事件处理和实时分析。在使用 TypeScript 开发时，与 Kafka 交互最流行的库之一是 KafkaJS。它提供了现代、异步的 API，并完全支持 TypeScript，是构建可靠且可扩展应用程序的理想选择。

什么是 KafkaJS？

KafkaJS 是一个用 JavaScript/TypeScript 编写的 Apache Kafka 客户端库。它支持所有基本操作：发送和读取消息、管理消费者组、处理主题和分区。凭借内置的类型定义，KafkaJS 在 IDE 中提供了严格的类型检查和自动补全功能，大大简化了开发和调试过程。

安装 KafkaJS

首先，通过 npm 安装该库：

npm install kafkajs

如果您使用 TypeScript，则无需额外安装类型定义——它们已包含在包中。

基本配置与连接

第一步是创建 Kafka 客户端实例。指定客户端 ID 和代理地址：

import { Kafka } from 'kafkajs';

const kafka = new Kafka({  clientId: 'my-app',  brokers: ['localhost:9092']});

// 检查连接const admin = kafka.admin();await admin.connect();console.log('Connected to Kafka');await admin.disconnect();

创建生产者 (Producer)

生产者将消息发送到主题。KafkaJS 支持在发送时自动创建主题（如果代理配置中启用了该功能）。

import { Kafka, Partitioners } from 'kafkajs';

const producer = kafka.producer({  createPartitioner: Partitioners.DefaultPartitioner});

await producer.connect();

// 发送单条消息await producer.send({  topic: 'orders',  messages: [    { key: 'order-123', value: JSON.stringify({ userId: 1, amount: 250 }) }  ]});

// 发送多条消息await producer.send({  topic: 'orders',  messages: [    { key: 'order-124', value: 'Message 2' },    { key: 'order-125', value: 'Message 3' }  ]});

await producer.disconnect();

创建消费者 (Consumer)

消费者从主题中读取消息。它必须订阅一个或多个主题，并注册到一个消费者组中。

const consumer = kafka.consumer({ groupId: 'order-group' });

await consumer.connect();await consumer.subscribe({ topic: 'orders', fromBeginning: true });

await consumer.run({  eachMessage: async ({ topic, partition, message }) => {    console.log({      partition,      offset: message.offset,      key: message.key?.toString(),      value: message.value?.toString(),    });  },});

使用消费者组

消费者组允许在多个消费者实例之间分配负载。Kafka 会自动将主题的分区分配给同一组内的不同消费者。如果某个消费者发生故障，其分区将被重新分配。

// 同一组中的两个消费者将读取不同的分区const consumer1 = kafka.consumer({ groupId: 'order-group' });const consumer2 = kafka.consumer({ groupId: 'order-group' });

// 每个消费者订阅同一个主题await consumer1.subscribe({ topic: 'orders' });await consumer2.subscribe({ topic: 'orders' });