在 Node.js 中使用 Apache Kafka：KafkaJS 库

使用 KafkaJS 库在 Node.js 中操作 Apache Kafka

Apache Kafka 是一个分布式事件流平台，已成为构建高负载消息交换系统的事实标准。在 Node.js 生态系统中，用于与 Kafka 交互的最流行且最稳定的库之一是 KafkaJS。这是一个现代、完全异步的客户端，使用纯 JavaScript 编写，支持 Kafka 的所有关键功能：生产者、消费者、消费组、偏移量管理和事务。

在本文中，我们将详细介绍如何安装 KafkaJS、配置与 Kafka 集群的连接、发送和读取消息，并探讨在生产环境中使用该库的最佳实践。

安装和配置 KafkaJS

使用标准包管理器 npm 安装该库：

npm install kafkajs

安装后，需要创建一个客户端，指定代理地址和客户端 ID。最小配置示例：

const { Kafka } = require('kafkajs');

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

对于生产环境，建议指定多个代理（用逗号分隔）以确保容错性，并配置 SSL 和 SASL 认证：

const kafka = new Kafka({  clientId: 'my-production-app',  brokers: ['broker1:9092', 'broker2:9092', 'broker3:9092'],  ssl: true,  sasl: {    mechanism: 'scram-sha-256',    username: 'myuser',    password: 'mypassword'  }});

生产者：向主题发送消息

生产者负责将消息发布到指定的 Kafka 主题。KafkaJS 提供两种生产者工作模式：简单模式（send）和事务模式（用于精确投递）。让我们看一个发送消息的基本示例：

const producer = kafka.producer();

async function sendMessage() {  await producer.connect();

  await producer.send({    topic: 'test-topic',    messages: [      { key: 'user1', value: 'Hello Kafka from Node.js!' },      { key: 'user2', value: JSON.stringify({ orderId: 123, amount: 2500 }) }    ]  });

  await producer.disconnect();}

sendMessage().catch(console.error);

关键点：

• topic — 消息发送到的主题名称。
• messages — 对象数组，每个对象可以包含 key（可选）和 value（字符串或 Buffer）。
• 建议在应用程序启动时只调用一次 producer.connect()，并复用该连接。

为了提高生产环境中的性能，请使用连接池并配置 acks：

const producer = kafka.producer({  allowAutoTopicCreation: false,  transactionTimeout: 30000});

await producer.send({  topic: 'orders',  messages: [{ value: 'New order' }],  acks: -1  // 等待所有副本确认});

消费者：从主题读取消息

消费者允许从一个或多个主题读取消息。最重要的概念是消费组，它负责负载均衡，并确保组内每条消息只被一个消费者处理。

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

async function consumeMessages() {  await consumer.connect();  await consumer.subscribe({ topic: 'test-topic', fromBeginning: true });

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