使用 Node-RED 处理 MQTT 数据

本文将介绍使用 Node-RED 连接到 MQTT 服务器，并对 MQTT 数据进行过滤和处理后再将其发送至 MQTT 服务器的完整操作流程。读者可以快速了解如何使用 Node-RED 对 MQTT 数据进行简单的流处理。

Node-RED 无论是在你本地的电脑上，还是树莓派等设备，亦或是云端服务器，都可以快速安装和使用，下面将使用两种比较常见的安装方式：

使用 npm 进行全局安装：

使用 Docker 进行安装：

如果使用的是 npm 进行的全局安装，那么在提示安装成功后，只需要在全局运行 node-red 命令就可以立即启动 Node-RED。

无论是使用 Docker 还是 npm 在启动成功后，我们只需要打开浏览器，输入当前地址加 1880 端口号，即可打开 Node-RED 的浏览器编辑器页面，例如在本地运行的话，打开浏览器，输入 http://127001:1880，当看到如下图所示页面后，说明 Node-RED 已经成功启动：

本文将使用 EMQ 提供的 免费公共 MQTT 服务器，该服务基于 MQTT 物联网云平台 - EMQX Cloud 创建，服务器接入信息如下：

在下面的功能演示中，我们将提供一个使用 Node-RED 来处理接收到的包含温湿度信息的 JSON 数据，然后对温度值进行规则判断，当温度发生改变的时候，就将当前发生改变的温度值通过 MQTT 再次发送出去的简单使用案例。

我们首先在左侧菜单栏中，拖拽一个 MQTT in 的节点到页面中，双击节点后，右侧出现一个编辑 MQTT 节点的配置页面，我们根据内容提示，新建一个连接信息后，再填入 MQTT 的其它连接信息后，点击 Done 按钮后，即可保存该节点信息。

接入数据：我们拖拽一个 JSON 节点到页面中，可以在 JSON 节点的配置页面中，配置一个 Action，我们设置为 Always convert to JavasScript Object ，因为我们无法确定发送过来的数据是一个 JSON 格式的数据还是一个 JSON 字符串，因此第一步都将接收到的消息进行一个 JSON 转换。配置完成后，我们将该节点与 MQTT in 节点进行连接。

过滤数据

我们配置完成格式化发送过来的消息数据后，我们就可以拖拽一个 filter 节点到页面中，同样双击节点后，在配置页面中配置规则，我们先选择一个 Mode，我们设置为 blcok unless value changes ，过滤规则为需要当前接收到数据的值发生改变，因为目前数据为 JSON 格式，我们判断的是 JSON 数据内的某一个值，因此我们需要在 Property 这里设置值为 msgpayloadtemperature 配置完成后我们点击 Done 按钮来保存数据过滤节点的配置，最后将该节点连接到上一步配置完成后的 JSON 节点。

使用模版

当过滤完数据后，同样拖拽一个 template 节点到页面中，双击节点后来配置模版内容，使过滤完成后的数据，能通过模版将数据进行输出。当然也可以不需要这个步骤，直接将过滤后的数据进行输出。

完成以上对数据的处理和过滤后，最后我们再来将处理完成后的数据使用 MQTT 将其发送出去，拖拽一个 MQTT out 的节点到页面中，填入和 MQTT in 节点相同的连接信息，配置一个用户接收数据的 Topic，最后保存完成后，再将其和 template 节点进行连接，点击右上角的 Deploy 按钮，即可对当前规则应用进行在线部署。

在完成整个流数据处理的功能编排以后，我们使用 MQTT 50 客户端工具 - MQTT X 来测试和验证该功能的可用性。我们新建一个连接，连接到刚才在 Node-RED 中配置的 MQTT 云服务地址，然后输入 MQTT in 节点内的 Topic 来发送一条消息，使 Node-RED 能够接收到我们发送的 MQTT 数据。

