企业物联网平台开发(工业物联网平台开发)

 

简介：IoT企业物联网平台开发实战

企业从联网平台开发实战

通过阅读本文你将学会一下技能：

设备通过MQTT协议与您在阿里云上购买的IoT企业实例建立双向连接，上报设备采集的数据，监听云端下达的指令；通过规则引擎配置把上报的数据实时存储到指定数据库，无需编写代码通过规则引擎配置把上报的数据实时流转到业务服务器，需要使用AMQP协议SDK业务服务器调用IoT平台的API，下达控制指令到设备端

创建企业实例

首先，我们登录物联网平台控制台(https://iot.console.aliyun.com), 点击购买实例来创建一个企业实例。

然后，在购买页面，根据实际业务需求，选择地域、实例类型、设备数量、消息上下行TPS、规则引擎TPS等参数，点击立即购买，付费成功后，即可看到企业实例创建中。

稍等几分钟后，企业实例创建完成。进入企业实例，我们可以看到当前规格参数，设备接入点信息，AMQP订阅接入点信息，云端API调用接入点信息。如下图：

创建产品和注册设备

在企业实例的设备管理页面，我们需要先创建一个产品家庭温控器，数据通信以JSON格式，认证方式为设备秘钥。

在产品的功能定义页面，我们添加温度和湿度两个属性，具体细节如下图：

最后，我们在设备管理页面，基于家庭温控器产品，注册一个物理设备，并获取设备身份认证的三元组。如下图：

设备接入和上报数据

获取设备身份三元组后，即可通过MQTT协议接入到我们开通的企业实例。设备端应用程序逻辑如下图：

完整的Nodejs示例代码如下：

constmqtt =require(aliyun-iot-mqtt);// 1. 设备身份信息varoptions = {productKey:"产品productKey",deviceName:"设备deviceName",deviceSecret:"设备deviceSecret",host:"实例化MQTT接入点"};// 2. 建立MQTT连接constclient = mqtt.getAliyunIotMqttClient(options); client.subscribe(`/${options.productKey}/${options.deviceName}/user/get`) client.on(message,function(topic, message){console.log("topic "+ topic)console.log("message "+ message) }) setInterval(function(){// 3.上报温湿度数据client.publish(`/sys/${options.productKey}/${options.deviceName}/thing/event/property/post`, getPostData(), {qos:0}); },5*1000);functiongetPostData(){constpayloadJson = {id:Date.now(),version:"1.0",params: {temperature:Math.floor((Math.random() *20) +10),humidity:Math.floor((Math.random() *20) +10) },method:"thing.event.property.post"}console.log("payloadJson "+JSON.stringify(payloadJson))returnJSON.stringify(payloadJson); }

启动模拟脚本后，我们看到设备状态为在线，从模型数据中可以看到最新上报的温度和湿度值。

在监控运维的日志服务里，也可以看到设备上报数据的日志。如下图：

数据存储到数据库

首先，我们创建一个表格存储实例 IoTDataStore，建立一张数据表iot_data，以deviceName和timestamp为主键。如下图：

在IoT企业实例，云产品流转中创建规则引擎，编写数据处理SQL，配置流转目的地为上面创建的数据库表。

互联网小常识：综合布线系统设计等级：基本型（1）、增强型（2）和综合性。

数据处理SQL编辑界面：

数据流转到表格存储编辑界面：

互联网小常识：DHCP服务器配置的术语、主要参数包括：作用域、排除范围、地址池、租约、保留和选项。

当设备有数据上报后，我们就可以在表格存储的iot_data表中实时看到存储的数据了。如下图：

在企业实例的日志服务中，我们可以查看到完整的流转日志，协助我们排查数据链路异常。如下图：

业务服务器实时接收数据

IoT场景中有些数据需要业务系统实时处理，这时我们可以通过服务端订阅AMQP方式，实时接收设备上报的数据。

首先，我们要在企业实例的服务端订阅中，创建一个新的消费组，用来接收特定类型的消息。如下图：

然后，我们在云产品流转中创建规则引擎，编写数据处理SQL，配置流转目的地为上面创建的服务端订阅消费组。

最后，我们在业务服务器编写程序，使用阿里云账号的AccessKey与IoT企业实例建立AMQP连接，参考代码如下：

publicstaticvoidmain(String[] args) throws Exception {//参数说明StringaccessKey ="子账号accessKey";StringaccessSecret ="子账号accessSecret";StringconsumerGroupId ="消费组Id";StringiotInstanceId ="企业实例Id"; long timeStamp = System.currentTimeMillis();//签名方法：支持hmacmd5，hmacsha1和hmacsha256StringsignMethod ="hmacsha1";StringclientId ="ecs_"+System.currentTimeMillis();StringuserName = clientId +"|authMode=aksign"+",signMethod="+ signMethod +",timestamp="+ timeStamp +",authId="+ accessKey +",iotInstanceId="+ iotInstanceId +",consumerGroupId="+ consumerGroupId +"|";//password组装StringsignContent ="authId="+ accessKey +"×tamp="+ timeStamp;Stringpassword = doSign(signContent,accessSecret, signMethod);//按照qpid-jms的规范，组装连接URL。StringconnectionUrl ="failover:(amqps://"+iotInstanceId+".amqp.iothub.aliyuncs.com:5671?amqp.idleTimeout=80000)"+"?failover.reconnectDelay=30"; Hashtable<String,String> hashtable =newHashtable<>(); hashtable.put("connectionfactory.SBCF",connectionUrl); hashtable.put("queue.QUEUE","default"); hashtable.put(Context.INITIAL_CONTEXT_FACTORY,"org.apache.qpid.jms.jndi.JmsInitialContextFactory"); Context context =newInitialContext(hashtable); ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF"); Destination queue = (Destination)context.lookup("QUEUE");// Create ConnectionConnection connection = cf.createConnection(userName, password); ((JmsConnection) connection).addConnectionListener(myJmsConnectionListener);// Create Session// Session.CLIENT_ACKNOWLEDGE: 收到消息后，需要手动调用message.acknowledge()// Session.AUTO_ACKNOWLEDGE: SDK自动ACK（推荐）Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); connection.start();// Create Receiver LinkMessageConsumer consumer = session.createConsumer(queue); consumer.setMessageListener(messageListener); }

在AMQP的回调中处理收到的业务数据，参考代码如下：

privatestaticMessageListener messageListener =newMessageListener() {@OverridepublicvoidonMessage(Message message){try{//如果要对收到的消息做耗时的处理，请异步处理，确保这里不要有耗时逻辑。byte[] body = message.getBody(byte[].class); String content =newString(body); String topic = message.getStringProperty("topic"); String messageId = message.getStringProperty("messageId"); String tag = message.getStringProperty("tag"); logger.info("receive message"+",\n topic = "+ topic +",\n messageId = "+ messageId +",\n tag = "+ tag +",\n content = "+ content +"\n"); System.out.println(); }catch(Exception e) { e.printStackTrace(); } } };

启动业务服务器后，我们看到不断有设备数据流转过来，如下图：

在企业实例的控制台，服务端订阅中，我们也可以看到消费组的运行情况，包括消费速率，消息堆积量，消费客户端列表，如下图：

在企业实例的控制台，日志服务中，我们可以看到完整的消息流转日志，如下图：

下达云端控制指令

业务系统通过调用IoT物联网平台提供的HTTPS API 可以给指定设备下发控制指令，调用过程如下：

Pub API调用的参考代码：

constco =require(co);constRPCClient =require(@alicloud/pop-core).RPCClient;constoptions = {accessKey:"子账号accessKey",accessKeySecret:"子账号accessKeySecret"};//1.初始化 API Clientconstclient =newRPCClient({accessKeyId: options.accessKey,secretAccessKey: options.accessKeySecret,endpoint:https://iot.cn-beijing.aliyuncs.com,apiVersion:2018-01-20, });// 指令内容constpayload = {washingMode:2,washingTime:30};//2.构建Pub API 请求constparams = {TopicFullName:"下行指令的Topic",MessageContent:newBuffer(JSON.stringify(payload)).toString("base64"),ProductKey:"产品ProductKey",IotInstanceId:"实例化Id",Qos:1}; co(function*(){//3.发起Pub API调用try{constresponse =yieldclient.request(Pub, params);console.log("Pub SUCCESS =====>",JSON.stringify(response)); }catch(err) {console.log("Pub ERROR =====>",JSON.stringify(err)); } });

运行日志：

在企业实例的控制台，日志服务中，我们也可以追踪到完整的下行链路日志，如下图：

互联网小常识：IEEE802.11无线局域网协议中，冲突检测存在一定的问题(Near/Far现象)，因为不能一边发送一边接收，所以无法使用CSMA/CD,将其改为CSMA/CA或DCF.另一个问题无线MAC层的问题是“hidden node”的问题，这个问题通过RTS/CTS解决。

免责声明：本站所有信息均搜集自互联网，并不代表本站观点，本站不对其真实合法性负责。如有信息侵犯了您的权益，请告知，本站将立刻处理。联系QQ：1640731186