规模设备管理与消息通信

开发者学堂课程【物联网应用开发课程：规模设备管理与消息通信】学习笔记，与课程紧密连接，让用户快速学习知识。

课程地址：https://developer.aliyun.com/learning/course/1055/detail/15307

内容介绍：

一、大规模设备管理概述

二、大规模设备管理原理详情

三、大规模设备管理接入流程

四、大规模设备管理接入实践

五、物联网平台消息通信概述

六、物联网平台消息通信案例演示

七、物联网平台消息通信案例小结

本节课主要学习的是大规模设备管理与消息通信，通过本节课程，大家可以独立完成设备的创建、管理，以及使用物联网平台进行消息通信。

本节课的分为大规模设备管理与消息通信两部分，大规模设备管理主分为以下三个部分内容：概述、原理详情和接入实践；物联网平台消息通信主要包括三部分：物联网平台架构和消息通信原理、案例演示和案例小结。

一、大规模设备管理概述

1、大规模设备管理

以下为大规模设备管理示意图：

通过观察上图可以发现，我们将整个物联网的生态圈划分为五个梯度，依次为客户层、应用开发者、物联网平台、集成商和设备厂商。

简单来说，首先由集成商制定设备协议、规定数据格式，为设备厂商提供一些协议，并将其设备去通过某种协议传输到我们的物联网平台，使得设备上传的数据存放在物联网平台进行统一管理。

物联网平台不仅能接收下游设备的数据，还可以将数据提供给上层，如为应用开发者和客户提供一个统一的API以供他们进行设备的管理。

2、设备圈选

如今物联网的发展越来越快，设备的数量与日俱增，对这些数量庞大的设备进行高效、统一、优良的管理的是非常必要的一个问题，接下来介绍一下物联网平台如何有效、高效的管理设备。在物联网平台，将与之有关的功能定义为设备圈选。

设备圈选又分为基础设备管理和检索基础能力。

（1）基础设备管理

这是一种较为简单的管理功能，它可以给设备进行标签分组，即给设备打上标签，通过标签筛选一些目的设备，还有静态分组，这其实与标签比较相像，但它相当于是一个非常独立的功能，因为它有单独的页面，可以根据分组展示当前分组下所有的设备，比标签更加灵活的。

（2）检索基础能力

该功能其实比较复杂，但也是最为灵活的一种圈选方式，它最为典型的就是可以通过运行时检索。上面所说的基础设备管理其实是基于元信息的管理，而所谓元信息，就是指从创建设备创建开始便基本不发生改变的一种信息，而运行时检索的信息则不同，该种状态下，设备通过不停地向云端传输最新的数据，如属性、事件服务等这些更新非常频繁的的数据，若客户的数据是这种频繁更新的数据，即运行时数据检索，其实是一个非常大的挑战，而这一功能阿里云物联网平台可以实现运行时检索。

通过运行时检索延伸出了新的功能——动态分组，动态分组与静态分组都具有分组能力，但后者是基于非常静态的规则，其是通过产品、标签等进行匹配，而动态分组，它是则可以基于运行时的属性去筛选设备，该分组中的设备是不是固定的，可以通过云端的规则动态匹配设备。

二、大规模设备管理原理详情

该部分介绍各类设备圈选的原理：

1、标签

可以为不同产品设备、设备或设备分组贴上不同标签，也就是说标签的圈选不仅限于设备维度，还可以在产品和设备分组进行圈选。产品标签、分组标签的统一结构都是通过键值对Key-Value来贴标签的。在阿里云物联网平台，标签的限制有产品、设备和设备分组三个维度，每种维度下的子分组（如单一分组、单一设备和单一产品）限制到100个。这一点也是考虑到总体性能瓶颈决定的。

2、分组

包括静态分组和动态分组两种不同类型的分组。

静态分组相当于是依靠用户手动添加设备到分组，也支持将分组中的设备移出，重点在于手动，平台不会自动将某些设备添加到某一分组中。而动态分组与静态分组最大的区别即在于“手动”和“自动”的区别，前者在创建之初就会给动态分组配置分组条件规则，然后基于配置好的条件规则动态匹配符合条件的设备的分组，即不需要用户或者开发者手动地将设备添加或者移出。

