消息服務實時消費設備狀態變化和數據

0.準備工做

0.1 註冊阿里雲帳號

使用我的淘寶帳號或手機號，開通阿里雲帳號，並經過實名認證(能夠用支付寶認證)

0.2 免費開通IoT物聯網套件

IoT套件產品官網 https://www.aliyun.com/product/iot
消息服務官網 https://mns.console.aliyun.com/#/list/cn-shanghai

0.3 軟件環境

Nodejs安裝 https://nodejs.org/en/download/
Java8 安裝
開發工具:Sublime Text3/ IntelliJ IDEA
https://www.sublimetext.com/3

1.阿里雲IoT控制檯配置服務端訂閱

阿里雲IoT物聯網套件開通 https://www.aliyun.com/product/iot

1.1 建立產品（基礎版）

咱們在阿里雲IoT控制檯，建立產品：空氣檢測，選擇基礎版。

1.2 在產品下添加設備

咱們在阿里雲IoT控制檯，設備管理裏空氣檢測產品下添加一個具體設備。

1.2 在產品下配置服務端訂閱

咱們在阿里雲IoT控制檯，爲空氣檢測產品開通服務端訂閱，勾選設備上報消息和設備狀態變化通知，開通後會有MNS的區域：cn-shanghai和隊列信息：aliyun-iot-a1jnUEKYhw4，以下。

經過閱讀阿里雲IoT文檔，咱們瞭解到隊列中消息結構體以下：java

複製代碼node

{
"payload": "Base64 Encode的數據",
"messagetype": "status",
"messageid": 996000000000000001,
"topic": "具體的設備Topic",
"timestamp": 1526450324
}

messageid：IoT套件生成的消息ID
messagetype：指的是消息類型:設備狀態status和設備上報消息upload
topic：表示該消息源自套件中的哪一個topic，當messageType=status時，topic爲null，當messageType=upload時，topic爲具體的設備Topic
payload：數據爲Base64 Encode的數據。當messageType=status時，數據是設備狀態數據；當messageType=upload時，data即爲設備發佈到Topic中的原始數據。
timestamp：隊列中消息生成時間戳，並不是業務的時間戳

2.設備端開發

咱們採用nodejs腳本模擬設備，與IoT雲端創建鏈接，上報數據。

2.1 獲取nodejs版IoT SDK

package.json中添加npm依賴"aliyun-iot-mqtt": "0.0.4"模塊apache

複製代碼npm

{
"name": "aliyun-iot",
"dependencies": {
"aliyun-iot-mqtt": "^0.0.4"
},
"author": "wongxming",
"license": "MIT"
}

下載安裝SDKjson

複製代碼dom

$npm install

2.2 編寫設備端應用程序代碼

咱們須要在控制檯獲取設備身份三元組：productKey，deviceName，deviceSecret，以及產品分區regionIdasync

複製代碼工具

/**
* package.json 添加依賴："aliyun-iot-mqtt": "0.0.4"
node iot-mns.js
*/
const mqtt = require('aliyun-iot-mqtt');
//設備三元組
const options = {
productKey: "RY8ExdyS6lU",
deviceName: "officeThermometer",
deviceSecret: "oauYYavdIV9QOt7d9WcrnIjXSNc2i26A",
regionId: "cn-shanghai"
};
//設備與雲創建鏈接，設備上線
const client = mqtt.getAliyunIotMqttClient(options);
//主題topic
const topic = `${options.productKey}/${options.deviceName}/update`;
var data = {
temperature: Math.floor((Math.random()*20)+10),
humidity: Math.floor((Math.random()*80)+20),
};
//指定topic發佈數據到雲端
client.publish(topic, JSON.stringify(data));
console.log("===postData topic=" + topic)
console.log(data)
//設備下線
//client.end()

2.3 啓動模擬設備腳本

post

複製代碼開發工具

$node iot-mns.js

腳本執行後，咱們在IoT雲端控制檯產品-日誌服務裏查看設備行爲分析日誌，根據時間順序，咱們看到

