一、企业物联网平台开发实战
image.png
通过阅读本文你将学会一下技能：
设备通过 MQTT 协议与您在阿里云上购买的 IoT 企业实例建立双向连接，上报设备采集的数据，监听云端下达的指令；
通过规则引擎配置把上报的数据实时存储到指定数据库，无需编写代码
通过规则引擎配置把上报的数据实时流转到业务服务器，需要使用 AMQP 协议 SDK
业务服务器调用 IoT 平台的 API，下达控制指令到设备端
二、创建企业实例
首先，我们登录物联网平台控制台(https://iot.console.aliyun.com), 点击购买实例来创建一个企业实例。 也可以用公共实例免费测试。
image
然后，在购买页面，根据实际业务需求，选择地域、实例类型、设备数量、消息上下行 TPS、规则引擎 TPS 等参数，点击立即购买，付费成功后，即可看到企业实例创建中。
image
稍等几分钟后，企业实例创建完成。进入企业实例，我们可以看到当前规格参数，设备接入点信息，AMQP 订阅接入点信息，云端 API 调用接入点信息。如下图：
image
三、创建产品和注册设备
在企业实例的设备管理页面，我们需要先创建一个产品家庭温控器，数据通信以 JSON 格式，认证方式为设备秘钥。
image
在产品的功能定义页面，我们添加温度和湿度两个属性，具体细节如下图：
image
最后，我们在设备管理页面，基于家庭温控器产品，注册一个物理设备，并获取设备身份认证的三元组。如下图：
image
image
四、设备接入和上报数据
获取设备身份三元组后，即可通过 MQTT 协议接入到我们开通的企业实例。设备端应用程序逻辑如下图：
image
完整的 Nodejs 示例代码如下：
const mqtt = require('aliyun-iot-mqtt');// 1. 设备身份信息var options = { productKey: "产品productKey", deviceName: "设备deviceName", deviceSecret: "设备deviceSecret", host: "实例化MQTT接入点"};// 2. 建立MQTT连接const client = 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);function getPostData() { const payloadJson = { 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)) return JSON.stringify(payloadJson);}
复制代码
启动模拟脚本后，我们看到设备状态为在线，物模型数据中可以看到最新上报的温度和湿度值。
image
在监控运维的日志服务里，也可以看到设备上报数据的日志。如下图：
image
也可以无需编程，采用设备模拟器做设备行为模拟
image
五、数据存储到数据库
首先，我们创建一个表格存储实例 IoTDataStore，建立一张数据表 iot_data，以 deviceName 和 timestamp 为主键。如下图：
image
在 IoT 企业实例，云产品流转中创建规则引擎，编写数据处理 SQL，配置流转目的地为上面创建的数据库表。
配置数据源：
image.png
配置数据目的地：
image.png
创建解析规则，将数据流转到表格存储脚本编辑界面：
image.png
当设备有数据上报后，我们就可以在表格存储的 iot_data 表中实时看到存储的数据了。如下图：
image
在企业实例的日志服务中，我们可以查看到完整的流转日志，协助我们排查数据链路异常。如下图：
image
六、业务服务器实时接收数据
IoT 场景中有些数据需要业务系统实时处理，这时我们可以通过服务端订阅 AMQP 方式，实时接收设备上报的数据。
首先，我们要在企业实例的服务端订阅中，创建一个新的消费组，用来接收特定类型的消息。如下图：
image
然后，我们在云产品流转中创建解析规则配置流转目的地为上面创建的服务端订阅消费组。
image.png
最后，我们在业务服务器编写程序，使用阿里云账号的 AccessKey 与 IoT 企业实例建立 AMQP 连接，参考代码如下：
image
public static void main(String[] args) throws Exception { //参数说明 String accessKey = "子账号accessKey"; String accessSecret = "子账号accessSecret"; String consumerGroupId = "消费组Id"; String iotInstanceId = "企业实例Id"; long timeStamp = System.currentTimeMillis(); //签名方法：支持hmacmd5，hmacsha1和hmacsha256 String signMethod = "hmacsha1"; String clientId = "ecs_"+System.currentTimeMillis(); String userName = clientId + "|authMode=aksign" + ",signMethod=" + signMethod + ",timestamp=" + timeStamp + ",authId=" + accessKey + ",iotInstanceId=" + iotInstanceId + ",consumerGroupId=" + consumerGroupId + "|"; //password组装 String signContent = "authId=" + accessKey + "×tamp=" + timeStamp; String password = doSign(signContent,accessSecret, signMethod); //按照qpid-jms的规范，组装连接URL。 String connectionUrl = "failover:(amqps://"+iotInstanceId+".amqp.iothub.aliyuncs.com:5671?amqp.idleTimeout=80000)" + "?failover.reconnectDelay=30"; Hashtable<String, String> hashtable = new Hashtable<>(); hashtable.put("connectionfactory.SBCF",connectionUrl); hashtable.put("queue.QUEUE", "default"); hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory"); Context context = new InitialContext(hashtable); ConnectionFactory cf = (ConnectionFactory)context.lookup("SBCF"); Destination queue = (Destination)context.lookup("QUEUE"); // Create Connection Connection 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 Link MessageConsumer consumer = session.createConsumer(queue); consumer.setMessageListener(messageListener); }
复制代码
在 AMQP 的回调中处理收到的业务数据，参考代码如下：
private static MessageListener messageListener = new MessageListener() { @Override public void onMessage(Message message) { try { //如果要对收到的消息做耗时的处理，请异步处理，确保这里不要有耗时逻辑。 byte[] body = message.getBody(byte[].class); String content = new String(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(); } } };
复制代码
启动业务服务器后，我们看到不断有设备数据流转过来，如下图：
image
在企业实例的控制台，服务端订阅中，我们也可以看到消费组的运行情况，包括消费速率，消息堆积量，消费客户端列表，如下图：
image
在企业实例的控制台，日志服务中，我们可以看到完整的消息流转日志，如下图：
image
七、下达云端控制指令
业务系统通过调用 IoT 物联网平台提供的 HTTPS API 可以给指定设备下发控制指令，调用过程如下：
image
Pub API 调用的参考代码：
const co = require('co');const RPCClient = require('@alicloud/pop-core').RPCClient;const options = { accessKey: "子账号accessKey", accessKeySecret: "子账号accessKeySecret"};//1.初始化 API Clientconst client = new RPCClient({ accessKeyId: options.accessKey, secretAccessKey: options.accessKeySecret, endpoint: 'https://iot.cn-beijing.aliyuncs.com', apiVersion: '2018-01-20',});// 指令内容const payload = { washingMode: 2, washingTime: 30};//2.构建Pub API 请求const params = { TopicFullName: "下行指令的Topic", MessageContent: new Buffer(JSON.stringify(payload)).toString("base64"), ProductKey: "产品ProductKey", IotInstanceId: "实例化Id", Qos: 1};co(function*() { //3.发起Pub API调用 try { const response = yield client.request('Pub', params); console.log("Pub SUCCESS =====>", JSON.stringify(response)); } catch (err) { console.log("Pub ERROR =====>", JSON.stringify(err)); }});
复制代码
运行日志：
image
在企业实例的控制台，日志服务中，我们也可以追踪到完整的下行链路日志，如下图：
image
物联网平台产品介绍详情：https://www.aliyun.com/product/iot/iot_instc_public_cn
阿里云物联网平台客户交流群
评论