阿裡雲物聯網平台自定義Topic腳本解析功能示範

作者：俏巴

概述

之前物聯網平台自定義Topic均做消息的直接透傳，不做類似物模型模式的資料腳本解析。平台最新推出的自定義Topic腳本解析功能，裝置通過攜帶解析标記（?_sn=default）的自定義Topic上報資料，物聯網平台收到資料後，調用您在控制台送出的資料解析腳本，将自定義格式資料轉換為JSON結構體，再流轉給後續業務系統。本文主要示範該功能的具體功能實作。

Step By Step

1、建立産品和裝置

2、添加腳本

示例腳本

3、裝置端通過自定義Topic模拟上行資料

import com.alibaba.taro.AliyunIoTSignUtil;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;           

// 透傳類裝置測試 public class IoTDemoPubSubDemoForPersonalTopic {

<span class="hljs-comment">// 裝置三元組資訊</span>
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> productKey = <span class="hljs-string">"a1wJG******"</span>;
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> deviceName = <span class="hljs-string">"Device1"</span>;
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> deviceSecret = <span class="hljs-string">"40YEyiGzXmvhDdpvbUVFCHjC********"</span>;
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> regionId = <span class="hljs-string">"cn-shanghai"</span>;

private <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> pubTopic = <span class="hljs-string">"/"</span>+ productKey + <span class="hljs-string">"/"</span> + deviceName  + <span class="hljs-string">"/user/update?_sn=default"</span>;<span class="hljs-comment">//自定義Topic做腳本解析</span>
private <span class="hljs-keyword">static</span> MqttClient mqttClient;

public <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> main(<span class="hljs-built_in">String</span> [] args){

    initAliyunIoTClient(); <span class="hljs-comment">// 初始化Client</span>
    ScheduledExecutorService scheduledThreadPool = <span class="hljs-keyword">new</span> ScheduledThreadPoolExecutor(<span class="hljs-number">1</span>,
            <span class="hljs-keyword">new</span> ThreadFactoryBuilder().setNameFormat(<span class="hljs-string">"thread-runner-%d"</span>).build());

    scheduledThreadPool.scheduleAtFixedRate(()-&gt;postDeviceProperties(), <span class="hljs-number">10</span>,<span class="hljs-number">10</span>, TimeUnit.SECONDS);
}

<span class="hljs-comment">/**
 * 初始化 Client 對象
 */</span>
private <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> initAliyunIoTClient() {

    <span class="hljs-keyword">try</span> {
        <span class="hljs-comment">// 構造連接配接需要的參數</span>
        <span class="hljs-built_in">String</span> clientId = <span class="hljs-string">"java"</span> + System.currentTimeMillis();
        <span class="hljs-built_in">Map</span>&lt;<span class="hljs-built_in">String</span>, <span class="hljs-built_in">String</span>&gt; params = <span class="hljs-keyword">new</span> HashMap&lt;&gt;(<span class="hljs-number">16</span>);
        params.put(<span class="hljs-string">"productKey"</span>, productKey);
        params.put(<span class="hljs-string">"deviceName"</span>, deviceName);
        params.put(<span class="hljs-string">"clientId"</span>, clientId);
        <span class="hljs-built_in">String</span> timestamp = <span class="hljs-built_in">String</span>.valueOf(System.currentTimeMillis());
        params.put(<span class="hljs-string">"timestamp"</span>, timestamp);
        <span class="hljs-comment">// cn-shanghai</span>
        <span class="hljs-built_in">String</span> targetServer = <span class="hljs-string">"tcp://"</span> + productKey + <span class="hljs-string">".iot-as-mqtt."</span>+regionId+<span class="hljs-string">".aliyuncs.com:1883"</span>;

        <span class="hljs-built_in">String</span> mqttclientId = clientId + <span class="hljs-string">"|securemode=3,signmethod=hmacsha1,timestamp="</span> + timestamp + <span class="hljs-string">"|"</span>;
        <span class="hljs-built_in">String</span> mqttUsername = deviceName + <span class="hljs-string">"&amp;"</span> + productKey;
        <span class="hljs-built_in">String</span> mqttPassword = AliyunIoTSignUtil.sign(params, deviceSecret, <span class="hljs-string">"hmacsha1"</span>);

        connectMqtt(targetServer, mqttclientId, mqttUsername, mqttPassword);

    } <span class="hljs-keyword">catch</span> (Exception e) {
        System.out.println(<span class="hljs-string">"initAliyunIoTClient error "</span> + e.getMessage());
    }
}