两种不同的分组都有各自的限制。当前阿里云物联网平台账号下，静态分组最多只能创建1000个分组，且这1000个分组包含所有分组及所有分组下的子分组，且分组层级也有限制的，一共三层（分组、子分组、子子分组），每层下的子分组只能有100个。以上即为静态分组的三个限制，如今，静态分组也在不断进行优化，后期对于总的分组数可能会逐步放开到1万、10万个。动态分组的限制目前仍处于观望状态，单个账号只能创建10个，这是考虑到运行时数据的匹配的复杂度决定的。

3、运行时检索

也叫做高级检索，其检索的难度、复杂度较高的，其功能是通过类SQL语句进行快速搜索满足决定条件的设备。例如，匹配圈选一些在线设备。运行时检索的主要功能在管理设备、设备分组（设置动态分组）和OTA升级三个方面有非常广泛的应用。

进行运行时检索的支持类SQL语句，类似与SQL的where语句，只要填写where条件即可。如：

product_key= ＂al*****＂order by active_time，

//利用基础元信息匹配，通过激活时间进行排序，使得product_key等于特定的值。

这条SQL语序不是非常典型，因为它的条件是基于元信息匹配的，但是高级检索最重要的特性使通过运行时的数据进行匹配，就如：设备上报的数据温度是20度，要匹配温度大于15度的条件，则temperature>15即可通过高级检索把此时温度大于15度的所有设备全部圈选出来，再提供下载csv列表的功能，把所有匹配的设备通过csv的文件形式导出，方便开发者和客户使用。

三、大规模设备管理接入流程

1、简单设备圈选使用流程

（1）标签

需要通过填写相关信息编辑标签，在创建标签时即会弹出如下窗口：

①地理位置标签

是一个预定义的标签，如果没有给设备、分组或产品打上位置标签的特殊需求，可不填，也可以选填。

②设备标签

在第一个输入框中填写标签的key，第二个输入框中填写标签的value。这里以为设备打标签为例，填写完标签之后，就会在设备页面上显示当前的设备的key（location）和value（bedroom）：

（2）分组

以下为创建静态分组时弹出的窗口：

在该窗口中，首先需要注意的是分组类型，选择默认即可；父组是指在创建分组时，可以指定分组是否要隶属于其他的分组，如果不是要创建子分组，可以不填，反之，则可以通过该项对分组进行层级管理，在创建新分组时指定父组隶属于分组下；分组名称即对分组进行命名；分组描述是为了后期的开发，即描述该分组的功用。以上最重要的是选择默认分组类型（静态分组），并填写分组名称即可。

以下为创建好的分组的例子：

其显示在分组的页面下。

2、复杂设备圈选使用流程

该过程为高级检索，即运行时检索，目前阶段其功能仅体现在企业实例中，因此，要体验高级检索，需要注意是否为企业实例。

首先，高级检索支持的检索大体上分为四类：一、通过基本信息圈选，即元信息，即图上显示的pk、产品、设备名称、设备激活时间，这些都是固定的值；二、通过运行时数据圈选，这是高级检索主打的功能，可以通过设备下的物模型所有的属性、事件服务进行检索；三、通过分组进行圈选；四、通过标签进行圈选。如刚刚所说，可以通过sql语句把分组和标签融合在一起进行检索。

其体现在设备页面下一个叫作高级检索的tab，以下为主要信息示意图：

高级检索其实是把所有的基础检索再加上运行时检索融合在一起的么功能。

四、大规模设备管理接入实践

1、运行时检索实例

（1）创建新产品

进入物联网平台，选取企业实例，进入企业实例iot-060a0bah。

为了进行测试，可以创建新产品，将产品名称命名为XLY_TEXT，并选定所属品类（图中以“空调温控器”为例），其他选项选择默认即可。

（2）创建并激活设备

基于产品创建设备，定义设备名称，即可创建成功。

在设备创建完成后，该设备显示“未激活”，在“监控运维”中设备模拟器功能，通过该功能可以模拟设备的上线计划。

选取刚创建的产品和设备，进行调试。点击启动模拟器，模拟把设备激活。

以下为进行属性上报的测试的案例：

当前页面显示设备下的物模型模块，模块下有各种属性，可以选择当前温度进行上报，模拟上报时写24度，即可模拟发送成功。