首先在10:53:05時，設備創建鏈接，上線；
而後在10:53:12時，設備斷開鏈接，下線。

查看設備上行消息分析日誌，根據時間順序，咱們看到

首先設備publish message，
而後流轉到RuleEngine規則引擎，
最終Transmit MNS的消息隊列aliyun-iot-a1jnUEKYhw4

咱們切換到MNS控制檯，選擇華東2區域，能夠看到消息隊列aliyun-iot-a1jnUEKYhw4有3條活躍消息MNS控制檯：https://mns.console.aliyun.com/#/list/cn-shanghai

3 消費隊列中設備消息

3.1 消息服務MNS使用

咱們以java開發爲例，pom中添加依賴aliyun-sdk-mns，httpasyncclient，fastjson，以下：

複製代碼

<dependencies>
<dependency>
<groupId>com.aliyun.mns</groupId>
<artifactId>aliyun-sdk-mns</artifactId>
<version>1.1.8</version>
<classifier>jar-with-dependencies</classifier>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpasyncclient</artifactId>
<version>4.0.1</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.42</version>
</dependency>
</dependencies>

咱們經過mns的sdk，建立鏈接，輪詢獲取隊列消息。爲了方便閱讀，咱們對消息的payload數據作base64解碼，完整應用程序代碼以下：

複製代碼

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.mns.client.CloudAccount;
import com.aliyun.mns.client.CloudQueue;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import org.apache.commons.codec.binary.Base64;
public class ComsumerDemo {
public static String accessKeyId = "AK";
public static String accessKeySecret = "AK祕鑰";
public static String endpoint = "mns公網Endpoint";
public static void main(String[] args) {
CloudAccount account = new CloudAccount(accessKeyId,accessKeySecret,endpoint);
MNSClient client = account.getMNSClient();
//獲取消息隊列
CloudQueue queue = client.getQueueRef("aliyun-iot-a1jnUEKYhw4");
while (true) {
Message popMsg = queue.popMessage(10);
if (popMsg != null) {
System.out.println("message id: " + popMsg.getMessageId());
System.out.println("message body: \n" + decodeBase64(popMsg.getMessageBodyAsString()));
//刪除消息
queue.deleteMessage(popMsg.getReceiptHandle());
}
}
}
public static String decodeBase64(String jsonString) {
try {
JSONObject json = JSON.parseObject(jsonString);
String payload = new String(Base64.decodeBase64(json.getString("payload")));
json.put("payload", JSON.parseObject(payload));
return json.toJSONString();
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}

3.2 運行程序

控制檯輸出以下，咱們看到1) timestamp=1526450324時，設備上線消息，

複製代碼

message id: E2CF179AD5686386-2-16367878417-200000009
message body:
{
"payload": {
"lastTime": "2018-05-16 13:58:44.413",
"clientIp": "42.120.74.246",
"time": "2018-05-16 13:58:44.427",
"productKey": "a1jnUEKYhw4",
"deviceName": "suw8umOqgJ72yCADerZp",
"status": "online"
},
"messagetype": "status",
"messageid": 996631012001751041,
"timestamp": 1526450324
}

2) timestamp=1526450334時，設備上報了數據，經過payload能夠看到完整數據

複製代碼

message id: "656A4F66B391367-1-1636787AAC0-20000000C"
message body:
{
"payload": {
"temperature": 14,
"humidity": 116
},
"messagetype": "upload",
"topic": "/a1jnUEKYhw4/suw8umOqgJ72yCADerZp/update",
"messageid": 996631053735047169,
"timestamp": 1526450334
}

3) timestamp=1526450353時，設備下線消息

複製代碼

message id: E2CF179AD5686386-2-1636787F5F1-20000000A
message body:
{
"payload": {
"lastTime": "2018-05-16 13:59:04.381",
"time": "2018-05-16 13:59:13.571",
"productKey": "a1jnUEKYhw4",
"deviceName": "suw8umOqgJ72yCADerZp",
"status": "offline"
},
"messagetype": "status",
"messageid": 996631134240534528,
"timestamp": 1526450353
}閱讀後請點擊