TypeScript 中的 Kafka：用于 Node.js 的 KafkaJS

KafkaJS 库：在 TypeScript 中使用 Apache Kafka

Apache Kafka 是一个分布式流处理平台，广泛用于构建高负载的消息交换系统。在 TypeScript（和 Node.js）生态系统中，KafkaJS 库已成为事实上的标准。它提供了一个简洁、现代的 API，并完全支持 TypeScript，使开发人员能够为 Kafka 生产者和消费者编写类型安全的代码。

在本文中，我们将详细介绍 KafkaJS 的安装、配置以及主要使用场景。您将学习如何发送和接收消息、处理错误以及如何正确配置客户端以用于生产环境。

安装和配置 KafkaJS

首先，通过 npm 或 yarn 安装该库：

npm install kafkajs# 或yarn add kafkajs

KafkaJS 是用 TypeScript 编写的，因此类型已内置——无需额外安装 @types/kafkajs。

现在，创建一个 Kafka 客户端实例。建议指定多个代理以实现容错：

import { Kafka, logLevel } from 'kafkajs';

const kafka = new Kafka({  clientId: 'my-app',  brokers: ['localhost:9092', 'localhost:9093'],  logLevel: logLevel.INFO, // 日志级别  retry: {    initialRetryTime: 100,    retries: 8  }});

生产者：向 Kafka 发送消息

生产者负责将消息写入 Kafka 主题。KafkaJS 提供了方便的 API 来发送单个消息或批量消息。

基本生产者示例：

import { Kafka, Producer } from 'kafkajs';

async function produceMessage(producer: Producer) {  await producer.send({    topic: 'orders',    messages: [      {         key: 'order-123',         value: JSON.stringify({ userId: 42, amount: 1500 })       },    ],  });  console.log('消息已发送');}

async function runProducer() {  const producer = kafka.producer();  await producer.connect();

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

  await produceMessage(producer);  await producer.disconnect();}

runProducer().catch(console.error);

重要提示： 完成工作后务必调用 disconnect() 以释放资源。对于长期运行的应用程序，生产者可以保持持续连接。

消费者：从 Kafka 读取消息

消费者订阅一个或多个主题并处理消息。KafkaJS 支持消费者组以实现并行处理。

带有消息处理的消费者示例：

import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';

async function runConsumer() {  const consumer: Consumer = kafka.consumer({     groupId: 'order-processor-group'   });

  await consumer.connect();  await consumer.subscribe({     topic: 'orders',     fromBeginning: true // 从最早的消息开始读取  });

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

runConsumer().catch(console.error);

请注意 fromBeginning 参数：如果设置为 true，消费者将读取所有先前累积的消息。对于生产环境，通常使用 false（仅读取新消息）。