阿里云物联网平台设备分发实战

一、源实例创建产品和设备

• 1.1 登录物联网平台控制台，选择实例进入

• 1.2 创建产品和设备

• 1.3 Java MQTT连接设备

源码参考： 基于开源JAVA MQTT Client连接阿里云IoT

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import 一机一密.AliyunIoTSignUtil;

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 shebenbufenfa {

    public static String productKey = "******";
    public static String deviceName = "sourcedevice1";
    public static String deviceSecret = "******";
    public static String instanceId = "iot-******";

    // 物模型-属性上报topic
    private static String pubTopic = "/sys/" + productKey + "/" + deviceName + "/thing/event/property/post";
    // 物模型-属性响应topic
    private static String subTopic = "/sys/" + productKey + "/" + deviceName + "/thing/event/property/post_reply";
    private static MqttClient mqttClient;

    public static void main(String [] args){

        initAliyunIoTClient();
        ScheduledExecutorService scheduledThreadPool = new ScheduledThreadPoolExecutor(1,
                new ThreadFactoryBuilder().setNameFormat("thread-runner-%d").build());

        scheduledThreadPool.scheduleAtFixedRate(()->postDeviceProperties(), 2,2, TimeUnit.SECONDS);

        try {
            mqttClient.subscribe(subTopic); // 订阅Topic
        } catch (MqttException e) {
            System.out.println("error:" + e.getMessage());
            e.printStackTrace();
        }

        // 设置订阅监听
        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable throwable) {
                System.out.println("connection Lost");
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                System.out.println("Sub message");
                System.out.println("Topic : " + s);
                System.out.println("message info " + new String(mqttMessage.getPayload())); //打印输出消息payLoad
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                try {
                    System.out.println("deliveryComplete: " + iMqttDeliveryToken.getMessage());
                } catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 初始化 Client 对象
     */
    private static void initAliyunIoTClient() {

        try {
            // 构造连接需要的参数
            String clientId = "java" + System.currentTimeMillis();
            Map<String, String> params = new HashMap<>(16);
            params.put("productKey", productKey);
            params.put("deviceName", deviceName);
            params.put("clientId", clientId);
            String timestamp = String.valueOf(System.currentTimeMillis());
            params.put("timestamp", timestamp);
            // cn-shanghai
            String targetServer = "tcp://" + instanceId + ".mqtt.iothub.aliyuncs.com:1883";
            String mqttclientId = clientId + "|securemode=3,signmethod=hmacsha1,timestamp=" + timestamp + "|";
            String mqttUsername = deviceName + "&" + productKey;
            String mqttPassword = AliyunIoTSignUtil.sign(params, deviceSecret, "hmacsha1");

            connectMqtt(targetServer, mqttclientId, mqttUsername, mqttPassword);

        } catch (Exception e) {
            System.out.println("initAliyunIoTClient error " + e.getMessage());
        }
    }

    public static void connectMqtt(String url, String clientId, String mqttUsername, String mqttPassword) throws Exception {

        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(url, clientId, persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        // MQTT 3.1.1
        connOpts.setMqttVersion(4);
        connOpts.setAutomaticReconnect(false);
        connOpts.setConnectionTimeout(10);
//        connOpts.setCleanSession(true);
        connOpts.setCleanSession(false);

        connOpts.setUserName(mqttUsername);
        connOpts.setPassword(mqttPassword.toCharArray());
        connOpts.setKeepAliveInterval(60);

        mqttClient.connect(connOpts);
    }

    /**
     * 汇报属性
     */
    private static void postDeviceProperties() {

        try {
            //上报数据
            //高级版 物模型-属性上报payload
            System.out.println("上报属性值");
            // 基于产品物模型定义设置
            String payloadJson = "{\"params\":{\"LightStatus\":1,\"LightAdjustLevel\":15}}";
            MqttMessage message = new MqttMessage(payloadJson.getBytes("utf-8"));
            message.setQos(1);
            mqttClient.publish(pubTopic, message);
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
}

• 1.4 查看设备在线及数据上报情况

二、设备分发（指定地域）

• 2.1 设备划归 -> 设备分发

• 2.2 目标账户查看

请求物联网平台Server，获取新的设备认证地址

可以基于官方SDK使用其：Bootstrap功能开发的具体操作，请参见概述。这里介绍使用okhttp自行接入方式接入

• 3.1 pom.xml

        <!-- https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp -->
        <dependency>
            <groupId>com.squareup.okhttp3</groupId>
            <artifactId>okhttp</artifactId>
            <version>4.4.1</version>
        </dependency>

• 3.2 Code Sample

import okhttp3.*;
import java.io.IOException;

public class demo1 {

    public static void main(String[] args) throws IOException {

        OkHttpClient client = new OkHttpClient().newBuilder()
                .build();
        MediaType mediaType =MediaType.parse("application/x-www-form-urlencoded");
        RequestBody body = RequestBody.create(mediaType,"productKey=hz56a******&deviceName=sourced*****&clientId=taroclient1&version=v2&timestamp=1669267341");

        Request request = new Request.Builder()
                .url("https://iot-auth-global.aliyuncs.com/auth/bootstrap/")
                .method("POST", body)
                .addHeader("Content-Type", "application/x-www-form-urlencoded")
                .build();
        Response response = client.newCall(request).execute();
        System.out.println(response.body().string());
    }
}

• 3.3 返回结果

{"code":200,"data":{"instanceId":"iotx-oxss********","resources":{"mqtt":{"host":"public.iot-as-mqtt.cn-shanghai.aliyuncs.com","ip":"139.224.42.2","port":1883}}},"message":"success"}

设备分发到新的用户或者实例后，设备三元组信息不变，仅连接域名会发生变化，连接域名修改为上面请求获取的域名即可。

• 3.4 划归后的设备测试情况

更多参考

使用okhttp时没有返回正常数据
设备分发
Can't get OkHttp's response.body.toString() to return a string