The Pulsar Node.js client

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

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

对于1.3.0或者更高版本，TypeScript中定义的类型同样可以使用。

安装

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

如果使用TLS加密或TLS认证，URL地址如下：

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

创建客户端

与Pulsar服务端交互之前，需要建立Pulsar客户端。 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 returns 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();
})();

消费者

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

Reader

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

消息

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 allows applications to encrypt messages at producers and decrypt at consumers.

Configuration

如果想在Node.js客户端配置端到端加密功能，那么需要在producer端和consumer端都配置publicKeyPath和privateKeyPath。

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

教程

本节提供详细步骤，说明在 Node.js客户端中如何使用端到端加密功能。

前提条件

• Pulsar C++ 客户端版本为2.7.1或者更高

步骤

1. 创建公钥和私钥密钥对。

   输入

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

2. 创建一个生产者用来发送加密消息。

   输入

   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. 创建消费者接收加密消息。

   输入

   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. 运行消费者接收加密消息。

   输入

   node consumer.js

5. 在一个新的终端窗口中，运行生产者来生成加密消息。

   输入

   node producer.js

   现在你可以看到生产者发送消息，消费者成功收到消息。

   输出

   生产端：

   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

   消费端：

   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