进入设备，核实设备是否接收到属性上报信息。点击“物模型数据”，该页面显了刚输入的模拟上报：

显示当前的温度是27度，表示模拟上报成功。接下来，尝试一下高级搜索功能。

（3）高级检索功能演示

进入到“设备”页面下的“高级搜索”选项，下面是一些搜索案例：

这里以“通过物模型数据搜索“为例：

首先需要配置物模型索引属性，点击“配置索引”可以发现，已经配置过了当前温度属性，无需再额外配置，点击确定即可。若没有配置过，则应先把属性配置到索引中才会去被高级搜索检索到，否则，无法检索到该属性。

若要通过当前的设备的温度进行匹配，可以直接点击属性标识符property.float TEXT>11或者进行复制粘贴到搜索框中,并对其中参数进行适当修改。property表示要通过属性进行搜索，float TEXT是一个需要填写的字段，因此需要找到“float TEXT”字段，即current temperature属性，将属性标识符中的“float TEXT”改为“current temperature”，即基于检索的目标字段。

此处模拟检索匹配大于室温（25℃）的设备：

由于刚在上报的温度是27℃，因此该设备会被检索到。若条件保持不变，重新上报为24℃，再次进行搜索，该设备不会被搜索到。总之，高级检索会时时基于物模型假设的数据进行上报。

2、动态分组实例

动态分组是基于高级检索所衍生的功能，它的创建逻辑与普通的静态分组相同。在创建分组时，分组类型选择“动态”，动态规则与上面演示的高级检索的检索条件相同，如动态规则property.current temperature>20，即分组会去自动将当前实例下包含current temperature字段且温度大于20℃的设备添加到该分组中，可以将分组名称命名为“高于室温”，最终进程符合该条件的动态分组。

关于动态规则，其是它不单指运行时检索运行的动态规则，也可以根据其他的方式，如分组、标签、OTA信息匹配等的动态规则。

创建分组，选择“动态”，动态规则为“property.current temper

ature>25”，分组名称命名为“greaterThanRoomTemp”，点击“确认”创建成功。当前设备列表没有任何设备，因为刚上报过的温度是24℃，重新上报温度“26℃”，刷新后，该设备就会动态地去匹配到动态分组中。

3、代码Demo-门禁定制OTA升级场景

该场景为基于以上所有的设备圈选功能的场景，大家之后可以根据实际需求仿照代码Demo编写。

我们生活中有很多智能设备需要经常进行OTA升级，例如运用CV技术（计算机视觉技术）的人脸识别门禁设备，因为在AI方面，CV技术的人脸识别模型是高速迭代的，需要设备（如人脸识别门禁）具备可经常更新的能力，但终端设备（门禁）往往由于技术硬件有限，导致限制条件较为复杂，因为设备硬件条件不一，可以通过设备圈选能力把符合条件的门禁圈选出来，再把OTA升级包推送给符合升级条件的设备。

这里的代码Demo会去模拟这个场景，为特定状况的设备进行过OTA推送，以下为实现过程：

（1）提供自动创建设备、自动上线的脚本

①初始化

在该自动化的脚本中，最开始把textswitch（开关）改为INIT

private static final Testswitch testswitch=TestSwitch.INIT;

②创建产品、设备、分组、动态分组

运行主程序，结果显示了许多设备，以及产品创建成功。

可以去在控制台上去确认一下，点击返回控制台的“产品”“设备”“分组”中可以查看到刚刚创建的设备、产品和动态分组

③自动上线激活

即把刚刚创建的设备去进行自动激活上线，先进入Mockbridge模块比赛，模拟网桥实现云云对接，一个网桥下挂九个设备来进行模拟上线，脚本如下：

public static void main(string[] args) {

inito

activate();

pushToDevice();

}

（2）提供数据自动上报脚本

运行脚本即可以自动激活上线，可以再去控制台确认一下，发现刚刚创建的设备由未激活状态变为了在线状态。任意点击一个设备，可以查看其物模型数据，由于刚才的脚本不止是激活和上线，而是一个随机脚本，将附加的属性的各个字段进行了随机上报。

把各个门禁中各个字段进行随机的上报，刚才显示的数据都是成功上报的数据。

