The Pulsar Node.js client

Pulsar Node.js 客户端可用于在 Node.js 中创建 Pulsar producer、consumer 和 reader。

Node.js 客户端中 producer、consumer 和 reader 中的所有方法都对线程可用。

安装

You can install the pulsar-client library via npm.

要求

Pulsar Node.js client library is based on the C++ client library. Follow these instructions and install the Pulsar C++ client library.

兼容性

Compatibility between each version of the Node.js client and the C++ client is as follows:

If an incompatible version of the C++ client is installed, you may fail to build or run this library.

使用 npm 安装

Install the pulsar-client library via npm:

$ npm install pulsar-client

Note

因为这个库使用 node-addon-api 模块来包装 C++ 库，因此只能在 Node.js 10.x 或更高版本中运行。

连接URL

To connect to Pulsar using client libraries, you need to specify a Pulsar protocol URL.

Pulsar protocol URLs are assigned to specific clusters, use the pulsar scheme and have a default port of 6650. Here is an example for localhost:

pulsar://localhost:6650

A URL for a production Pulsar cluster may look something like this:

pulsar://pulsar.us-west.example.com:6650

If you are using TLS encryption or TLS Authentication, the URL will look like something like this:

pulsar+ssl://pulsar.us-west.example.com:6651

创建客户端

In order to interact with Pulsar, you will first need a client object. You can create a client instance using a new operator and the Client method, passing in a client options object (more on configuration below).

Here is an example:

const Pulsar = require('pulsar-client');
(async () => {
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  await client.close();
})();

客户端配置

The following configurable parameters are available for Pulsar clients:

Producers

Pulsar producers publish messages to Pulsar topics. You can configure Node.js producers using a producer configuration object.

Here is an example:

const producer = await client.createProducer({
  topic: 'my-topic',
});
await producer.send({
  data: Buffer.from("Hello, Pulsar"),
});
await producer.close();

Promise 操作

When you create a new Pulsar producer, the operation will return Promise object and get producer instance or an error through executor function.
In this example, using await operator instead of executor function.

Producer operations

Pulsar Node.js producers have the following methods available:

Producer 配置

生产者示例

This example creates a Node.js producer for the my-topic topic and sends 10 messages to that topic:

const Pulsar = require('pulsar-client');
(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  // Create a producer
  const producer = await client.createProducer({
    topic: 'my-topic',
  });
  // Send messages
  for (let i = 0; i < 10; i += 1) {
    const msg = `my-message-${i}`;
    producer.send({
      data: Buffer.from(msg),
    });
    console.log(`Sent message: ${msg}`);
  }
  await producer.flush();
  await producer.close();
  await client.close();
})();

Consumers

Pulsar consumers subscribe to one or more Pulsar topics and listen for incoming messages produced on that topic/those topics. You can configure Node.js consumers using a consumer configuration object.

Here is an example:

const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
});
const msg = await consumer.receive();
console.log(msg.getData().toString());
consumer.acknowledge(msg);
await consumer.close();

Promise 操作

When you create a new Pulsar consumer, the operation will return Promise object and get consumer instance or an error through executor function.
In this example, using await operator instead of executor function.

消费者选项

Pulsar Node.js consumers have the following methods available:

Consumer configuration

消费者示例

This example creates a Node.js consumer with the my-subscription subscription on the my-topic topic, receives messages, prints the content that arrive, and acknowledges each message to the Pulsar broker for 10 times:

const Pulsar = require('pulsar-client');
(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  // Create a consumer
  const consumer = await client.subscribe({
    topic: 'my-topic',
    subscription: 'my-subscription',
    subscriptionType: 'Exclusive',
  });
  // Receive messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await consumer.receive();
    console.log(msg.getData().toString());
    consumer.acknowledge(msg);
  }
  await consumer.close();
  await client.close();
})();

Instead a consumer can be created with listener to process messages.

// Create a consumer
const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
  subscriptionType: 'Exclusive',
  listener: (msg, msgConsumer) => {
    console.log(msg.getData().toString());
    msgConsumer.acknowledge(msg);
  },
});

Readers

Pulsar readers process messages from Pulsar topics. Readers are different from consumers because with readers you need to explicitly specify which message in the stream you want to begin with (consumers, on the other hand, automatically begin with the most recently unacked message). You can configure Node.js readers using a reader configuration object.

Here is an example:

const reader = await client.createReader({
  topic: 'my-topic',
  startMessageId: Pulsar.MessageId.earliest(),
});
const msg = await reader.readNext();
console.log(msg.getData().toString());
await reader.close();

Reader operations

Pulsar Node.js readers have the following methods available:

Reader configuration

