Pulsar WebSocket API

推荐使用单机模式的 Pulsar 进行开发，在环境中启用 WebSocket 服务。

在非单机模式下，有两种方法可以部署 WebSocket 服务：

• 嵌入 Pulsar Broker
• 作为一个的组件

在这种模式下，WebSocket 服务会使用已经在 broker 中运行的 HTTP 服务。 要启用此模式，需在安装目录下的 conf/broker.conf 文件中设置 参数。

作为一个独立的组件

在这种模式下，WebSocket 会作为单独的服务在 Pulsar 上运行。 运行此模式，需在 conf/websocket.conf 文件中进行配置。 You’ll need to set at least the following parameters:

• webServicePort

下面是一个示例：

webServicePort=8080
clusterName=my-cluster

安全设置

在 WebSocket 服务上启用 TLS 加密：

tlsEnabled=true
tlsCertificateFilePath=/path/to/broker.cert.pem
tlsKeyFilePath=/path/to/broker.key-pk8.pem
tlsTrustCertsFilePath=/path/to/ca.cert.pem

配置完成后，你可以使用 命令来启动服务：

$ bin/pulsar-daemon start websocket

Pulsar 的 WebSocket API 提供三个端点，用于生产消息、和阅读消息。

所有通过 WebSocket API 的数据都使用 JSON 进行交互。

认证

Browser javascript WebSocket client

使用查询参数 token 传送身份验证令牌。

ws://broker-service-url:8080/path?token=token

Producer 端

Producer 端需要在 URL 中指定租户、命名空间和 topic，例如：

ws://broker-service-url:8080/ws/v2/producer/persistent/:tenant/:namespace/:topic

查询参数

发布消息

{
  "payload": "SGVsbG8gV29ybGQ=",
  "properties": {"key1": "value1", "key2": "value2"},
  "context": "1"
}

响应成功示例

{
   "result": "ok",
   "messageId": "CAAQAw==",
   "context": "1"
 }

响应失败示例

 {
   "result": "send-error:3",
   "errorMsg": "Failed to de-serialize from JSON",
   "context": "1"
 }

Concumer 端要求在 URL 中指定租户、命名空间、topic 和订阅：

查询参数

接收消息

Server will push messages on the WebSocket session:

{
 "messageId": "CAMQADAA",
  "payload": "hvXcJvHW7kOSrUn17P2q71RA5SdiXwZBqw==",
  "properties": {},
  "publishTime": "2021-10-29T16:01:38.967-07:00",
  "redeliveryCount": 0,
  "encryptionContext": {
    "keys": {
      "client-rsa.pem": {
        "keyValue": "jEuwS+PeUzmCo7IfLNxqoj4h7txbLjCQjkwpaw5AWJfZ2xoIdMkOuWDkOsqgFmWwxiecakS6GOZHs94x3sxzKHQx9Oe1jpwBg2e7L4fd26pp+WmAiLm/ArZJo6JotTeFSvKO3u/yQtGTZojDDQxiqFOQ1ZbMdtMZA8DpSMuq+Zx7PqLo43UdW1+krjQfE5WD+y+qE3LJQfwyVDnXxoRtqWLpVsAROlN2LxaMbaftv5HckoejJoB4xpf/dPOUqhnRstwQHf6klKT5iNhjsY4usACt78uILT0pEPd14h8wEBidBz/vAlC/zVMEqiDVzgNS7dqEYS4iHbf7cnWVCn3Hxw==",
        "metadata": {}
    },
    "param": "Tfu1PxVm6S9D3+Hk",
    "compressionType": "NONE",
    "uncompressedMessageSize": 0,
    "batchSize": {
      "empty": false,
      "present": true
    }
  }

Below are the parameters in the WebSocket consumer response.

• General parameters

  Key	类型	是否必需	说明
  messageId	string	是	消息 ID
  payload	string	是	Base-64 编码的负载
  publishTime	string	是	发布时间戳
  redeliveryCount	number	是	此消息已经发送的次数
  properties	键值对	否	应用程序定义的属性
  key	string	否	Producer 设置的原始路由密钥
  	EncryptionContext	否	Encryption context that consumers can use to decrypt received messages
  param	string	否	Initialization vector for cipher (Base64 encoding)
  batchSize	string	否	Number of entries in a message (if it is a batch message)
  uncompressedMessageSize	string	否	Message size before compression
  compressionType	string	否	Algorithm used to compress the message payload

• encryptionContext related parameter

• encryptionKey related parameters

  Key	类型	是否必需	说明
  keyValue	string	是	Encryption key (Base64 encoding)
  metadata	键值对	否	Application-defined metadata

ACK 确认消息

Consumer needs to acknowledge the successful processing of the message to have the Pulsar broker delete it.

{
  "messageId": "CAAQAw=="
}

ACK 否认消息

{
  "type": "negativeAcknowledge",
  "messageId": "CAAQAw=="
}

流量控制

推送模式

默认情况下（pullMode=false），consumer 端使用 receiverQueueSize 参数设置内部接收队列的大小，并限制传递到 WebSocket 客户端的未确认消息数。 在这种模式下，如果不发送消息确认，发送到 WebSocket 客户端的消息达到 receiverQueueSize时，Pulsar WebSocket 将停止发送消息。

拉取模式

如果设置 pullMode 为 true，则 WebSocket 客户端需要使用 permit 命令允许 Pulsar WebSocket 服务发送更多消息。

{
  "type": "permit",
  "permitMessages": 100
}

注意：在这种模式下，可以在不同的连接中确认消息。

检查是否到达主题的结尾

消费者可以通过发送 isEndOfTopic 请求来检查它是否已经到达主题的结尾。

Request

{
  "type": "isEndOfTopic"
}

Response

{
   "endOfTopic": "true/false"
 }

Reader 端

The reader endpoint requires you to specify a tenant, namespace, and topic in the URL:

ws://broker-service-url:8080/ws/v2/reader/persistent/:tenant/:namespace/:topic

查询参数

接收消息

{
  "messageId": "CAAQAw==",
  "payload": "SGVsbG8gV29ybGQ=",
  "properties": {"key1": "value1", "key2": "value2"},
  "publishTime": "2016-08-30 16:45:57.785",
  "redeliveryCount": 4
}

ACK 确认消息

In WebSocket, Reader needs to acknowledge the successful processing of the message to have the Pulsar WebSocket service update the number of pending messages. If you don’t send acknowledgements, Pulsar WebSocket service will stop sending messages after reaching the pendingMessages limit.

检查是否到达主题的结尾

消费者可以通过发送 isEndOfTopic 请求来检查它是否已经到达主题的结尾。

Request

{
  "type": "isEndOfTopic"
}

Response

{
   "endOfTopic": "true/false"
 }

错误代码

In case of error the server will close the WebSocket session using the following error codes:

应用程序负责在后台重新建立 WebSocket 连接。

Below you’ll find code examples for the Pulsar WebSocket API in and Node.js.

This example uses the package. You can install it using pip:

$ pip install websocket-client

You can also download it from .

Python producer

Here’s an example Python producer that sends a simple message to a Pulsar :

#!/usr/bin/python
#coding=utf-8
import websocket, base64, json
# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'
TOPIC = scheme + '://localhost:8080/ws/v2/producer/persistent/public/default/my-topic'
ws = websocket.create_connection(TOPIC)
# Send one message as JSON
ws.send(json.dumps({
    'payload' : base64.b64encode('Hello World'),
    'properties': {
        'key1' : 'value1',
    },
    'context' : 5
}))
response =  json.loads(ws.recv())
if response['result'] == 'ok':
    print 'Message published successfully'
else:
    print 'Failed to publish message:', response
ws.close()

Python consumer

Here’s an example Python consumer that listens on a Pulsar topic and prints the message ID whenever a message arrives:

#!/usr/bin/python
#coding=utf-8
import websocket, base64, json
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'
TOPIC = scheme + '://localhost:8080/ws/v2/consumer/persistent/public/default/my-topic/my-sub'
ws = websocket.create_connection(TOPIC)
while True:
    msg = json.loads(ws.recv())
    if not msg: break
    print "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload']))
    # Acknowledge successful processing
    ws.send(json.dumps({'messageId' : msg['messageId']}))
ws.close()

Python reader

Here’s an example Python reader that listens on a Pulsar topic and prints the message ID whenever a message arrives:

#!/usr/bin/python
#coding=utf-8
import websocket, base64, json
# If set enableTLS to true, your have to set tlsEnabled to true in conf/websocket.conf.
enable_TLS = False
scheme = 'ws'
if enable_TLS:
    scheme = 'wss'
TOPIC = scheme + '://localhost:8080/ws/v2/reader/persistent/public/default/my-topic'
ws = websocket.create_connection(TOPIC)
while True:
    msg = json.loads(ws.recv())
    if not msg: break
    print "Received: {} - payload: {}".format(msg, base64.b64decode(msg['payload']))
    # Acknowledge successful processing
    ws.send(json.dumps({'messageId' : msg['messageId']}))
ws.close()

Node.js

This example uses the package. You can install it using npm:

$ npm install ws

Node.js producer

const WebSocket = require('ws');
// 如果设置启用TLS 为 true，您必须设置 tls 启用为 conf/websocket.conf。
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/producer/persistent/public/default/my-topic`;
const ws = new WebSocket(topic);
var message = {
  "payload" : new Buffer("Hello World").toString('base64'),
  "properties": {
    "key1" : "value1",
    "key2" : "value2"
  },
  "context" : "1"
};
ws.on('open', function() {
  // Send one message
  ws.send(JSON.stringify(message));
});
ws.on('message', function(message) {
  console.log('received ack: %s', message);
});

Node.js consumer

Here’s an example Node.js consumer that listens on the same topic used by the producer above:

NodeJS reader

const WebSocket = require('ws');
// 如果设置启用TLS 为 true，您必须设置 tls 启用为 conf/websocket.conf。
const enableTLS = false;
const topic = `${enableTLS ? 'wss' : 'ws'}://localhost:8080/ws/v2/reader/persistent/public/default/my-topic`;
const ws = new WebSocket(topic);
ws.on('message', function(message) {
    var receiveMsg = JSON.parse(message);
    console.log('Received: %s - payload: %s', message, new Buffer(receiveMsg.payload, 'base64').toString());
    var ackMsg = {"messageId" : receiveMsg.messageId};
});