在刚刚创建的动态分组中，其动态规则是预先定义好的，门禁设备的动态规则包括两个，即人脸识别级的大小小于10万、人脸识别级的存储能力大于17万，由此圈选符合两个条件的门禁设备。因为，圈选的目的是筛选具有更多剩余容量的门禁设备，避免导致设备的终端的存储能力不够的状况出现。

确认、手续申报完毕后，查看动态分组中是否有满足这些条件的设备存在

结果显示有三个设备，再次确认是否符合动态规则的两点识别级，点击设备，查看物模型数据，发现圈选的动态分组中的单个设备均符合，接下来对这些设备进行OTA升级。

（3）OTA升级

添加OTA升级包并进行配置：

对升级包不进行平台验证，因为仅只是一个演示。创建好升级包之后，可以基于刚才的动态分组，创建动态升级的任务。

（4）OTA升级包任务创建

选择刚创建好的升级包进行批量升级：

升级方式选择动态升级，即通过动态分组的方式圈选设备，升级范围选择分组升级，分组列表选择刚刚创建的动态分组，点击下一步，“恒定速率升级”是每分钟推送的设备数设置为“10”，其他保持默认，点击“完成”，设备状态从“被升级”变为“升级中”，设备列表中刚圈选在动态分组中的三个设备的状态显示“已推送”。

（5）清除数据脚本

如果以上过程中遇到问题，可以初始化把创建的产品、设备都删除

初始化之后返回控制台可以发现创建的动态分组中的设备、产品都已经删除了。

此外，附录中还包括标签分组、高级检索等的开放接口的文档，以及演示的整个代码Demo，可以多进行实操。

五、物联网平台消息通信概述

1、物联网平台架构

在设备侧可以通过物联网平台提供的SDK实现设备的接入，在物联网平台提供了消息通信、设备管理、监控运维、数据分析等服务，同时也提供了安全认证、全线策略的服务，以确保设备和云端数据的安全。物联网平台同阿里云产品也可以进行数据的流转，将设备上报的数据流转到阿里云产品进行存储，或者进一步的实时处理。

2、消息通信介绍

（1）通信过程

设备可以通过SDK进入到物联网平台，进入时支持多种的协议选择，如MQTT、CoAP、HTTPS。

设备接入之后，上报消息到物联网平台，首先，消息会到达消息中心做进一步的存储和分发处理，消息中心做了计算和存储分离的架构，在架构层面可以支持水平扩容以应对海量的设备消息。消息中心一方面可以将消息分发到规则引擎，进一步对消息数据进行计算、过滤等处理之后流转到原产品；另一方面，也可以通过消息队列的形式推送到消息的消费端。

消息队列具有实时优先、推拉结合的特点。所谓的实时优先，设备上报的消息会实时的通过队列网关推送到应用的消费端，如果消费端有不在线或者推送失败的情况，队列则会把消息堆积起来。队列网关会不断拉取堆积的消息，进一步重新投递。最后，在应用层提供了API，能够下行控制设备。

（2）通信方式

整体上，物联网平台提供以下消息通信方式：

①设备商报数据

设备可以通过MQTT等协议与物联网平台建联，基于topic的形式上报数据，上报的数据可以是用户自定义的，也可以是通过模型定义的一些数据。

②规则引擎流转到云产品

在规则引擎中，可以通过SQL或者脚本的形式对数据做进一步的处理和过滤，流转的云产品也相对来说比较丰富。如消息队列MQ、DataHub、函数计算FC或直接存储的数据库，如RDS、表格存储TSDB等。

③服务端订阅

服务端订阅即队列，用户的应用服务器可以通过开源的AMQP协议接入，接入之后，设备上报的消息优先会实时推送到消费端，如果设备（消费端）不在线，或实时推送失败，将进入堆积队列，堆积的消息不会影响实时的消息推送，即实时优先的策略。

④应用服务器下行控制设备

通过API的形式来远程控制设备，如应用服务器可以调用Pub API向设备发送消息，或调用Rrpc API向设备发送消息并且同步获取设备响应结果，或者说是直接调用Broadcast的API向设备广播消息。

以上是物联网平台比较常见的四种消息通信的方式。

六、物联网平台消息通信案例演示

1、案例分析

（1）业务背景