然后我们再在 MQTT X 中订阅一个在 MQTT out 节点内配置的 Topic，用于接收处理过的消息数据。当发送一条包含了温湿度的消息数据后，我们可以接收到一条根据我们设定的消息模版发送过来的消息，再次发送就无法接收到。

因为此时温度值没有发生变化，当我们再次修改温度值后，就会发现我们又接收到了一条包含提醒温度值发生变化的消息。

至此，我们完成了安装并使用 Node-RED 连接到 MQTT 云服务，以及对 MQTT 消息数据进行过滤和处理，最后再将处理完成后的数据消息发送至 MQTT 服务器的全部流程。

Node-RED 的交互和使用方式，即用 UI 方式描述通用业务逻辑，可以降低非专业开发人员的上手门槛，使用一个可视化工具快速地创建需要的复杂执行任务，可以通过简单 Node 即节点连接构建出复杂的任务，特别是针对一些物联网的应用场景，都很有帮助。

在使用 MQTT 协议连接 MQTT 服务器时，可能会遇到连接失败的情况。这通常是由于以下几个原因之一：

服务器配置错误：确保您的 MQTT 服务器已经正确配置，并且已经启动。您可以检查 MQTT 服务器的配置文件，确保 MQTT 服务器的端口号、认证方式、消息格式等参数都正确。

客户端配置错误：确保您的客户端已经正确配置，并且已经启动。您可以检查 MQTT 客户端的配置文件，确保 MQTT 客户端的 URL、端口号、认证方式、消息格式等参数都正确。

网络问题：确保您的网络连接正常，并且可以访问 MQTT 服务器。如果您的网络连接不稳定或者无法访问 MQTT 服务器，可能会导致连接失败。

配置文件错误：确保您的配置文件没有错误。如果您的配置文件中有错误或者不正确的参数，可能会导致连接失败。

如果您已经确保了以上几个方面都没有问题，但是仍然无法连接到 MQTT 服务器，您可以尝试以下几个步骤：

检查 MQTT 服务器的配置文件：确保您的 MQTT 服务器的配置文件没有错误。如果您的配置文件中有错误或者不正确的参数，可能会导致连接失败。

检查 MQTT 客户端的配置文件：确保您的 MQTT 客户端的配置文件没有错误。如果您的配置文件中有错误或者不正确的参数，可能会导致连接失败。

检查网络连接：确保您的网络连接正常，并且可以访问 MQTT 服务器。如果您的网络连接不稳定或者无法访问 MQTT 服务器，可能会导致连接失败。

检查 MQTT 服务器的状态：确保您的 MQTT 服务器已经正确启动，并且可以正常工作。如果您的 MQTT 服务器出现了故障，可能会导致连接失败。

如果以上步骤都没有解决问题，您可以尝试查看 MQTT 服务器的日志，以了解更多有关连接失败的信息。同时，您也可以尝试使用其他的 MQTT 客户端或者其他的 MQTT 服务器来连接，以确保您的连接是正常的。

　　MQTT是一个轻量级的消息发布/订阅协议，它是实现基于手机客户端的消息推送服务器的理想解决方案。

　　我们可以从这里下载该项目的实例代码，并且可以找到一个采用PHP书写的服务器端实现。

　　架构如下所示：

　　wmqttjar 是IBM提供的MQTT协议的实现。你可以从如下站点下载它。你可以将该jar包加入你自己的Android应用程序中。

　　Really Small Message Broker (RSMB) ，他是一个简单的MQTT代理，同样由IBM提供。缺省打开1883端口，应用程序当中，它负责接收来自服务器的消息并将其转发给指定的移动设备。

　　SAM是一个针对MQTT写的PHP库。你可以从这个下载它

　　send_mqttphp是一个通过POST接收消息并且通过SAM将消息发送给RSMB的PHP脚本。

　　实例代码：

　　Ø  采用XMPP协议实现Android推送

　　这是我在项目中采用的方案。事实上Google官方的C2DM服务器底层也是采用XMPP协议进行的封装。

