阿里云物联网平台自定义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概述

LoRaWAN设备数据解析及开源MQTT SDK设备端模拟