Reader 示例

This example creates a Node.js reader with the my-topic topic, reads messages, and prints the content that arrive for 10 times:

const Pulsar = require('pulsar-client');
(async () => {
  // Create a client
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
    operationTimeoutSeconds: 30,
  });
  // Create a reader
  const reader = await client.createReader({
    topic: 'my-topic',
    startMessageId: Pulsar.MessageId.earliest(),
  });
  // read messages
  for (let i = 0; i < 10; i += 1) {
    const msg = await reader.readNext();
    console.log(msg.getData().toString());
  }
  await reader.close();
  await client.close();
})();

Messages

In Pulsar Node.js client, you have to construct producer message object for producer.

Here is an example message:

const msg = {
  data: Buffer.from('Hello, Pulsar'),
  partitionKey: 'key1',
  properties: {
    'foo': 'bar',
  },
  eventTimestamp: Date.now(),
  replicationClusters: [
    'cluster1',
    'cluster2',
  ],
}
await producer.send(msg);

The following keys are available for producer message objects:

消息对象操作

In Pulsar Node.js client, you can receive (or read) message object as consumer (or reader).

The message object have the following methods available:

消息 ID 对象操作

In Pulsar Node.js client, you can get message id object from message object.

The message id object have the following methods available:

The client has static method of message id object. You can access it as Pulsar.MessageId.someStaticMethod too.

The following static methods are available for the message id object:

End-to-end encryption

End-to-end encryption allows applications to encrypt messages at producers and decrypt at consumers.

Configuration

If you want to use the end-to-end encryption feature in the Node.js client, you need to configure publicKeyPath and privateKeyPath for both producer and consumer.

publicKeyPath: "./public.pem"
privateKeyPath: "./private.pem"

Tutorial

This section provides step-by-step instructions on how to use the end-to-end encryption feature in the Node.js client.

前提条件

• Pulsar C++ client 2.7.1 or later

Step

1. Create both public and private key pairs.

   Input

   openssl genrsa -out private.pem 2048
   openssl rsa -in private.pem -pubout -out public.pem

2. Create a producer to send encrypted messages.

   Input

   const Pulsar = require('pulsar-client');
   (async () => {
     // Create a client
     const client = new Pulsar.Client({
       serviceUrl: 'pulsar://localhost:6650',
       operationTimeoutSeconds: 30,
     });
     // Create a producer
     const producer = await client.createProducer({
       topic: 'persistent://public/default/my-topic',
       sendTimeoutMs: 30000,
       batchingEnabled: true,
       publicKeyPath: "./public.pem",
       privateKeyPath: "./private.pem",
       encryptionKey: "encryption-key"
     });
     console.log(producer.ProducerConfig)
     // Send messages
     for (let i = 0; i < 10; i += 1) {
       const msg = `my-message-${i}`;
       producer.send({
         data: Buffer.from(msg),
       });
       console.log(`Sent message: ${msg}`);
     }
     await producer.flush();
     await producer.close();
     await client.close();
   })();

3. Create a consumer to receive encrypted messages.

   Input

   const Pulsar = require('pulsar-client');
   (async () => {
     // Create a client
     const client = new Pulsar.Client({
       serviceUrl: 'pulsar://172.25.0.3:6650',
       operationTimeoutSeconds: 30
     });
     // Create a consumer
     const consumer = await client.subscribe({
       topic: 'persistent://public/default/my-topic',
       subscription: 'sub1',
       subscriptionType: 'Shared',
       ackTimeoutMs: 10000,
       publicKeyPath: "./public.pem",
       privateKeyPath: "./private.pem"
     });
     console.log(consumer)
     // Receive messages
     for (let i = 0; i < 10; i += 1) {
       const msg = await consumer.receive();
       console.log(msg.getData().toString());
       consumer.acknowledge(msg);
     }
     await consumer.close();
     await client.close();
   })();

4. Run the consumer to receive encrypted messages.

   Input

   node consumer.js

5. In a new terminal tab, run the producer to produce encrypted messages.

   Input

   node producer.js

   Now you can see the producer sends messages and the consumer receives messages successfully.

   输出

   This is from the producer side.

   Sent message: my-message-0
   Sent message: my-message-1
   Sent message: my-message-2
   Sent message: my-message-3
   Sent message: my-message-4
   Sent message: my-message-5
   Sent message: my-message-6
   Sent message: my-message-7
   Sent message: my-message-8
   Sent message: my-message-9

   This is from the consumer side.

   my-message-0
   my-message-1
   my-message-2
   my-message-3
   my-message-4
   my-message-5
   my-message-6
   my-message-7
   my-message-8
   my-message-9