　　XMPP(可扩展通讯和表示协议)是基于可扩展标记语言（XML）的协议，它用于即时消息（IM）以及在线探测。这个协议可能最终允许因特网用户向因特网上的其他任何人发送即时消息。

　　androidpn是一个基于XMPP协议的java开源Android push notification实现。它包含了完整的客户端和服务器端。经过源代码研究我发现，该服务器端基本是在另外一个开源工程openfire基础上修改实现的，不过比较郁闷的是androidpn的文档是由韩语写的，所以整个研究过程基本都是读源码。它的实现示意图如下：

　　androidpn客户端需要用到一个基于java的开源XMPP协议包asmack，这个包同样也是基于openfire下的另外一个开源项目smack，不过我们不需要自己编译，可以直接把androidpn客户端里面的asmackjar拿来使用。客户端利用asmack中提供的XMPPConnection类与服务器建立持久连接，并通过该连接进行用户注册和登录认证，同样也是通过这条连接，接收服务器发送的通知。

　　androidpn服务器端也是java语言实现的，基于openfire开源工程，不过它的Web部分采用的是spring框架，这一点与openfire是不同的。Androidpn服务器包含两个部分，一个是侦听在5222端口上的XMPP服务，负责与客户端的XMPPConnection类进行通信，作用是用户注册和身份认证，并发送推送通知消息。另外一部分是Web服务器，采用一个轻量级的HTTP服务器，负责接收用户的Web请求。服务器架构如下：

　　最上层包含四个组成部分，分别是SessionManager，Auth Manager，PresenceManager以及Notification Manager。SessionManager负责管理客户端与服务器之间的会话，Auth Manager负责客户端用户认证管理，Presence Manager负责管理客户端用户的登录状态，NotificationManager负责实现服务器向客户端推送消息功能。

　　服务器端界面如下，分别对应了上述的几个功能模块：

　　发送以后，我们可以在手机端看到接收的消息：

　　

  MQTT 协议 是基于发布/订阅模式的物联网通信协议，凭借简单易实现、支持 QoS、报文小等特点，占据了物联网协议的半壁江山。

  常用于 IOT 物联网和一些需要服务端主动通知客户端的场景。

1 导入依赖

2 创建 MqttHelper 辅助类，设置回调监听

3 连接 MQTT

  连接成功或失败，以及中途的连接掉线，会触发 OnMqttStatusChangeListener 回调

4 MQTT 连接状态监听

5 MQTT 收发消息监听

  onSubMessage 订阅的消息回调，因为存在订阅多个 topic 的情况，所以回调能知道是来自哪个 Topic 的消息；

  onPubMessage 发布的消息回调，用于确认发布的消息是否发送成功。

6 MQTT 订阅 Topic

  需要在 MQTT 连接成功后才能订阅 topic，否则订阅 Topic 不成功，收不到对应消息

7 MQTT 取消订阅 Topic

8 MQTT 发布消息

9 MQTT 断开连接

10 通知设置

  由于 MQTT 启动了一个 Service，而 Android 80 以上对于后台 Service 限制时长 5 秒；所以将 MqttService 绑定到 Notification 上成为了一个前台通知；通知的标题和内容显示可以在 stringsxml 中设置，对应属性如下：

  Android 80 及以上开启前台服务绑定到通知，80 以下默认不启用，可将 mqtt_foreground_notification_low_26 设为 true，将 80 以下设备也开启前台通知服务

  创建 MQTT 实例时需要传送参数 MqttOptions,下面将介绍下部分参数;

1 Topic

  MQTT 是一种发布/订阅的消息协议, 通过设定的主题 Topic,

发布者向 Topic 发送的 payload 负载消息会经过服务器, 转发到所有订阅

