Pulsar WebSocket API

一、简介

Pulsar WebSocket API 提供了一种使用没有官方客户端库的语言与 Pulsar 交互的简单方法。通过 WebSocket，您可以发布和使用消息并使用客户端功能矩阵页面上提供的功能。

官方文档：https://pulsar.apache.org/docs/client-libraries-websocket/

二、配置

1、在Broker的8080端口开起

conf/broker.conf

webSocketServiceEnabled=true

2、作为独立组件启用

conf/websocket.conf

configurationMetadataStoreUrl=zk1:2181,zk2:2181,zk3:2181
webServicePort=8080
clusterName=my-cluster

# 如果要开起TLS，需要设置一下参数
tlsEnabled=true
tlsAllowInsecureConnection=false
tlsCertificateFilePath=/path/to/client-websocket.cert.pem
tlsKeyFilePath=/path/to/client-websocket.key-pk8.pem
tlsTrustCertsFilePath=/path/to/ca.cert.pem

bin/pulsar-daemon start websocket

三、Websocket工具

• Chrome插件：PieSocket WebSocket Tester
• 网页工具：http://wstool.js.org/
• Python：pip3 install websocket-client
  • 文档：https://pulsar.apache.org/docs/client-libraries-websocket/#python

四、API Endpoint

1、生产者Endpoint

请求URL

ws://broker-service-url:8080/ws/v2/producer/persistent/:tenant/:namespace/:topic?参数1=值&参数2=值

消息体样本

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

消息体属性

2、消费者Endpoint

请求URL

ws://broker-service-url:8080/ws/v2/consumer/persistent/:tenant/:namespace/:topic/:subscription?参数1=值&参数2=值

消息体样本

{
  "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
    }
  }
}

消息体属性

• 基础属性

  | Key | Type | Required? | Explanation | | ------------------------- | ----------------- | --------- | ------------------------------------------------------------ | | messageId | string | yes | 消息 ID | | payload | string | yes | Base-64 encoded payload | | publishTime | string | yes | 推送时间戳 | | redeliveryCount | number | yes | Number of times this message was already delivered | | properties | key-value pairs | no | Application-defined properties | | key | string | no | Original routing key set by producer | | encryptionContext | EncryptionContext | no | Encryption context that consumers can use to decrypt received messages | | param | string | no | Initialization vector for cipher (Base64 encoding) | | batchSize | string | no | Number of entries in a message (if it is a batch message) | | uncompressedMessageSize | string | no | Message size before compression | | compressionType | string | no | Algorithm used to compress the message payload |

• encryptionContext related parameter

  | Key | Type | Required? | Explanation | | ------ | ----------------------- | --------- | ------------------------------------------------------------ | | keys | key-EncryptionKey pairs | yes | Key in key-EncryptionKey pairs is an encryption key name. Value in key-EncryptionKey pairs is an encryption key object. |

• encryptionKey related parameters

  | Key | Type | Required? | Explanation | | ---------- | --------------- | --------- | -------------------------------- | | keyValue | string | yes | Encryption key (Base64 encoding) | | metadata | key-value pairs | no | Application-defined metadata |