public <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> connectMqtt(<span class="hljs-built_in">String</span> url, <span class="hljs-built_in">String</span> clientId, <span class="hljs-built_in">String</span> mqttUsername, <span class="hljs-built_in">String</span> mqttPassword) throws Exception {

    MemoryPersistence persistence = <span class="hljs-keyword">new</span> MemoryPersistence();
    mqttClient = <span class="hljs-keyword">new</span> MqttClient(url, clientId, persistence);
    MqttConnectOptions connOpts = <span class="hljs-keyword">new</span> MqttConnectOptions();
    <span class="hljs-comment">// MQTT 3.1.1</span>
    connOpts.setMqttVersion(<span class="hljs-number">4</span>);
    connOpts.setAutomaticReconnect(<span class="hljs-literal">false</span>);
    connOpts.setCleanSession(<span class="hljs-literal">true</span>);

    connOpts.setUserName(mqttUsername);
    connOpts.setPassword(mqttPassword.toCharArray());
    connOpts.setKeepAliveInterval(<span class="hljs-number">60</span>);

    mqttClient.connect(connOpts);
}

<span class="hljs-comment">/**
 * 彙報屬性
 */</span>
private <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> postDeviceProperties() {

    <span class="hljs-keyword">try</span> {
        <span class="hljs-comment">//上報資料</span>
        System.out.println(<span class="hljs-string">"自定義Topic上報屬性值"</span>);
        <span class="hljs-comment">//0x000000000100320100000000</span>
        <span class="hljs-built_in">String</span> hexString = <span class="hljs-string">"000000000100320100000000"</span>;
        byte[] payLoad = hexToByteArray(hexString);
        MqttMessage message = <span class="hljs-keyword">new</span> MqttMessage(payLoad);
        message.setQos(<span class="hljs-number">0</span>);
        mqttClient.publish(pubTopic, message);
    } <span class="hljs-keyword">catch</span> (Exception e) {
        System.out.println(e.getMessage());
    }
}

<span class="hljs-comment">/**
 * hex字元串轉byte數組
 * @param inHex 待轉換的Hex字元串
 * @return  轉換後的byte數組結果
 */</span>
public <span class="hljs-keyword">static</span> byte[] hexToByteArray(<span class="hljs-built_in">String</span> inHex){
    int hexlen = inHex.length();
    byte[] result;
    <span class="hljs-keyword">if</span> (hexlen % <span class="hljs-number">2</span> == <span class="hljs-number">1</span>){
        <span class="hljs-comment">//奇數</span>
        hexlen++;
        result = <span class="hljs-keyword">new</span> byte[(hexlen/<span class="hljs-number">2</span>)];
        inHex=<span class="hljs-string">"0"</span>+inHex;
    }<span class="hljs-keyword">else</span> {
        <span class="hljs-comment">//偶數</span>
        result = <span class="hljs-keyword">new</span> byte[(hexlen/<span class="hljs-number">2</span>)];
    }
    int j=<span class="hljs-number">0</span>;
    <span class="hljs-keyword">for</span> (int i = <span class="hljs-number">0</span>; i &lt; hexlen; i+=<span class="hljs-number">2</span>){
        result[j]=hexToByte(inHex.substring(i,i+<span class="hljs-number">2</span>));
        j++;
    }
    <span class="hljs-keyword">return</span> result;
}

<span class="hljs-comment">/**
 * Hex字元串轉byte
 * @param inHex 待轉換的Hex字元串
 * @return  轉換後的byte
 */</span>
public <span class="hljs-keyword">static</span> byte hexToByte(<span class="hljs-built_in">String</span> inHex) {
    <span class="hljs-keyword">return</span> (byte) Integer.parseInt(inHex, <span class="hljs-number">16</span>);
}                

}

參考連結： LoRaWAN裝置資料解析及開源MQTT SDK裝置端模拟

4、上行消息檢視

5、注意事項

1、在物聯網平台建立自定義Topic時按正常Topic定義，不添加該解析标記；
2、僅解析裝置上報雲端的資料，不解析雲端下行資料；
3、僅解析上報資料的Payload，并傳回解析後的Payload；
4、解析前後，資料所在Topic不變。例如，裝置發送到/${productKey}/${deviceName}/user/update的資料，解析後仍在該Topic中。           

參考連結

自定義資料解析Topic概述