本文整理自 EMQX 产品经理李国伟，在PolarDB开源社区中关于EMQX与PolarDB-X构建一站式IoT数据解决方案的分享。本篇内容主要分为四个部分：

1. IoT数据特性
2. EMQX介绍
3. EMQX与PolarDB-X集成
4. EMQX+PolarDB-X方案DEMO

一、IoT数据特性

物联网应用场景离不开数据的采集、传输、存储、分析等过程。大家可以按照数据特性以及业务需求，把物联网数据划分为元数据、消息数据、时序数据。如上图所示，数据集通过EMQX连接设备，实现数据存储。

元数据是设备最新的状态数据，如在线状态、当前传感器数值。消息数据是设备发布的消息，包括上报数据和下发指令。时序数据是持续变化的元数据和消息数据。

在物联网应用中，数据存储的需求无处不在。元数据、消息数据经过存储使用后，可以支撑起各类业务需求。由于连接数规模、采集点数量庞大，不同的数据业务，需要对应不同的数据库选型。

元数据需要频繁插入更新，并且支持结构化查询，所以推荐使用关系数据库，进行存储。
消息数据不需要全量存储，只需要记录关键操作，或者提取消息数据的关键数据，应当根据业务情况适当的选型。

时序数据的特点是，数据修改频次低，海量数据对写入速度和存储压缩比敏感，查询需求多样，所以推荐使用时序数据库。

二、EMQX介绍

MQTT是一个轻量的发布订阅模式消息传输协议，专门针对低带宽和不稳定网络环境的物联网应用设计。

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

MQTT协议广泛应用于物联网、移动互联网、智能硬件、车联网、电力、能源等领域，既能作为网关在设备侧接入通信，也能作为设备-云端的通信协议：行业内ZigBee、LoRa、Modbus等绝大多数网关协议最终都转换为MQTT协议接入上云。

EMQX是MQTT代理，即MQTT消息服务器。它将设备和应用联系起来，充当设备与设备、设备与应用之间的桥梁。

EMQX是一个大规模分布式的物联网专用的MQTT消息服务器。它可以高效可靠的连接海量物联网设备，提供高可靠、高性能的实时数据移动、处理和集成，助力构建关键业务的物联网平台和应用。

EMQX基于Apache 2.0许可证完全开源，自2013年起200+开源版本迭代，在全球都有广泛的用户。

EMQX在协议的支持度、接入、吞吐、延迟等方面有显著的优势。在最新发布的5.0版本中，EMQX单节点支持500万MQTT设备连接，集群可扩展至1亿并发MQTT连接，可以支撑超大规模的物联网应用。

EMQX可以连接任何设备。通过开放标准物联网协议MQTT、QUIC、LwM2M/CoAP，支持连接所有车联网、工业、能源电力等关键业务场景的异构终端设备。

EMQX提供了安全的双向通信。通过TLS/SSL、QUIC和多样化的认证机制确保安全的双向MQTT安全通信。支持用户名/密码、JWT、PSK和X.509证书认证，确保数据安全。

除此之外，EMQX实时数据处理能力，通过一个强大的基于SQL的规则引擎，以数百万条/秒的速度实时过滤、转换与处理设备与云端之间双向移动的MQTT消息数据。

在运维管理以及可观测性方面，EMQX通过CLI、HTTP API和Dashboard，轻松管理EMQX集群。支持使用Datadog、Statsd、Prometheus和Granfana进行监控和报警。

三、EMQX与PolarDB-X集成

接下来，看一下EMQX如何与PolarDB-X集成，实现物联网数据PolarDB-X存储方案。这个方案广泛用于EMQX开源版本。由于数据存储业务众多，连接数规模与采集点数量庞大，所以需要超高并发写入和海量存储支持。

通过EMQX的数据集成能力，以及MQTT共享订阅功能，将内部的设备、事件、消息，发送到用户编写的程序中，实现PolarDB-X的插入或更新。

数据集成是EMQX在发布订阅模型的基础之上的数据处理与分发组件，通过简单的可视化的配置，即可将消息流以及设备事件与Kafka、RabbitMQ等消息中间件，以及各类SQL/NoSQL/时序数据库等数据系统集成。

通过上述规则EMQX可以处理各类数据，包括MQTT消息和设备生命中心事件。其中，MQTT消息包括设备端的上报，或者云端的下发。

设备生命周期事件是指，整个设备在运行过程中的事件。这些事件对物联网的应用建设有很大帮助。比如进行设备管理，安全审计啊，消息确认等等。消息丢弃、消息投递、消息确认是精细化的MQTT消息传输过程。

共享订阅是在多个订阅者之间实现负载均衡的订阅方式，相当于订阅端的负载均衡功能。当发布者的生产能力较强时，可能会出现订阅者的消费能力无法及时跟上的情况。

