消息队列 CKafka 版
有奖捉虫：办公协同&微信生态&物联网文档专题 HOT
文档中心 > 消息队列 CKafka 版 > SDK 文档 > 连接器相关 SDK > 弹性 Topic 收发消息 SDK > Node.js SDK

Node.js SDK

最近更新时间：2022-11-25 15:38:59

我的收藏

本页目录：

操作场景

本文介绍使用 Node.js 客户端连接 CKafka 弹性 Topic 并收发消息的操作步骤。

前提条件

安装 GCC
安装 Node.js

操作步骤

步骤1：准备环境

安装 C++依赖库

1. 执行以下命令切换到 yum 源配置目录 /etc/yum.repos.d/。
 cd /etc/yum.repos.d/
2. 创建 yum 源配置文件 confluent.repo。
   [Confluent.dist]
   name=Confluent repository (dist)
   baseurl=https://packages.confluent.io/rpm/5.1/7
   gpgcheck=1
   gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
   enabled=1   
   [Confluent]
   name=Confluent repository
   baseurl=https://packages.confluent.io/rpm/5.1
   gpgcheck=1
   gpgkey=https://packages.confluent.io/rpm/5.1/archive.key
   enabled=1
3. 执行以下命令安装 C++ 依赖库。
yum install librdkafka-devel

安装 Node.js 依赖库

1. 执行以下命令为预处理器指定 OpenSSL 头文件路径。
export CPPFLAGS=-I/usr/local/opt/openssl/include
2. 执行以下命令为连接器指定 OpenSSL 库路径。
export LDFLAGS=-L/usr/local/opt/openssl/lib
3. 执行以下命令安装 Node.js 依赖库。
npm install i --unsafe-perm node-rdkafka

步骤2：创建 Topic 和订阅关系

1.


2. 单击 Topic 的 “ID” 进入基本信息页面，获取用户名、密码和地址信息。


3. 在订阅关系页签，新建一个订阅关系（消费组）。
img



步骤2：添加配置文件

module.exports = {
    'sasl_plain_username': 'your_user_name',
    'sasl_plain_password': 'your_user_password',
    'bootstrap_servers': ["xxx.xx.xx.xx:port"],
    'topic_name': 'xxx',
    'group_id': 'xxx'
}
参数
描述
bootstrapServers
接入地址，在控制台的弹性 Topic 基本信息页面获取。



sasl_plain_username
用户名，在控制台的弹性 Topic 基本信息页面获取。
sasl_plain_password
用户密码，在控制台的弹性 Topic 基本信息页面获取。
topic_name
Topic 名称，在控制台的弹性 Topic 基本信息页面获取。
group.id
消费组名称，在控制台的弹性 Topic 的订阅关系列表获取。



步骤3：生产消息

1. 编写生产消息程序 producer.js
   const Kafka = require('node-rdkafka');
   const config = require('./setting');
   console.log("features:" + Kafka.features);
   console.log(Kafka.librdkafkaVersion);

   var producer = new Kafka.Producer({
       'api.version.request': 'true',
       'bootstrap.servers': config['bootstrap_servers'],
       'dr_cb': true,
       'dr_msg_cb': true,
       'security.protocol' : 'SASL_PLAINTEXT',
       'sasl.mechanisms' : 'PLAIN',
       'sasl.username' : config['sasl_plain_username'],
       'sasl.password' : config['sasl_plain_password']
   });

   var connected = false

   producer.setPollInterval(100);

   producer.connect();

   producer.on('ready', function() {
   connected = true
   console.log("connect ok")

   });

   function produce() {
   try {
       producer.produce(
       config['topic_name'],
       new Buffer('Hello CKafka SASL'),
       null,
       Date.now()
       );
   } catch (err) {
       console.error('Error occurred when sending message(s)');
       console.error(err);
   }
   }

   producer.on("disconnected", function() {
   connected = false;
   producer.connect();
   })

   producer.on('event.log', function(event) {
       console.log("event.log", event);
   });

   producer.on("error", function(error) {
       console.log("error:" + error);
   });

   producer.on('delivery-report', function(err, report) {
       console.log("delivery-report: producer ok");
   });
   // Any errors we encounter, including connection errors
   producer.on('event.error', function(err) {
       console.error('event.error:' + err);
   })

   setInterval(produce,1000,"Interval");
2. 执行以下命令发送消息。
node producer.js
3. 查看运行结果。



步骤4：消费消息

1. 创建消费消息程序 consumer.js。
    consumer.on('event.log', function(event) {
       console.log("event.log", event);
   });

   consumer.on('error', function(error) {
       console.log("error:" + error);
   });

   consumer.on('event', function(event) {
           console.log("event:" + event);
   });
   const Kafka = require('node-rdkafka');
   const config = require('./setting');
   console.log(Kafka.features);
   console.log(Kafka.librdkafkaVersion);
   console.log(config)

   var consumer = new Kafka.KafkaConsumer({
       'api.version.request': 'true',
       'bootstrap.servers': config['bootstrap_servers'],
       'security.protocol' : 'SASL_PLAINTEXT',
       'sasl.mechanisms' : 'PLAIN',
       'message.max.bytes': 32000,
       'fetch.message.max.bytes': 32000,
       'max.partition.fetch.bytes': 32000,
       'sasl.username' : config['sasl_plain_username'],
       'sasl.password' : config['sasl_plain_password'],
       'group.id' : config['group_id']
   });

   consumer.connect();

   consumer.on('ready', function() {
   console.log("connect ok");
   consumer.subscribe([config['topic_name']]);
   consumer.consume();
   })

   consumer.on('data', function(data) {
   console.log(data);
   });

   consumer.on('event.log', function(event) {
       console.log("event.log", event);
   });

   consumer.on('error', function(error) {
       console.log("error:" + error);
   });

   consumer.on('event', function(event) {
           console.log("event:" + event);
   });
2. 执行以下命令消费消息。
node consumer.js
3. 查看运行结果。


中国站