该 Topic 的订阅者

   通配符 : 假想移动端消息推送场景,有的系统消息是全体用户接收,有的消息是 Android 或 iOS 设备接收, 又或者是某些消息具体推送到用户,当然, 对应的多种类型消息可以通过多订阅几个对应的 Topic 解决,也可以使用通配符;

  通配符有两个, " + " 和 " # ", 与正斜杠 " / " 组合使用;加号只能表示一级Topic, 井号可以表示任意层级 Topic; 例如: 订阅 Topic为 " System/+ ", 发布者发布的 Topic 可以是 System、System/Android、System/iOS; 但是不能是 System/iOS/123, 而订阅的 Topic 如果是" System/# " 则可以收到;

   注意,只有订阅的 Topic 才可以使用 通配符, 发布和遗嘱的 Topic 不能包含通配符

2 ClientID

  发布者和订阅者都是属于客户端, 客户端与服务端建立连接之后,发送的第一个报文消息必须是 Connect 消息,而 Connect 的消息载荷中必须包含 clientID 客户端唯一标识;

  如果两个客户端的 clientID 一样, 则服务端记录第一个客户端连接之后再收到第二个客户端连接请求,则会向一个客户端发送 Disconnect 报文断开连接, 并连接第二个客户端, 而如果此时设置了自动重连, 第一个客户端再次连接,服务端又断开与第二个的连接, 连上第一个客户端, 如此将导致两个客户端不断的被挤掉重连

  注意: clientID 使用的字符最好是 大小写字母和数字, 长度最好限制在[1, 23] 之间;

3 遗嘱消息

  可选参数, 客户端没有主动向服务端发起 disconnect 断开连接消息,然而服务端检测到和客户端之间的连接已断开, 此时服务端将该客户端设置的遗嘱消息发送出去

  应用场景: 客户端因网络等情况掉线之后, 可以及时通知到所有订阅该遗嘱 Topic 的客户端;

  遗嘱 Topic 中不能存在通配符

4 Session

  客户端和服务端之间建立的会话状态, 一般用于消息保存, 如果设置清除 Session,则每次客户端和服务端建立连接会创建一个新的会话,之前连接中的消息不能恢复,

  而设置不清除会话, 对应发布者发送的 qos 为 1和2 的消息,还未被订阅者接收确认,则需要保存在会话中, 以便订阅者下次连接可以恢复这些消息;

  注意: Session 存储的消息是保存在内容中的, 所以如果不是重要的消息,最好是设置清除 Session, 或者设置 qos = 0;

5 心跳包

  标识客户端传输一次控制报文到下一次传输之间允许的空闲时间；在这段时间内，如果客户端没有其他任何报文发送，必须发送一个 PINGREQ 报文到服务器，而如果服务端在 15 倍心跳时间内没有收到客户端消息，则会主动断开客户端的连接，发送其遗嘱消息给所有订阅者。而服务端收到 PINGREQ 报文之后，立即返回 PINGRESP 报文给客户端

  心跳时间单位为秒，占用2个字节，最大 2^16 - 1 = 65535秒（18小时12分钟15秒）,设置为 0 表示不使用心跳机制； 心跳时间一般设置为几分钟或几十秒即可，时间短点可以更快的发出遗嘱消息通知掉线，但是时间短会增加消息频率，影响服务端并发； 微信长连接为 300 秒，而三大运营商貌似也有个连接时间最小的为 5 分钟。

6 qos

  服务质量等级 qos 对应两部分,一是客户端到服务端发送的消息, 一是服务端到客户端订阅的消息; 从发布者到订阅者实际 qos 为两段路中 qos 最小的。

  qos 可选值 0(最多交付一次)、1(最少交付一次)、2(正好交付一次);

   qos = 0 ：接收方不发送响应，发送方不进行重试；发送方只管发一次，不管是否发成功，也不管接收方是否成功接收，适用于不重要的数据传输；

   qos = 1 ：确保消息至少有一次到达接收方，发送方向接收方发送消息，需要等待接收方返回应答消息，如果发送方在一定时间之内没有收到应答，发送方继续下一次消息发送，直到收到应答消息，删除本地消息缓存，不再发送；所以接收方可能收到1-n次消息；适用于需要收到所有消息，客户端可以处理重复消息。

   qos = 2 ：确保消息只一次到达接收方，发送方和接收方之间消息处理流程最复杂；

   Mqtt Qos 深度解读 和 MQTT协议QoS2 准确一次送达的实现

