「干货」Node-RED MQTT IN节点动态用法

静态连接

非动态连接的“MQTT IN”不支持输入，需要完整配置连接参数以及订阅地址；

配置连接MQTT服务器认证参数。

动态连接

复制下段JSON文本到Node-RED导入窗口，防止参数输入错误导致连接失败；

[{"id":"9a5655d35c064b4a","type":"inject","z":"f2918211b951343b","name":"手动连接","props":[{"p":"action","v":"connect","vt":"str"},{"p":"broker","v":"{\"type\":\"mqtt-broker\",\"broker\":\"192.168.88.5\",\"port\":1883,\"force\":true,\"clientid\":\"nodered_debug\",\"autoConnect\":false,\"protocolVersion\":\"3.1.1\",\"keepalive\":\"120\",\"username\":\"admin_nodered_debug\",\"password\":\"2E04DD7795E3B09D0C18350C14B204AC\"}","vt":"json"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":880,"y":120,"wires":[["2fbfe744fc88bd59"]]},{"id":"2fbfe744fc88bd59","type":"mqtt in","z":"f2918211b951343b","name":"","topic":"","qos":"2","datatype":"auto","broker":"b68d71bd98332c07","nl":false,"rap":true,"rh":0,"inputs":1,"x":1090,"y":180,"wires":[["23764904f218a6d5"]]},{"id":"b68d71bd98332c07","type":"mqtt-broker","name":"动态连接MQTT","broker":"192.168.88.5","port":"1883","clientid":"","autoConnect":false,"usetls":false,"protocolVersion":"4","keepalive":"60","cleansession":true,"birthTopic":"","birthQos":"0","birthPayload":"","birthMsg":{},"closeTopic":"","closeQos":"0","closePayload":"","closeMsg":{},"willTopic":"","willQos":"0","willPayload":"","willMsg":{},"sessionExpiry":""}]

导入完成后会输出成功日志如下图。

打开刚才导入的“MQTT IN”节点，与常规连接一样，使用动态连接也需要配置一个用于存储动态连接参数的空间，需要配置一个空的“服务端”，可以像笔者一样将IP地址配置为127.0.0.1，端口为8888（也可以为其他任意值，不能为空，其余参数可以为空），实际节点在连接时不会优先使用输入的参数而是使用传入的参数连接（导入的“手动连接”JSON中对应的字段参数），缺少的参数会采用此处默认参数（根据说明得出，并未实际测试），需要注意使用动态连接节点应当通过上一节点驱动而不是底层自动驱动因此需要取消勾选“Connect automatically”属性。

如何配置

打开导入的“手动连接”，一共配置了两个字段，一个用于存在请求信息（msg.action，相当于告知MQTT IN节点应当如何解析数据），一个用于存储连接参数（msg.broker）。

msg.action支持以下功能：

“connect”，连接请求，用于告知MQTT IN节点使用msg.broker参数连接服务器；

“subscribe”，订阅请求，用于告知MQTT IN节点使用msg.topic订阅服务器的主题；

“unsubscribe”，取消订阅，与“subscribe”对应使用，取消主题的订阅；

“disconnect”，断开服务器连接；

这里先说明“connect”方式需要的参数，需要见参数存在broker字段，因此需要选择使用JOSN，如下图所示：

“type”，用于标记参数，可以任意配置，也可缺省；

“broker”，用于存储MQTT服务器地址；

“port”，用于存储MQTT服务器端口；

“force”，强制断开连接，一般使能，缺省为失能，可能返回错误（没有断开服务器连接时再次请求连接）；

“clientid”，MQTT连接三要素的客户端ID；

“autoConnect”，自动连接使能，建议关闭，避免发生错误；

“protocolVersion”，MQTT连接的版本；

“keepalive”，MQTT的保活心跳，建议配置为120；

“username”，MQTT连接三要素的用户名；

“password”，MQTT连接三要素的密钥；

配置以上参数MQTT动态连接即可实现，下面是连接的服务器日志，由于添加了数据库认证会比无认证的多几个认证日志；

点击触发后“MQTT IN”节点会提示已连接；

断开连接就要比连接要简单得多，只要传入请求方式“msg.action”等于“disconnect”，Node-RED上面会连接很多的MQTT服务器，断开对应的连接就需要再次使用到上面配置的“空服务器”。

动态订阅与取消订阅

上面已经提到订阅请求需要配置“msg.action”为“subscribe”，这时不再使用“broker”存储参数，而是使用“topic”存储，这里的topic需要存储两个信息：一个是订阅的地址（msg.topic.topic），一个是订阅的等级（msg.topic.qos），如下图：

取消订阅只需要令“msg.action”为“unsubscribe”，其余与订阅一致。

动态连接测试

使用动态连接方式连接自建MQTT服务器，并通过“123”主题回传当前时间戳测试；

Debug调试窗口输入时间戳，动态连接订阅有效；

MQTT服务器调试日志：

笔者工程文件如下，导入Node-RED使用：

[{"id":"a906262bda80a272","type":"inject","z":"5d4fa2aa0a8f4fcf","name":"手动连接","props":[{"p":"action","v":"connect","vt":"str"},{"p":"broker","v":"{\"type\":\"mqtt-broker\",\"broker\":\"192.168.88.5\",\"port\":1883,\"force\":true,\"clientid\":\"nodered_debug\",\"autoConnect\":false,\"protocolVersion\":\"3.1.1\",\"keepalive\":\"120\",\"username\":\"admin_nodered_debug\",\"password\":\"2E04DD7795E3B09D0C18350C14B204AC\"}","vt":"json"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":180,"y":440,"wires":[["ce21e6f7efcbbea1"]]},{"id":"ce21e6f7efcbbea1","type":"mqtt in","z":"5d4fa2aa0a8f4fcf","name":"","topic":"","qos":"2","datatype":"auto","broker":"b68d71bd98332c07","nl":false,"rap":true,"rh":0,"inputs":1,"x":370,"y":520,"wires":[["666a19ab1c8cb2bd"]]},{"id":"666a19ab1c8cb2bd","type":"debug","z":"5d4fa2aa0a8f4fcf","name":"","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":530,"y":520,"wires":[]},{"id":"553cee9604c2a8ff","type":"inject","z":"5d4fa2aa0a8f4fcf","name":"断开连接","props":[{"p":"action","v":"disconnect","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":180,"y":620,"wires":[["ce21e6f7efcbbea1"]]},{"id":"07a0589c3a502788","type":"inject","z":"5d4fa2aa0a8f4fcf","name":"手动订阅","props":[{"p":"action","v":"subscribe","vt":"str"},{"p":"topic","v":"{\"topic\":\"123\",\"qos\":0}","vt":"json"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":180,"y":500,"wires":[["ce21e6f7efcbbea1"]]},{"id":"d81035879f55d047","type":"inject","z":"5d4fa2aa0a8f4fcf","name":"手动断开订阅","props":[{"p":"action","v":"unsubscribe","vt":"str"},{"p":"topic","v":"{\"topic\":\"123654\",\"qos\":0}","vt":"json"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","x":170,"y":560,"wires":[["ce21e6f7efcbbea1"]]},{"id":"b8542c59cc5a49bd","type":"mqtt out","z":"5d4fa2aa0a8f4fcf","name":"","topic":"","qos":"","retain":"","respTopic":"","contentType":"","userProps":"","correl":"","expiry":"","broker":"b68d71bd98332c07","x":370,"y":740,"wires":[]},{"id":"77600c8d5cc27a7e","type":"inject","z":"5d4fa2aa0a8f4fcf","name":"动态订阅测试","props":[{"p":"topic","vt":"str"},{"p":"payload"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"123","payloadType":"date","x":180,"y":740,"wires":[["b8542c59cc5a49bd"]]},{"id":"b68d71bd98332c07","type":"mqtt-broker","name":"动态连接MQTT","broker":"127.0.0.1","port":"8888","clientid":"","autoConnect":false,"usetls":false,"protocolVersion":"4","keepalive":"60","cleansession":true,"birthTopic":"","birthQos":"0","birthPayload":"","birthMsg":{},"closeTopic":"","closeQos":"0","closePayload":"","closeMsg":{},"willTopic":"","willQos":"0","willPayload":"","willMsg":{},"sessionExpiry":""}]