小A在某个物业大厦物业技术部工作，某一天他接到了大厦的环境智能化升级的任务。因此，他决定将大厦类的环境监测、能耗统计等设备都接入物联网平台，并且通过物联网平台将数据流转到自己设计的智能化应用系统当中，对监测的数据进行分析计算。在计算后，可以自动的去控制大厦中的一些相关设备，从而提升大厦整体的环境质量，降低能源消耗。

要实现业务，可以抽象出来几个跟物联网平台相关的功能列表。

（2）功能列表

①设备上报监测数据到物联网平台

②将数据流转到表格存储供后续的统计和分析

③若监测的环境指标超过了预警，则可以将消息流转到消息队列去做预警的通知，如发送钉钉通知到相关的人员

④可以通过AMPQ队列将设备的监测数据推送到小a自己实现的智能化应用系统，当中做实时的业务处理

⑤应用系统可以控制空调或者通风设备，做一些指定的操作

（3）整体设计

包括七个步骤：

①监测设备

环境数据的监测设备将数据上报到物联网平台，从物联网平台将数据流转到云产品用作存储，以供后续的统计分析以及预警通知，设备上报的监测数据可以通过AMQP服务端订阅的形式实时推送到智能化应用系统当中，做一些实时的业务处理判断。如温度超标，可以调用下行控制的app直接去控制空调或者通风设备，开启通风或设定空调的温度。

（4）相关定义

①Topic定义

整个通信是基于Topic的，Topic是消息的发布和订阅之间的传输中介，设备可以通过Topic实现消息的发送和接收，从而实现服务端和设备端的通信。

在该案例中，我们定义了两个Topic：

第一个Topic用在设备监测数据的上报

Topic:/${productKey}/${deviceName}/user/data/post

第二个Topic用在服务端下行去控制设备。

Topic:/${productKey}/${deviceName}/user/invoke

②数据格式定义

针对设备监测信息的上报，数据可以定义为以下JSON对象：

/${productKey}/${deviceName}/user/data/post

{

"location:"A 03F_001" //设备所在位置，如A幢3楼001

"metricType":"temperature //监测设备监测的指标：温度指标

"timestamp:1656484971847873. //当前时间，即监测数据上报的时间

"metricValue:25 //监测值，温度值25摄氏度

}

对于服务器下行控制设备的Topic数据格式，相对来说比较简单，仅定义开关打开的格式

/${productKey}//${deviceName}/user/invoke

"switch:"on //设备开关打开

（5）操作演示

①创建产品设备及 topic

要完成系统的接入，首先需要创建产品和设备。进入到物联网平台看是否页面上有相关的实例，如果没有，则需要购买实例。在有实例的情况下，进入实例创建产品，如之前创建好的楼宇环境智能化的产品，在产品下创建了两个设备，温度转改器和空调设备，因此，可以模拟温度传感器检测到数据上报到物联网平台，空调接受服务端的控制打开空调或调整温度。

设备和产品创建好之后，再进一步的做消息流转的一些配置。

②云产品流转

所谓的云产品流转即将设备上报的数据流转到阿里云的其他的云产品，如表格存储或消息服务MQ。进入到规则引擎栏目，在云产品流转里面，先创建一下那解析器，即用来处理消息的实体信息。

创建好解析器之后前往编辑。

首先，第一步是关联数据源，那所谓的数据源即解析器要处理的数据。若没有关联数据源，则可以新建数据源，有了数据之后，可以把它关联过来。TEST数据源具体要去处理哪些数据呢？在整个物联网平台，消息通信都是基于Topic的，点击添加Topic，且整个数据的通信都是基于自定义的Topic，因此选择自定义的Topic，选择产品楼宇环境智能化产品，设备可以选所有的设备，所有设备上报的消息都会由解析器来处理。Topic余下的部分，因为这里主要是处理设备上报的Topic，因此选择设备上报的Topic ：user/data/post，点击确定。如果设备上报了Topic ：user/data/post数据，解析器就会处理数据。

第二步，关联数据目的，所谓的数据目的是指我们要把消息流转到哪个地方，可以点击关联数据目的，若没有数据目的，可以创建数据目的。要流转到table store中，就创建table store数据目的，“地域”取华东2，“实例”选择之前创建过的，“数据表”选择之前创建的表device_metric，“角色”需要访问物联网平台OTS，获取角色授权。填写好之后，点击确定，就创建好了数据目的将其把它关联进来。创建流转到MQ的视频，“地域”按提示选择填写，利用之前创建的实例，topic也选用之前创建好的post topic，如果没有，可以点击创建Topic进行创建，同样也需要做授权，物联网平台才能访问MQ，将数据流转过去之后，点击确定，并将之关联过来。

