利用 EMQX 消息队列解决关键物联网消息传递挑战

来源：EMQ中文社区
链接：https://mp.weixin.qq.com/s/H3mOxNYhfYyEJtN1chkDgw

您成功构建了一套出色的物联网系统：设备数据传输稳定流畅，实时仪表盘运行良好。

然而，一旦涉及异步操作，问题便层出不穷：发送给离线智能锁的关键固件更新失败；数据处理节点负载失衡，部分节点处于闲置状态，其他节点已不堪重负。

MQTT 的实时性优势在面对异步操作时显得捉襟见肘，这正是许多物联网项目变得复杂的根本原因。

为了弥补这一差距，架构师不得不引入独立、功能强大的消息队列，如 RabbitMQ 或 Kafka。这种方案不仅增加了部署和维护的负担，还迫使后端团队不得不学习 AMQP 等不同协议，并适配全新的客户端库。

如果无需引入任何额外系统，就能解决这些问题呢？如果您的开发人员仅凭熟悉的 MQTT 协议和现有客户端库，就能同时实现实时发布/订阅与可靠的消息队列功能呢？

基于 EMQX 消息队列功能，这种统一、高效的消息处理模式已成为现实。

现在，让我们通过物联网领域中三个常见的难题，来看看这个内置功能是如何解决这些问题的。

三大常见难题，一个集成方案

规则引擎是 EMQX 内置的基于 SQL 的数据处理组件，专为物联网场景设计，能够高效、低成本地实现实时数据流转与处理。如今，大多数物联网系统看起来像这样：

EMQX 消息队列填补了实时 MQTT 发布/订阅与持久化消息队列的功能空白。

以下是一些针对特定挑战的解决方案：

实战应用：构建高可用的工作流程

场景描述

假设有一个基于 AI 的图像处理进程，物联网摄像头将图像发布到主题 images/process，一组工作节点应用程序需要订阅该主题，获取图像并执行 OCR（光学字符识别）任务。   面临的问题

采用标准的 MQTT 发布/订阅模型，您将面临两个问题：

• 如果没有订阅者在线，消息就会被直接丢弃。
• 即使您使用共享订阅来分配负载，消息依然缺乏持久性：工作节点在接收消息后、处理完成前发生崩溃，该任务将永久丢失。

解决方案：

我们可以将主题映射至一个持久化队列：

• 消息发布：生产者（摄像头）照常向主题 images/process 发布消息。
• 持久化存储：EMQX 接收该消息，并根据为 images/+ 主题过滤器配置的队列规则，将消息持久化存储至磁盘。
• 队列订阅：工作节点应用程序通过主题 $q/images/+ 订阅共享队列。
• 智能调度：通过将分发策略设置为 least_inflight，EMQX 会将下一个任务智能分配给待处理任务最少的节点，有效避免瓶颈产生。

最终结果：

我们获得了一个在 EMQX 内部完成的高弹性、持久可靠的工作流：任务永无丢失，负载完美均衡，同时彻底免除了外部消息代理、部署和管理的费用。

实战应用：使用最后值队列进行状态同步

场景描述

假设您有一个用于控制智能灯的云端应用程序。用户想要更改灯光的颜色，但由于 Wi-Fi 连接不稳定，智能灯暂时处于离线状态。在它离线期间，用户多次改变主意，先后发送了蓝色、红色和绿色的命令。

面临的问题

如果使用标准队列，智能灯在重新连接时将收到积压的三条命令。它将依次变为蓝色、红色和绿色，造成令人困惑的效果。而我们只希望它实现最终想要的状态：绿色。

解决方案：

这是使用最后值队列（Last Value Queue）的完美用例。

您可以将主题 lights/light-123/commands 配置为一个最后值队列，并指定消息中的一个键。例如，使用名为 command_type、值为 set-color 的用户属性。

现在，当命令发布时，队列将只存储针对 set-color 键的最新一条消息。「蓝色」和「红色」的消息会在「绿色」消息到达时自动被丢弃。

最终结果：

当智能灯重新连接并订阅时，它只会收到一条消息：最终的「绿色」命令。设备会立即反映出最终预期的状态，而不是过时的历史记录。

最后值队列与 MQTT 保留消息的区别

这是一个常见的混淆点。虽然两者看起来相似，但它们的用途截然不同。

我们可以用一个比喻来理解它们的差异：

MQTT 保留消息就像贴在主题入口的一张通知单。它只为整个主题保留唯一一条最新的消息，任何新订阅者连接后都会收到这张通知单的副本。

EMQX 最后值队列就像是为消费者专门设计的智能邮箱。它能为多个不同的键（例如：设置颜色和更新固件）保留各自的最新指令，并且能够跟踪消费者已读取的位置（偏移量）。它是一个真正具备状态感知能力的通信通道。

以下是二者的详细对比：

统一的平台，简化的架构

通过将持久化队列功能直接集成到 MQTT broker 中，EMQX 有效消除了部署独立消息系统所带来的架构与运维复杂性。

更重要的是，这一实现完全基于 MQTT 协议。您的设备和后端应用团队可以使用他们早已熟悉的同一套 MQTT 客户端与库，同时处理实时主题与持久化队列。无需引入新协议，没有额外复杂度，只有一个统一而强大的消息平台。

在进行功能实践时，请牢记以下要点：

• 选择合适的调度策略：对于简单的任务分发，可使用轮询模式；但在处理时间不确定的工作节点场景中，更推荐采用最少未确认消息策略。
• 密切关注队列状态：实时监控队列深度与消费者延迟，以便及时扩展应用容量。