如果在EMQX集群中，某个节点挂了，同时订阅多个节点避免某个节点故障导致数据丢失。

用户可以通过使用$share/{group}/{topic}或$queue/{topic}格式的主题，发起共享订阅。

如上图所示，多个客户端发布的消息。经过EMQX之后，共享订阅会在多个订阅者之间均衡派发消息。

四、EMQX+PolarDB-X方案DEMO

在本次演示中，使用PolarDB-X存储MQTT设备在线状态，在上下线时更新并记录更新时间。

使用PolarDB-X记录设备事件，记录设备上下线历史、订阅、取消订阅历史记录等行为轨迹。

使用消息存储功能，将存储设备上报的消息，来源设备、主题、消息内容进行存储，实现消息入库。

上图是worker的实例代码，负责接收MQTT消息，进行数据插入。在处理时，把它转换成JSON，根据JSON event判断插入历史表，或者进行其他操作。

上图是设备在线状态表，包括一个ID、设备IP、在线状态，创建时间、更新时间。

历史事件和消息表都有对应字段，对应的数据类型。消息表的Payload可以存储较大的消息文本。

EMQX规则将所需事件数据通过republish动作，转发至worker处理。

接下来，进入实际的操作演示。首先，选择本地部署PolarDB-X，按照文档中的快速入门指南，在本地部署。

然后，创建表结构。上图展示了具体的SQL文件，方便大家进行创建。

接下来，插入数据并更新语句。更新ip_address和状态。当唯一键重复时，将会执行更新操作。如果不重复，会写入数据。

client_events表包括客户端的事件。当设备断开连接，没有topic。当会话订阅时，有topic。

接下来，查看表里面都有哪些数据。如上图所示，表里包括了id、clientid、event、topic、created_at、updated_at。

接下来，插入messages表。将clientid，topic，payload插入。

接下来，在本地启动EMQX。启动成功之后，访问localhost18083，把语言改成简体中文。

在数据集成功能中，主要包括数据桥接以及规则，用规则获取数据。

EMQX的事件通过主题进行获取。在客户端建立上线事件连接，用调试的方式，执行SQL。EMQX可以在客户端里，拿到相关信息，包括ID、关键事件、连接时间、连接属性等等

然后，断开连接。断开连接对应的是disconnected事件，跟连接的结构很像。但在event里，是断开连接事件。所以，用户可以通过event字段，区分是否连接。

在设备的事件历史记录方面，获取订阅跟取消订阅事件。

session_subscribed是订阅事件。当会话订阅执行之后，可以获取当前的上下文信息、订阅的topic等等。

session_unsubscribed是取消订阅事件，跟订阅大同小异。event_topic变成了取消订阅的主题。

在消息存储方面，用户需要从客户端发出的主题中，获取消息。比如从t-1主题发布消息时，需要从t-1主题获取消息。其中，最主要的是Payload信息。其他数据，可以根据需要进行存储。

当用户获取相关数据后，可以使用数据桥接进行转发，通过外置数据库或者MQTT进行数据发送。除此之外，用户也可以使用控制台输出，把结果打印到控制台。

接下来，进行测试。在问题分析中，找到WebSocket客户端建立连接。已发送是t-1，已接收是t-2。消息经过republic，进行转发。

在控制台里，除了消息发布，还有控制台打印。把动作和上下文信息打印出来，方便用户进行调试。

在EMQX开源版里，由于没有直接编写数据库的能力。所以用户需要使用Worker建立一个mysql链接，连接到PolarDB，使用PolarDB的兼容模式。

然后，创建订阅者。把客户端连接到EMQX上。在主程序里，会创建十个客户端。每个客户端都共享订阅主题。当收到消息时，注册一个handleMessage程序就。它会把data转成对象。

转换之后，就可以获取相应的字段。当event已连接或断开连接时，需要写入或更新设备表。把数据插入到历史表里。

如果是消息发布事件，把消息插入message表里。如果是其他事件，把这个事件插入到客户端事件表里。每来一条消息，会轮流派发到客户端。不会出现一条消息同时给十个客户端处理。

PolarDB-X超高并发、海量存储的特性可以应对物联网大规模设备接入所需的各类数据存储场景。

其分布式特性以及存储计算分离架构带来的水平扩展、分布式事务、混合负载等能力，可以与同样是分布式的EMQX结合使用，打造真正的可伸缩物联网应用，应对从数千到数千万的设备接入。

除了本次分享的开源版方案，EMQX Cloud和EMQX企业版中还提供了MySQL数据集成能力，可以直接通过PolarDB-X兼容语法完成数据集成，更简单高效的实现物联网数据集成。

原文链接

本文为阿里云原创内容，未经允许不得转载。