第三步，脚本编辑。将之前写好的脚本拷贝过来。在脚本编辑时，要注意以下几点：首先可以简单的看一下脚本的意思，脚本的语法与JS的语法比较类似：

//获取消息payload数据

var data=payloadC"json");

//payload是系统内置的函数，它的含义是将消息上报，取出设备上报topic里的payload数据取出来，而上报的是JSON格式的数据

if(data.metricType"temperature’&&datametricValue>28) {

//如果温度大于28度时，流转当前消息到MQ，并发起预警

//数据目的Id1005需与当前规则配置关联数据id保持一致

writeMq(1005.data);

}

//所有上报消息都流转到TableStore，用于统计分析

//数据目的Id1005需要与当前规则配置关联的数据id保持一致

writeTableStore(1004{"deviceName":deviceName(),"type":

data.metricType,"value":data.metricValue，"timestamp": data.timestamp});

//把数据流转到ots的存储中，前面“1004”也是数据目的id，后面是JSON结构的内容，如deviceName，deviceName()是deviceName的函数，取当前设备上报的消息的设备名，有type、value字段，增加timestamp字段把数据流转到TableStore中，保存。如果是在平时使用的时候，还可以利用调试功能做一下调试，如调试脚本是否有语法错误，计算逻辑是否正确？无误后，把它发布，脚本发布成功之后，可以看到解析器，他还是不会启动状态，将其启动，当启动解析器之后，如果此时设备上报一条数据，解析器就能够进行监听和处理了。

最后要再看一下服务端订阅的配置，默认已经有消费组，无需再新建，把消费组创建一下订阅，订阅刚刚的产品，所要监听的消息即为设备上报的消息，点击确认。相对来说，消费组的配置是非常的简单的，只需要配置一下订阅关系即可。

整个消息使用在控制台的配置已经完成了。

③设备和服务端订阅

设备上面运行的代码，如下：

这里有两个Python脚本，使用Python脚本来模拟设备端，大家可以根据自己的具体情况参考官方文档使用对应的语言stk来处理。

以下为温度传感器的脚本，具体分析Python脚本中的代码：

配置设备信息，如设备的device_name、密钥，最后要做认证。

定义了很多回调函数，如建联以后、断联以后、收到消息订阅、取消订阅、发送消息等等。

初始化sdk，最后调用最后一行，发起建联。

模拟设备定时上报温度，温度从25℃到35℃随机取值。上报格式前面提到过。有了内容之后，调用方法publish_topic把数据上报，通过topic上报到物联网平台，这里做了定时器，每十秒钟上报数据。

以下为空调的脚本，具体分析Python脚本中的代码：

初始化的配置相同，即一些设备信息，还有一些监听的回调函数，对于空调设备，需要处理服务端控制下行的topic，因此设备需要订阅topic，否则，服务端下行控制的指令发送后，并不会推送到设备上。因此在建联成功后需要做topic的订阅函数。

建联发起连接设备

对于空调来说，在案例中，只是订阅了topic做对应的响应，在收到消息以后，调用topic_message的回调函数，显示当前设备收到了topic的消息，打印一行“>”，不做额外的处理。

如果结果中有一行“>”，说明设备收到了下行控制的消息了。在实际使用当中，可以自己定义收到控制的消息。

以上是两个设备端的代码，最后再看一下服务端订阅的代码。

服务端订阅通过MQP协议接入，简单分析一下：

拼接认证参数，具体的些参数格式、如何去拼接签名可以参考对应的文档进行操作。

通过sdk建立连接，即消费端同MPP网关建立连接，这里的messageListener，即收到推送消息之后的处理逻辑。

处理逻辑中直接把它提交到了另外线程池中处理，可以看一下process_message函数，该函数种进行了比较简单的处理，如果我们上报的数据是温度数据，并且当前的温度已经大于30℃，开启空调设备。那如何去开启？服务端控制设备，那我们可以发送topic到物联网平台，物联网平台会把topic推送到订阅过topic的设备中。