一区二区在线线欧美,久久精品欧美日韩精品,日本亚洲色大成网站ww

本文介紹如何使用Node.js SDK通過接入點接入云消息隊列 Kafka 版并收發消息。

環境配置

安裝GCC
安裝Node.js
重要
Node.js版本必須大于等于4.0.0。
安裝OpenSSL

安裝C++依賴庫

執行以下命令切換到yum源配置目錄/etc/yum.repos.d/。
```
cd /etc/yum.repos.d/
```

創建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

執行以下命令安裝C++依賴庫。
```
yum install librdkafka-devel
```

安裝Node.js依賴庫

執行以下命令為預處理器指定OpenSSL頭文件路徑。

# 配置為您的系統安裝的OpenSSL頭文件路徑。
export CPPFLAGS=-I</usr/local/opt/openssl/include>

執行以下命令為連接器指定OpenSSL庫路徑。

# 配置為您的系統安裝的OpenSSL庫路徑。
export LDFLAGS=-L</usr/local/opt/openssl/lib>

執行以下命令安裝Node.js依賴庫。
```
npm install i --unsafe-perm node-rdkafka
```

說明

在命令行窗口執行whereis openssl命令獲取OpenSSL頭文件路徑和庫路徑。

準備配置

可選：下載SSL根證書。如果是SSL接入點，需下載該證書。
訪問Aliware-kafka-demos，單擊，下載Demo工程到本地并解壓。

在解壓的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 管理頁面獲取。

配置完成后，將配置文件所在文件夾下的全部文件（如果是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',
    'ssl.endpoint.identification.algorithm':'none',
    '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