Apache Pulsar Node.js客户端开发指南

概述

Apache Pulsar是一个开源的分布式发布-订阅消息系统，Node.js客户端库允许开发者使用JavaScript/TypeScript语言与Pulsar集群进行交互。本文将详细介绍如何使用Pulsar Node.js客户端创建生产者、消费者和读取器，以及相关的配置选项和最佳实践。

环境准备

安装要求

Pulsar Node.js客户端库基于C++客户端库构建，因此需要先安装Pulsar C++客户端库。Node.js版本要求10.x或更高，因为使用了node-addon-api模块来封装C++库。

版本兼容性

Node.js客户端与C++客户端的版本兼容性如下：

Node.js客户端	C++客户端要求
1.0.0	2.3.0或更高
1.1.0	2.4.0或更高
1.2.0	2.5.0或更高

安装客户端库

使用npm安装pulsar-client库：

npm install pulsar-client

连接配置

连接URL格式

连接Pulsar集群需要使用特定的URL格式：

• 普通连接：pulsar://localhost:6650
• TLS加密连接：pulsar+ssl://pulsar.us-west.example.com:6651

创建客户端实例

const Pulsar = require('pulsar-client');

(async () => {
  const client = new Pulsar.Client({
    serviceUrl: 'pulsar://localhost:6650',
  });
  
  // 使用客户端...
  
  await client.close();
})();

客户端配置选项

参数	描述	默认值
serviceUrl	Pulsar集群连接URL	必填
authentication	认证配置	无
operationTimeoutSeconds	操作超时时间(秒)	30
ioThreads	处理broker连接的线程数	1
messageListenerThreads	消息监听器线程数	1
tlsTrustCertsFilePath	TLS证书路径	无
tlsValidateHostname	是否验证TLS主机名	false

生产者(Producer)

创建生产者

const producer = await client.createProducer({
  topic: 'my-topic',
});

生产者方法

方法	描述	返回类型
send()	发送消息	Promise
flush()	刷新发送队列	Promise
close()	关闭生产者	Promise

生产者配置

参数	描述	默认值
topic	目标主题	必填
sendTimeoutMs	发送超时(毫秒)	30000
compressionType	压缩类型(LZ4/Zlib等)	无压缩
batchingEnabled	是否启用批量发送	true

生产者示例

(async () => {
  const producer = await client.createProducer({topic: 'my-topic'});
  
  for (let i = 0; i < 10; i++) {
    await producer.send({
      data: Buffer.from(`Message ${i}`),
    });
  }
  
  await producer.close();
})();

消费者(Consumer)

创建消费者

const consumer = await client.subscribe({
  topic: 'my-topic',
  subscription: 'my-subscription',
});

消费者方法

方法	描述	返回类型
receive()	接收消息	Promise
acknowledge()	确认消息	void
close()	关闭消费者	Promise

消费者配置

参数	描述	默认值
topic	订阅主题	必填
subscription	订阅名称	必填
subscriptionType	订阅类型(Exclusive/Shared等)	Exclusive
listener	消息监听器函数	无

消费者示例

(async () => {
  const consumer = await client.subscribe({
    topic: 'my-topic',
    subscription: 'my-subscription',
  });

  for (let i = 0; i < 10; i++) {
    const msg = await consumer.receive();
    console.log(msg.getData().toString());
    consumer.acknowledge(msg);
  }

  await consumer.close();
})();

读取器(Reader)

创建读取器

const reader = await client.createReader({
  topic: 'my-topic',
  startMessageId: Pulsar.MessageId.earliest(),
});

读取器方法

方法	描述	返回类型
readNext()	读取下一条消息	Promise

读取器配置

参数	描述	默认值
topic	读取主题	必填
startMessageId	起始消息ID	必填

最佳实践

1. 资源管理：始终记得关闭生产者、消费者和客户端，避免资源泄漏
2. 错误处理：妥善处理Promise拒绝情况
3. 性能调优：根据实际场景调整批量发送和接收队列大小
4. 订阅模式选择：根据业务需求选择合适的订阅类型

总结

Pulsar Node.js客户端提供了强大而灵活的消息处理能力，通过合理配置可以满足各种消息场景需求。本文介绍了客户端的基本使用方法，开发者可以根据实际业务需求进行扩展和优化。