7 payload 负载消息

  字节流类型, 是 MQTT 通信传输的真实数据

8 保留消息

  发布消息时设置, 对应参数 retain, 服务端将保留对应 Topic 最新的一条消息记录; 保留消息的作用是每次客户端连接上线都会收到其 Topic 的最后一条保留消息, 所以可能存在网络不稳定,频繁掉线重连,每次重连重复收到保留消息;

   可以向对应的 Topic 发送一条 空消息,用于清除保留消息。

MQTT 服务搭建 Apache Apollo 服务器 搭建 MQTT 服务

Github 仓库

mqtt 协议

"Mqtt"是一种消息传输协议，用于在物联网中传输数据。当您说打印机无法连接到MQTT服务器时，这意味着打印机无法通过Mqtt协议和服务器进行通信。

可能的原因包括：

1 MQTT服务器地址或端口号设置不正确。

2 MQTT服务器无法访问或已停止服务。

3 打印机没有正确配置Mqtt客户端配置，以便与MQTT服务器进行通信。

4 打印机与网络连接不良或网络故障。

为了解决这个问题，您可以尝试以下步骤：

1 检查MQTT服务器地址和端口是否正确，并确认服务器是否正常运行。

2 检查打印机的Mqtt客户端配置是否正确，包括认证信息、主题等，并尝试重新配置。

3 检查打印机与网络的连接状态并排除网络故障。

如果您需要进一步的帮助，请查阅打印机或MQTT的相关手册或联系相关技术支持。

MQTT（Message Queuing Telemetry Transport，消息队列遥测传输）是IBM开发的一个即时通讯协议。它是一种发布/订阅，极其简单和轻量级的消息传递协议，专为受限设备和低带宽，高延迟或不可靠的网络而设计。它的设计思想是轻巧、开放、简单、规范，易于实现。这些特点使得它对很多场景来说都是很好的选择，特别是对于受限的环境如机器与机器的通信（M2M）以及物联网环境。相对于XMPP，MQTT更加轻量级，并且占用的宽带低。

MQTT协议有以下特点：

那么问题来了？重连连接成功后重复接收到最后一条消息

MQTT推送消息订阅端重复接收问题。

（背景）订阅端断开的时候，发布端多次推送消息。

（现象）订阅端启动时，接收到最后一条推送消息有两次；即使Qos设置为2；依然是两次。

经排查是因为

MqttMessage的Retained设置为了true;

该值很多文章上只说了是 消息保留机制，若设置为true，mqtt服务器会保留每次发布的消息；较少提到 若订阅某主题的客户端重启，则会把此主题之前发布的消息重新推送到客户端。该值默认为false;去掉修改该值即可

那么问题来了？重连连接后手动那么多遗漏的消息，怎么选择只接收最新的一条消息呢？

MQTT推送消息订阅端重复接收问题。

（背景）订阅端断开的时候，发布端多次推送消息。

（现象）订阅端启动时，接收到msg1，msg2，msg3 （这三个消息都是同一个类型消息，只需要处理最新的msg3就好，不然界面会刷新三次）这个谁有什么好办法没呢？

GitHub地址： https://githubcom/eclipse/pahomqttandroid

mqtt的官方文档： http://mqttorg/documentation

Github上有中文翻译： https://githubcom/mcxiaoke/mqtt

在module的buildgradle文件中添加依赖

在 AndroidManifestxml 添加限权

在 AndroidManifestxml 注册Service （MyMqttService为自己写的服务，下文会讲到）