关于储存:物联网平台从设备端到云端业务系统全链路开发实战实践类

传送门：5 个视频解说，30 个场景案例汇总 https://www.yuque.com/cloud-d…

通过浏览本文你将学会一下技能：
1. 设施通过 MQTT 协定与您在阿里云上购买的 IoT 企业实例建设双向连贯，上报设施采集的数据，监听云端下达的指令；
2. 通过规定引擎配置把上报的数据实时存储到指定数据库，无需编写代码
3. 通过规定引擎配置把上报的数据实时流转到业务服务器，须要应用 AMQP 协定 SDK
4. 业务服务器调用 IoT 平台的 API，下达控制指令到设施端

首先，咱们登录物联网平台控制台(https://iot.console.aliyun.com), 点击购买实例来创立一个企业实例。也能够用公共实例收费测试。

而后，在购买页面，依据理论业务需要，抉择地区、实例类型、设施数量、音讯上下行 TPS、规定引擎 TPS 等参数，点击立刻购买，付费胜利后，即可看到企业实例创立中。

稍等几分钟后，企业实例创立实现。进入企业实例，咱们能够看到以后规格参数，设施接入点信息，AMQP 订阅接入点信息，云端 API 调用接入点信息。如下图：

在企业实例的设施 治理页面 ，咱们须要先创立一个产品 家庭温控器 ，数据通信以JSON 格局，认证形式为 设施秘钥。

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

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

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

残缺的 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);
}

启动模仿脚本后，咱们看到设施状态为在线， 物模型数据 中能够看到最新上报的 温度和湿度值。

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

也能够无需编程，采纳设施模拟器做设施行为模仿

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

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

配置数据目的地：

创立解析规定，将数据流转到表格存储脚本编辑界面：

当设施有数据上报后，咱们就能够在表格存储的 iot_data 表中实时看到存储的数据了。如下图：

在企业实例的日志服务中，咱们能够查看到残缺的流转日志，帮助咱们排查数据链路异样。如下图：

IoT 场景中有些数据须要业务零碎实时处理，这时咱们能够通过服务端订阅 AMQP 形式，实时接管设施上报的数据。
首先，咱们要在企业实例的服务端订阅中，创立一个新的生产组，用来接管特定类型的音讯。如下图：

而后，咱们在云产品流转中创立解析规定配置流转目的地为下面创立的服务端订阅生产组。

最初，咱们在业务服务器编写程序，应用阿里云账号的 AccessKey 与 IoT 企业实例建设 AMQP 连贯，参考代码如下：

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 + "&timestamp=" + 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();
            }
        }
    };

启动业务服务器后，咱们看到一直有设施数据流转过来，如下图：

在企业实例的控制台，服务端订阅中，咱们也能够看到生产组的运行状况，包含生产速率，音讯沉积量，生产客户端列表，如下图：

在企业实例的控制台，日志服务中，咱们能够看到残缺的音讯流转日志，如下图：

业务零碎通过调用 IoT 物联网平台提供的 HTTPS API 能够给指定设施下发控制指令，调用过程如下：

Pub API 调用的参考代码：

const co = require('co');
const RPCClient = require('@alicloud/pop-core').RPCClient;
const options = {
    accessKey: "子账号 accessKey",
    accessKeySecret: "子账号 accessKeySecret"
};
//1. 初始化 API Client
const 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));
    }
});

运行日志：

在企业实例的控制台，日志服务中，咱们也能够追踪到残缺的上行链路日志，如下图：

物联网平台产品介绍详情：https://www.aliyun.com/produc…

             阿里云物联网平台客户交换群

关于储存:物联网平台从设备端到云端业务系统全链路开发实战实践类

一、企业物联网平台开发实战

二、创立企业实例

三、创立产品和注册设施

四、设施接入和上报数据

五、数据存储到数据库

六、业务服务器实时接收数据

七、下达云端控制指令