Use the SDK for Node.js to connect to ApsaraMQ for Kafka to send and receive messages - ApsaraMQ for Kafka

Utilisez la bibliothèque node-rdkafka pour connecter une application Node.js à ApsaraMQ for Kafka afin de produire et de consommer des messages.

Prérequis

Avant de commencer, vérifiez que vous disposez des éléments suivants :

GCC installé
Node.js version 4.0.0 ou ultérieure installée
OpenSSL installé
Une instance ApsaraMQ for Kafka avec un topic et un groupe de consommateurs créés
(SSL uniquement) Le certificat racine SSL téléchargé et enregistré sous le nom ca-cert.pem

Installer la bibliothèque C++

Le package node-rdkafka dépend de librdkafka, une bibliothèque native C++. Installez-la via le dépôt Confluent.

Accédez au répertoire des dépôts yum :
```
   cd /etc/yum.repos.d/
```

Créez un fichier de configuration de dépôt nommé confluent.repo avec le contenu suivant :

   [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

Installez librdkafka :
```
   yum install librdkafka-devel
```

Installer la bibliothèque Node.js

Définissez les chemins d'accès aux en-têtes et bibliothèques OpenSSL pour la compilation native :

   # Replace with your actual OpenSSL paths.
   # Run `whereis openssl` to find these paths.
   export CPPFLAGS=-I/usr/local/opt/openssl/include
   export LDFLAGS=-L/usr/local/opt/openssl/lib

Installez node-rdkafka :

   npm install i --unsafe-perm node-rdkafka

Configurer la connexion

Créez un fichier setting.js contenant vos paramètres de connexion. Choisissez la configuration correspondant à votre type d'endpoint.

Paramètres de connexion

Paramètre	Description	Où le trouver
`bootstrap_servers`	Endpoint de votre instance ApsaraMQ for Kafka	Section Endpoint Information sur la page Instance Details dans la console ApsaraMQ for Kafka
`topic_name`	Nom du topic vers lequel produire ou depuis lequel consommer	Page Topics dans la console ApsaraMQ for Kafka
`consumer_id`	ID du groupe de consommateurs	Page Groups dans la console ApsaraMQ for Kafka
`sasl_plain_username`	Nom d'utilisateur SASL (endpoint SSL uniquement)	Champ Username dans la section Configuration Information sur la page Instance Details
`sasl_plain_password`	Mot de passe SASL (endpoint SSL uniquement)	Champ Password dans la section Configuration Information sur la page Instance Details

Si la liste de contrôle d'accès (ACL) n'est pas activée sur votre instance, récupérez le nom d'utilisateur et le mot de passe Simple Authentication and Security Layer (SASL) dans les champs Username et Password de la section Configuration Information sur la page Instance Details dans la console ApsaraMQ for Kafka . Si l'ACL est activée, assurez-vous que l'utilisateur SASL est autorisé à produire et consommer des messages. Pour plus de détails, consultez la rubrique Grant permissions to SASL users .

Endpoint par défaut

module.exports = {
    'bootstrap_servers': ["<bootstrap-servers>"],
    'topic_name': '<topic-name>',
    'consumer_id': '<consumer-group-id>'
}

Endpoint SSL

module.exports = {
    'sasl_plain_username': '<sasl-username>',
    'sasl_plain_password': '<sasl-password>',
    'bootstrap_servers': ["<bootstrap-servers>"],
    'topic_name': '<topic-name>',
    'consumer_id': '<consumer-group-id>'
}

Après avoir configuré setting.js, téléchargez-le ainsi que tous les fichiers du dossier de démonstration correspondant sur votre serveur. Si vous utilisez l'endpoint SSL, incluez le fichier de certificat racine SSL (ca-cert.pem) dans le même répertoire.

Conseil : Téléchargez le projet de démonstration complet depuis le dépôt aliware-kafka-demos . Les exemples Node.js se trouvent dans le dossier kafka-nodejs-demo , organisés par type d'endpoint.

Produire des messages

Enregistrez le code suivant sous le nom producer.js. Cet exemple envoie un message chaque seconde et journalise les rapports de livraison.

Exécutez le producteur :

node producer.js

Endpoint par défaut

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',   // Negotiate API version with the broker automatically
    'bootstrap.servers': config['bootstrap_servers'],
    'dr_cb': true,                   // Enable delivery report callback
    'dr_msg_cb': true                // Include message payload in delivery reports
});

var connected = false

producer.setPollInterval(100);       // Poll for delivery reports every 100 ms

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,                          // Partition (null = auto-assign)
      new Buffer('Hello Ali Kafka'), // Message payload
      null,                          // Message key
      Date.now()                     // Timestamp
    );
  } 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");

Endpoint 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,

    // SASL/SSL authentication
    'security.protocol': 'sasl_ssl',
    'ssl.ca.location': './ca-cert.pem',         // Path to the SSL root certificate
    'sasl.mechanisms': 'PLAIN',
    'ssl.endpoint.identification.algorithm': 'none',  // Disable hostname verification
    '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");

Consommer des messages

Enregistrez le code suivant sous le nom consumer.js. Cet exemple s'abonne à un topic et affiche chaque message dans la console.

Exécutez le consommateur :

node consumer.js

Endpoint par défaut

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);
});

Endpoint 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'],

    // SASL/SSL authentication
    'security.protocol': 'sasl_ssl',
    'ssl.endpoint.identification.algorithm': 'none',
    'ssl.ca.location': './ca-cert.pem',
    'sasl.mechanisms': 'PLAIN',

    // Message size limits
    '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);
});