Pulsar mqtt协议插件安装
- git clone GitHub - tw-iot/mop: MQTT on Pulsar implemented using Pulsar Protocol Handler
- 去下载一个 pulsar-protocol-handler-mqtt-2.8.3.5.nar （.nar的文件）
官方下载地址:
https://github.com/streamnative/mop/releases/
下载对应版本的nar包,如: pulsar-protocol-handler-mqtt-2.8.1.0.nar
放到apache-pulsar-2.8.1/protocols目录下,没有protocols目录就创建一个
配置Mqtt
添加如下属性,在pulsar配置文件中,
集群配置:conf/broker.conf或单机配置:conf/standalone.conf
messagingProtocols=mqtt
protocolHandlerDirectory=./protocols
#brokerServicePort=6650
mqttListeners=mqtt://127.0.0.1:1883
advertisedAddress=127.0.0.1
#开启代理
mqttProxyEnable=true
mqttProxyPort=5682
MoP 目前支持basic和token身份验证方法。身份验证方法适用于任何基于令牌的tokenPulsar 身份验证提供程序，例如内置的 JWT 提供程序和外部令牌身份验证提供程序，例如biscuit-pulsar。
要对 MQTT 连接使用身份验证，您的 Pulsar 集群必须已经启用了身份验证，并配置了您选择的身份验证提供程序。
然后，您可以使用以下配置属性启用 MQTT 身份验证：
mqttAuthenticationEnabled=true
mqttAuthenticationMethods=token
mqttAuthenticationMethods如果您希望启用多个身份验证提供程序，可以将其设置为逗号分隔的列表。MoP 将在验证客户端连接时按顺序尝试每个。
启用身份验证后，MoP 当前将不允许匿名连接。
验证客户端连接
基本认证
设置 MQTT 用户名和密码客户端设置。
令牌认证 将 MQTT 密码设置为令牌正文，当前用户名将被忽略
启用授权
MoP 目前支持授权。启用授权后，MoP 会检查已认证的角色是否具有发布/订阅主题的能力，例如：发送消息时，您需要拥有该主题的生产权限。订阅主题时，需要拥有该主题的消费权限。您可以参考此处授予权限。
然后，您可以使用以下配置属性启用 MQTT 授权：
mqttAuthorizationEnabled=true
| Property | Default value | Comment |
| messagingProtocols | mqtt | available values [mqtt, kafka, amqp] |
| protocolHandlerDirectory | ./protocols | Protocol handler directory |
| mqttListeners | MoP listener address. available listener prefix: [mqtt, mqtt+ssl, mqtt+ssl+psk] | |
| advertisedAddress | Keep the same as Pulsar broker's advertisedAddress | |
| mqttAuthenticationEnabled | false | Enable mqtt authentication |
| mqttAuthenticationMethods | null | Mqtt authentication methods, available values [basic, token] |
| defaultTenant | public | Default Pulsar tenant that the MQTT server used |
| defaultNamespace | default | Default Pulsar namespace that the MQTT server used |
| defaultTopicDomain | persistent | Default Pulsar topic domain that the MQTT server used |
| mqttMessageMaxLength | 8092 | Max length for per message. |
MoP Proxy
| Property | Default value | Comment |
| mqttProxyEnabled | false | Enable MoP proxy |
| mqttProxyPort | 5682 | Default MoP proxy port |
| mqttProxyTlsEnabled | false | Enable MoP proxy TLS or not |
| mqttProxyTlsPort | 5683 | Default mqtt TLS port |
| mqttProxyTlsPskPort | 5684 | Default mqtt proxy tls psk port |
###---- mqtt
messagingProtocols=mqtt
protocolHandlerDirectory=./protocols
#brokerServicePort=6650
mqttListeners=mqtt://127.0.0.1:1884
advertisedAddress=127.0.0.1
#开启代理
mqttProxyEnable=true
mqttProxyPort=5682
mqttAuthorizationEnabled=true
mqttAuthenticationEnabled=true
mqttAuthenticationMethods=token
mqttMessageMaxLength=80920
pulsar 执行命令创建token,可能会找不到jdk ，
解决方案：只需要把你java安装路径下bin下的java软连接到usr/bin下即可
[root@localhost ~]# ln -s /usr/java/bin/java /usr/bin/java
建议配置conf/standalone.conf
是否自动创建
allowAutoTopicCreation=false
删除不活跃的topic
brokerDeleteInactiveTopicsEnabled=false
强制删除Namespace 开发和测试环境配置成true方便清理
forceDeleteNamespaceAllowed=false
强制删除Tenant 开发和测试环境配置成true方便清理
forceDeleteTenantAllowed=false
这个很重要，根据业务情况配，我是7天= 7 * 24 * 60
topic如果存在很多订阅没有消费，那么bk存储也不会删除
解决：
1、删除订阅 下面这个
2、配置TTL
subscriptionExpirationTimeMinutes=10080
