消息队列Kafka版：Node.js SDK收发消息

环境配置

• 安装GCC
• 安装Node.js
  注意 Node.js版本必须大于等于4.0.0。
• 安装OpenSSL

安装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头文件路径。

   # 配置为您的系统安装的OpenSSL头文件路径。
   export CPPFLAGS=-I</usr/local/opt/openssl/include>

2. 执行以下命令为连接器指定OpenSSL库路径。

   # 配置为您的系统安装的OpenSSL库路径。
   export LDFLAGS=-L</usr/local/opt/openssl/lib>

3. 执行以下命令安装Node.js依赖库。

   npm install i --unsafe-perm node-rdkafka

准备配置

1. 可选：下载SSL根证书。如果是SSL接入点，需下载该证书。
2. 访问aliware-kafka-demos，单击，下载Demo工程到本地并解压。
3. 在解压的Demo工程中找到kafka-nodejs-demo文件夹，根据接入点类型打开对应的文件夹，配置setting.js文件。

   module.exports = {
       'sasl_plain_username': 'XXX',
       'sasl_plain_password': 'XXX',
       'bootstrap_servers': ["XXX"],
       'topic_name': 'XXX',
       'consumer_id': 'XXX'
   }

   参数	描述
   sasl_plain_username	SASL用户名。如果是默认接入点，则无此配置项。 说明 如果实例未开启ACL，您可以在消息队列Kafka版控制台的实例详情页面的配置信息区域获取默认的用户名和密码。 如果实例已开启ACL，请确保要使用的SASL用户已被授予向消息队列Kafka版实例收发消息的权限。具体操作，请参见SASL用户授权。
   sasl_plain_password	SASL用户名密码。如果是默认接入点，则无此配置项。
   bootstrap_servers	SSL接入点。您可在消息队列Kafka版控制台的实例详情页面的接入点信息区域获取。
   topic_name	Topic名称。您可在消息队列Kafka版控制台的Topic 管理页面获取。
   consumer_id	Group名称。您可在消息队列Kafka版控制台的Group 管理页面获取。

4. 配置完成后，将配置文件所在文件夹下的全部文件（如果是SSL接入点实例，包含证书SSL根证书文件），上传至服务器Node.js依赖库安装目录下。

发送消息

执行如下命令发送消息。

node 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({
      /*'debug': 'all', */
      'api.version.request': 'true',
      'bootstrap.servers': config['bootstrap_servers'],
      'dr_cb': true,
      'dr_msg_cb': true
  });
  
  var connected = false
  
  producer.setPollInterval(100);
  
  producer.connect();
  
  
  producer.on('ready', function() {
    connected = true
    console.log("connect ok")
  });
  
  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);
  });
  
  function produce() {
    try {
      producer.produce(
        config['topic_name'],   
        null,      
        new Buffer('Hello Ali Kafka'),      
        null,   
        Date.now()
      );
    } catch (err) {
      console.error('A problem occurred when sending our message');
      console.error(err);
    }
  
  }
  
  producer.on('delivery-report', function(err, report) {
      console.log("delivery-report: producer ok");
  });
  
  producer.on('event.error', function(err) {
      console.error('event.error:' + err);
  })
  
  setInterval(produce,1000,"Interval");

• SSL接入点

  const Kafka = require('node-rdkafka');
  const config = require('./setting');
  console.log("features:" + Kafka.features);
  console.log(Kafka.librdkafkaVersion);
  
  var producer = new Kafka.Producer({
      /*'debug': 'all', */
      'api.version.request': 'true',
      'bootstrap.servers': config['bootstrap_servers'],
      'dr_cb': true,
      'dr_msg_cb': true,
      'security.protocol' : 'sasl_ssl',
      'ssl.ca.location' : './ca-cert.pem',
      '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 Ali Kafka'),
        null,
        Date.now()
      );
    } catch (err) {
      console.error('A problem occurred when sending our message');
      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");

订阅消息

执行如下命令消费消息。

node consumer.js

代码示例

• 默认接入点

  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({
      /*'debug': 'all',*/
      'api.version.request': 'true',
      'bootstrap.servers': config['bootstrap_servers'],
      'group.id' : config['consumer_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);
  });

• SSL接入点

  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({
      /*'debug': 'all',*/
      'api.version.request': 'true',
      'bootstrap.servers': config['bootstrap_servers'],
      'security.protocol' : 'sasl_ssl',
      'ssl.ca.location' : './ca-cert.pem',
      'sasl.mechanisms' : 'PLAIN',
      'message.max.bytes': 32000,
      'fetch.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['consumer_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);
  });