在 JavaScript (Node.js) 中使用 Kafka：node-rdkafka 库

Apache Kafka 和 Node.js：认识 node-rdkafka

Apache Kafka 是一个分布式事件流平台，已成为构建可靠且可扩展的数据交换系统的事实标准。在 Node.js 生态系统中，用于操作 Kafka 的最高性能和最成熟的库之一是 node-rdkafka。它提供了对 librdkafka 库（用 C/C++ 编写）的原生绑定，从而确保了最高的速度和稳定性。

在本文中，我们将详细介绍如何配置和使用 node-rdkafka 来发送（生产者）和读取（消费者）消息。您将学习如何将 Kafka 集成到您的 Node.js 应用程序中，处理错误以及管理数据流。

安装 node-rdkafka

安装该库需要 Node.js 12 或更高版本以及 C++ 编译器（gcc、clang 或 MSVC）。安装使用标准包管理器进行：

npm install node-rdkafka

如果构建失败，请确保您已安装 build-essential（Linux）或 Python 以及 Windows 的构建工具（npm install --global windows-build-tools）。

创建生产者（消息发送者）

Kafka 中的生产者负责将消息发布到主题。让我们看一个基本示例：

const Kafka = require('node-rdkafka');

// 创建生产者const producer = new Kafka.Producer({  'metadata.broker.list': 'localhost:9092',  'dr_cb': true  // 用于确认投递的回调});

// 连接时的事件处理程序producer.on('event.log', function(event) {  console.log('日志:', event.message);});

// 错误处理程序producer.on('event.error', function(err) {  console.error('生产者错误:', err);});

// 连接到代理producer.connect();

// 连接后发送消息producer.on('ready', function() {  try {    // 向 'test-topic' 主题发送消息    producer.produce(      'test-topic',  // 主题名称      null,          // 分区（null = 自动分配）      Buffer.from('你好, Kafka!'),  // 消息体（Buffer）      'my-key',      // 消息键（可选）      Date.now()     // 时间戳（可选）    );    console.log('消息已发送');  } catch (err) {    console.error('发送时出错:', err);  }});

// 不要忘记调用 poll 来处理回调setInterval(() => {  producer.poll();}, 1000);

// 进程结束时关闭连接process.on('SIGINT', () => {  producer.disconnect();  process.exit();});

请注意：producer.produce() 方法接受消息体作为 Buffer。这允许传递字符串和二进制数据。参数 dr_cb: true 启用了投递报告回调通知。

创建消费者（消息接收者）

消费者订阅一个或多个主题并处理传入的消息。要使用消费者组，请使用 KafkaConsumer：

const Kafka = require('node-rdkafka');

const consumer = new Kafka.KafkaConsumer({  'group.id': 'my-group',  'metadata.broker.list': 'localhost:9092',  'enable.auto.commit': true}, {});

consumer.on('event.error', function(err) {  console.error('消费者错误:', err);});

consumer.on('ready', function() {  console.log('消费者已就绪');  // 订阅主题  consumer.subscribe(['test-topic']);  // 启动消费循环  consumer.consume();});

// 处理每条消息consumer.on('data', function(data) {  console.log('收到消息:');  console.log('  主题:', data.topic);  console.log('  分区:', data.partition);  console.log('  键:', data.key ? data.key.toString() : null);  console.log('  值:', data.value.toString());