npm 包 node-rdkafka 使用教程

前言

node-rdkafka 是 Node.js 平台上使用 Kafka 的一个优秀的 npm 包，它提供了 Node.js 与 Kafka 的高效通信接口，可以轻松地实现应用中的消息队列、日志记录和大规模事件处理等功能。本文就如何使用 node-rdkafka 包来实现一个Kafka消费者进行详细介绍。

安装

在安装 node-rdkafka 之前，先确保你本地已经安装了 node 和 npm，可以通过以下命令来进行确认：

node -v
npm -v

如果都有相应的版本号信息，则说明已经安装成功。

接下来，我们就可以通过 npm 包管理工具，进行 node-rdkafka 包的安装：

npm install node-rdkafka

安装完成后，我们便可以在代码中引入 node-rdkafka 包：

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

创建 Consumer

在开始使用 node-rdkafka 包时，首先需要创建一个 Kafka Consumer，代码如下：

const kafkaConf = {
  'group.id': 'consumer-group',
  'metadata.broker.list': 'localhost:9092',
};
const consumer = new kafka.KafkaConsumer(kafkaConf, {});

这里的 kafkaConf 对象用来配置 Consumer 的相关属性，其中包括 group.id 用来标识 Consumer 所属的消费组，以及 metadata.broker.list 用来指定 Kafka Broker 的连接地址。

订阅 Topic

在创建 Consumer 后，需要将其与某个 Topic 进行关联，以便从该 Topic 中获取 Kafka 的消息。代码如下：

-- -------------------- ---- -------
-------------------
--------
  ------------ -- -- -
    -----------------------------------
    -------------------
  --
  ----------- ------ -- -
    -----------------------------------
  ---

在这段代码中，使用 consumer.subscribe(['test-topic']) 方法来订阅指定的 Kafka Topic，consumer.consume() 方法表示开始从 Topic 中消费数据。

处理 Kafka 消息

在订阅了 Kafka Topic 后，需要判断是否有新的消息到来，可以使用 consumer.on('data', callback) 方法监听 data 事件，代码如下：

consumer.on('data', (data) => {
  console.log(data.value.toString());
});

在 callback 对象中，获取到的 data 参数即为 Kafka 消息中的 Payload，通过 toString() 方法将其转换为字符串类型，以便进行下一步的处理。

消费者关闭

在 Consumer 订阅了 Topic 并消费了其中的消息后，需要手动断开与 Kafka 的连接，代码如下：

consumer.disconnect();

这样会释放在内存中创建的 Consumer 对象，并关闭已经建立的网络连接。

总结

本文详细介绍了如何使用 node-rdkafka 包来创建一个 Kafka Consumer，并进行消息的订阅和消费。通过以上的学习和实践，相信读者已经掌握了基本的 Kafka Consumer 开发方法和技巧，可以轻松应用在实际项目中，提升工作效率和代码质量。

来源：JavaScript中文网 ，转载请注明来源 https://www.javascriptcn.com/post/157880