IoT 企业物联网平台,从设备端到云端业务系统全链路开发实战——实践类
一、企业物联网平台开发实战
通过阅读本文你将学会一下技能:
设备通过 MQTT 协议与您在阿里云上购买的 IoT 企业实例建立双向连接,上报设备采集的数据,监听云端下达的指令;
通过规则引擎配置把上报的数据实时存储到指定数据库,无需编写代码
通过规则引擎配置把上报的数据实时流转到业务服务器,需要使用 AMQP 协议 SDK
业务服务器调用 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 + "×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();
}
}
};
启动业务服务器后,我们看到不断有设备数据流转过来,如下图:
在企业实例的控制台,服务端订阅中,我们也可以看到消费组的运行情况,包括消费速率,消息堆积量,消费客户端列表,如下图:
在企业实例的控制台,日志服务中,我们可以看到完整的消息流转日志,如下图:
七、下达云端控制指令
业务系统通过调用 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/product/iot/iot_instc_public_cn
阿里云物联网平台客户交流群
相关内容推荐
BCSS医学术语学术诚信报告3000陈浮 学术学术人格化赵县职高学术造假学术先锋培训学习国际学术出版社仝卓学术造假事件你如何看待学术写作如何增强学术诚信做学术报道英文农学类学术海报学术调查方案格式医学术语TE学术高雅的成语校长学术思维大学电子学术直接材料学术定义张可锋学术钛学术可靠吗怎样给学术投资学术不端典例阮元的学术地位藏医学术翻译公司创业学术培训长沙卫星导航学术畅享学术资源吞咽障碍学术研究根哥学术入口对待学术严谨专业中国的学术泡沫我的学术部落学术节幼儿园学术委员会方案雅思学术类作文历史史料学术会学术创新好青年多个蛋黄学术名词学术空白的领域学术论文2000以上书的学术定位电子支付学术ppt海报学术宣传图片结核学术会议通知美国封锁学术组织学术书籍好看吗严肃的学术讨论刘凤翥学术数字媒体学术问题学术整顿 王贻芳培育教师的学术精神学术资源被垄断了医院学术邀请函静态学术都有什么就学术水平而言选题学术性格萨尔学术动态学术推广英文缩写为学术主任团队工作最近的学术翻译学术 会议通知2021 图像学术会议学术演讲有趣主题鲁实先学术积极进行学术钻研谢晖学术名著青塔学术统计怎样做学术的学术文件检索时学术道德收获大文学术英语我要学术的赵海金学术泰安丁珊珊学术周济的学术成就机构学术与个人化学术美女的专业学术李福军学术李应存学术成就发端的学术说法德国教师荒学术2022学术期刊医学术语MBS学术素养意义何在学术博弈中的对话个人学术创新学术论文 报纸汉武帝的学术王俊利学术问题王奕森 学术黑猫学术研究动态谷歌学术怎么删除文章学术作风自我总结学术圈怎么接触预防学术不端总结国际学术分级阅读答案学术圈怎么翻译路径选择学术用语唐的学术流派上海交大 学术之星学术成果如何介绍医学术语reb拓展学术组织功能学术眼睛疼吗国外高级学术殿堂论文如何形容学术道德陈玉海学术点阵激光 控油 学术生死疲劳学术评价良好学术氛围照片学术专著延迟交稿说明学术理论日推改造google学术搜索达超学术国内年度学术热点黄兴涛学术爵士乐学术介绍杨柳学术不端学术 hzq613绿色矿山主要学术贡献钛学术客服电话学术理论可行性学术比赛赛后采访云端学术资源app雅思写作 学术教育学术问卷app原创学术讲座海报山东学术不端聊城uus技术国际学术音乐类学术总结学术考研互动社区nsfc 学术交流学术应用平台项目00后学术称谓学术小萌新学术支撑教学潇湘溪苑学术推出学术论文高校学术舞会活动博士学术型学位点写作学术热点学术论文建议品牌营销 学术营销高超学术水平英文必应x学术学术写在大地上学术论文奖励学术制照片学术诈骗论文题目心系学术为民担当学术志趣内容包括刘玫学术成就成人学术定义医学术语animal艺术与学术学位学术研究与实践学术造假亲告医学术语fnb学术讲师待遇如何学术课教学原则学术推广的图片如何远离学术偏见雅思nawl学术进阶式阅读学术价值学术格子下架重学术轻临床英语学术讲座听力国外学术著作学术沙龙开展方案学术型硕士点澳中学术杂志凌海学术期刊润色王力学术讲座英文李昌钰破案实录学术思考职业和学术道德杰出学术兼职人员学术论文档次学术著作稿件优质学术分享论文学术 庞氏骗局韦伯学术与政治txt读秀学术怎么进入靳琦学术实用性学术学术自由摘要生源合成学术古代朝堂学术学术创业大赛赡养英语学术用法学术侦探挖出2000专业是学术圣地学校学术竞赛主题山东学术会议网盘古学术会议张筑生学术学术资源积累宋史学术大佬学术论坛文件要求学术团队经费预算论文404学术不端