备案控制台
开发者社区 云服务技术课堂 文章 正文

设备通过mqtt通道的动态预注册

2022-11-17 451
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 在物联网平台为产品开启动态注册功能后，直连设备可使用一型一密安全认证方式完成动态注册，通过MQTT通信协议连接物联网平台。设备先基于TLS建立与物联网平台的连接，获取MQTT连接所需的设备密钥，再断开连接，然后重新建立MQTT连接进行通信。

动态预注册设备时，当设备属于新版公共实例或企业版实例时，动态注册参数如下：

mqttClientId: clientId+"|securemode=2,authType=register,random=xxxx,signmethod=xxxx,instanceId=xxxx|"

mqttUserName: deviceName+"&"+productKey

mqttPassword: sign_hmac(productSecret,content)

当设备属于旧版公共实例时，动态注册参数如下：

mqttClientId: clientId+"|securemode=2,authType=register,random=xxxx,signmethod=xxxx|"

mqttUserName: deviceName+"&"+productKey

mqttPassword: sign_hmac(productSecret,content)

其中signmethod目前支持签名算法为：hmacmd5、hmacsha1、hmacsha256。


注册流程为：


代码实现：

/*
 * Copyright © 2019 Alibaba. All rights reserved.
 */
package com.aliyun.paho;
import java.nio.charset.StandardCharsets;
import java.util.Random;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import com.alibaba.fastjson.JSONObject;
/**
 * 设备动态注册。
 */
public class DynamicRegister {
    // 地域ID，填写您的产品所在地域ID。
    private static String regionId = "cn-shanghai";
    // 定义加密方式。可选MAC算法：HmacMD5、HmacSHA1、HmacSHA256，需和signmethod取值一致。
    private static final String HMAC_ALGORITHM = "hmacsha1";
    // 接收物联网平台下发设备证书的Topic。无需创建，无需订阅，直接使用。
    private static final String REGISTER_TOPIC = "/ext/register";
    private static final String REGNWL_TOPIC = "/ext/regnwl";
    /**
     * 动态注册。
     *
     * @param productKey 产品的ProductKey
     * @param productSecret 产品密钥
     * @param deviceName 设备名称
     * @throws Exception
     */
    public void register(String productKey, String productSecret, String deviceName) throws Exception {
        // 接入域名，只能使用TLS。
        String broker = "ssl://" + productKey + ".iot-as-mqtt." + regionId + ".aliyuncs.com:1883";
        // 表示客户端ID，建议使用设备的MAC地址或SN码，64字符内。
        String clientId = productKey + "." + deviceName;
        // 获取随机值。
        Random r = new Random();
        int random = r.nextInt(1000000);
        // securemode只能为2表示只能使用TLS；signmethod指定签名算法。
        String clientOpts = "|securemode=2,authType=register,signmethod=" + HMAC_ALGORITHM + ",random=" + random + "|";
        // MQTT接入客户端ID。
        String mqttClientId = clientId + clientOpts;
        // MQTT接入用户名。
        String mqttUsername = deviceName + "&" + productKey;
        // MQTT接入密码，即签名。
        JSONObject params = new JSONObject();
        params.put("productKey", productKey);
        params.put("deviceName", deviceName);
        params.put("random", random);
        String mqttPassword = sign(params, productSecret);
        try {
            MemoryPersistence persistence = new MemoryPersistence();
            MqttClient sampleClient = new MqttClient(broker, mqttClientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setMqttVersion(4);// MQTT 3.1.1
            connOpts.setUserName(mqttUsername);// 用户名
            connOpts.setPassword(mqttPassword.toCharArray());// 密码
            connOpts.setAutomaticReconnect(false); // MQTT动态注册协议规定必须关闭自动重连。
            System.out.println("----- register params -----");
            System.out.print("server=" + broker + ",clientId=" + clientId);
            System.out.println(",username=" + mqttUsername + ",password=" + mqttPassword);
            sampleClient.setCallback(new MqttCallback() {
                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                    System.out.println("----- register result -----");
                    System.out.println(topic);
                    System.out.println(payload);
                    // 仅处理动态注册返回消息。
                    if (REGISTER_TOPIC.equals(topic)) {
                        //sampleClient.disconnect();
                        connect(payload);
                    }else if (REGNWL_TOPIC.equals(topic)) {
                        //sampleClient.disconnect();
                    }
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.print("register deliveryComplete");
                }
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.print("register connectionLost");
                }
            });
            sampleClient.connect(connOpts);
        } catch (MqttException e) {
            System.out.print("register failed: clientId=" + clientId);
            System.out.println(",username=" + mqttUsername + ",password=" + mqttPassword);
            System.out.println("reason " + e.getReasonCode());
            System.out.println("msg " + e.getMessage());
            System.out.println("loc " + e.getLocalizedMessage());
            System.out.println("cause " + e.getCause());
            System.out.println("excep " + e);
            e.printStackTrace();
        }
    }
    private void connect(String payload) {
        System.out.println("----- payload -----"+payload);
        JSONObject request = JSONObject.parseObject(payload);
        String productKey = request.getString("productKey");
        String deviceName = request.getString("deviceName");
        String deviceSecret = request.getString("deviceSecret");
        // 接入域名，只能使用TLS。
        String broker = "ssl://" + productKey + ".iot-as-mqtt." + regionId + ".aliyuncs.com:1883";
        System.out.println("----- broker -----"+broker);
        String clientId = productKey+"."+deviceName;
        System.out.println("----- clientId -----"+clientId);
        // securemode只能为2表示只能使用TLS；signmethod指定签名算法。
        String timestamp = Long.toString(System.currentTimeMillis());
        String clientOpts = "|timestamp=" + timestamp + ",_v=paho-java-1.0.0,securemode=2,signmethod="+HMAC_ALGORITHM+"|";
        // MQTT接入客户端ID。
        String mqttClientId = clientId + clientOpts;
        // MQTT接入用户名。
        String mqttUsername = deviceName + "&" + productKey;
        System.out.println("----- mqttUsername -----"+mqttUsername);
        JSONObject params = new JSONObject();
        params.put("productKey", productKey);
        params.put("deviceName", deviceName);
        params.put("clientId", clientId);
        params.put("timestamp", timestamp);
        String mqttPassword = sign(params, deviceSecret);
        System.out.println("----- mqttPassword -----"+mqttPassword);
        try {
            MemoryPersistence persistence = new MemoryPersistence();
            MqttClient sampleClient = new MqttClient(broker, mqttClientId, persistence);
            MqttConnectOptions connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            connOpts.setKeepAliveInterval(180);
            connOpts.setUserName(mqttUsername);
            connOpts.setPassword(mqttPassword.toCharArray());
            System.out.println("----- register params -----");
            System.out.print("server=" + broker + ",clientId=" + clientId);
            System.out.println(",username=" + mqttUsername + ",password=" + mqttPassword);
            sampleClient.setCallback(new MqttCallback() {
                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    String payload = new String(message.getPayload(), StandardCharsets.UTF_8);
                    System.out.println("----- register result -----");
                    System.out.println(topic);
                    System.out.println(payload);
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("connect deliveryComplete");
                }
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("connect connectionLost");
                }
            });
            sampleClient.connect(connOpts);
            sampleClient.isConnected();
            String topic = "/"+ productKey + "/" + deviceName + "/user/update";
            String content = "{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"a\":1}}";
            MqttMessage message = new MqttMessage(content.getBytes());
            message.setQos(0);
            sampleClient.publish(topic, message);
            content = "{\"id\":\"1\",\"version\":\"1.0\",\"params\":{\"b\":2}}";
            message = new MqttMessage(content.getBytes());
            message.setQos(0);
            sampleClient.publish(topic, message);
        } catch (MqttException e) {
            System.out.print("register failed: clientId=" + clientId);
            System.out.println(",username=" + mqttUsername + ",password=" + mqttPassword);
            System.out.println("reason " + e.getReasonCode());
            System.out.println("msg " + e.getMessage());
            System.out.println("loc " + e.getLocalizedMessage());
            System.out.println("cause " + e.getCause());
            System.out.println("excep " + e);
            e.printStackTrace();
        }
    }
    /**
     * 动态注册签名。
     *
     * @param params 签名参数
     * @param productSecret 产品密钥
     * @return 签名十六进制字符串
     */
    private String sign(JSONObject params, String productSecret) {
        // 请求参数按字典顺序排序。
        Set<String> keys = getSortedKeys(params);
        // sign、signMethod除外。
        keys.remove("sign");
        keys.remove("signMethod");
        // 组装签名明文。
        StringBuffer content = new StringBuffer();
        for (String key : keys) {
            content.append(key);
            content.append(params.getString(key));
        }
        // 计算签名。
        String sign = encrypt(content.toString(), productSecret);
        System.out.println("sign content=" + content);
        System.out.println("sign result=" + sign);
        return sign;
    }
    /**
     * 获取JSON对象排序后的key集合。
     *
     * @param json 需要排序的JSON对象
     * @return 排序后的key集合
     */
    private Set<String> getSortedKeys(JSONObject json) {
        SortedMap<String, String> map = new TreeMap<String, String>();
        for (String key : json.keySet()) {
            String vlaue = json.getString(key);
            map.put(key, vlaue);
        }
        return map.keySet();
    }
    /**
     * 使用HMAC_ALGORITHM加密。
     *
     * @param content 明文
     * @param secret 密钥
     * @return 密文
     */
    private String encrypt(String content, String secret) {
        try {
            byte[] text = content.getBytes(StandardCharsets.UTF_8);
            byte[] key = secret.getBytes(StandardCharsets.UTF_8);
            SecretKeySpec secretKey = new SecretKeySpec(key, HMAC_ALGORITHM);
            Mac mac = Mac.getInstance(secretKey.getAlgorithm());
            mac.init(secretKey);
            return byte2hex(mac.doFinal(text));
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
    /**
     * 二进制转十六进制字符串。
     *
     * @param b 二进制数组
     * @return 十六进制字符串
     */
    private String byte2hex(byte[] b) {
        StringBuffer sb = new StringBuffer();
        for (int n = 0; b != null && n < b.length; n++) {
            String stmp = Integer.toHexString(b[n] & 0XFF);
            if (stmp.length() == 1) {
                sb.append('0');
            }
            sb.append(stmp);
        }
        return sb.toString().toUpperCase();
    }
    public static void main(String[] args) throws Exception {
        String productKey = "*******";
        String productSecret = "*******";
        String deviceName = "*******";
        //进行动态注册。
        DynamicRegister client = new DynamicRegister();
        client.register(productKey, productSecret, deviceName);
    }
}


执行上面的代码动态注册成功后，平台会返回{ "productKey" : "***",   "deviceName" : "***",   "deviceSecret" : "***"}，然后拿着这些参数再连接平台发送上行数据，当再次运行上面的代码时，会发现报错，报错截图：


image.png


当设备通过该三元组成功连接过平台，设备就注册成功了，之后获取设备三元组就会失败，此时需要调用ResetThing接口，重置设备的为未注册，然后设备才能再次获取到设备三元组。所以设备在成功获取到三元组后，需要存储到设备中。

ResetThing接口说明：https://help.aliyun.com/document_detail/155680.html

文章标签：
云消息队列 MQ
物联网
安全
算法
存储
关键词：
云消息队列 MQ设备
设备云消息队列 MQ
云消息队列 MQ注册
云消息队列 MQ通道
设备云消息队列 MQ通道动态注册
相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv，使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列：消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验：基于MNS，0基础轻松构建 Web Client 本节课带您一起基于MNS，0基础轻松构建 Web Client
AIoT_qiao
目录
相关文章
三分钟热度的鱼
|
2月前
|
消息中间件 网络协议 物联网
MQTT常见问题之物联网设备端申请动态注册时MQTT服务不可用如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
49 3
java知路
|
8月前
|
消息中间件 Java Linux
rocketmq linux注册服务开机启动配置
rocketmq linux注册服务开机启动配置
java知路
266 1
三分钟热度的鱼
|
2月前
|
物联网 网络性能优化 API
MQTT常见问题之MQTT获取某个时间点的设备的状态失败如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
28 1
三分钟热度的鱼
|
2月前
|
物联网 Serverless
MQTT常见问题之通过mqtt控制台查询不到设备轨迹如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
33 1
帐篷Li
|
2月前
|
监控 网络性能优化 网络安全
【MODBUS】Modbus主站为边缘设备通过MQTT协议上云
【MODBUS】Modbus主站为边缘设备通过MQTT协议上云
帐篷Li
38 1
wljslmz
|
6月前
|
传感器 JavaScript 物联网
MQTT 保持活动计时器：让您的设备保持连接
MQTT 保持活动计时器：让您的设备保持连接
wljslmz
109 0
石云升
|
12月前
|
Java 测试技术 Shell
使用JMeter模拟设备通过MQTT发送数据
使用JMeter模拟设备通过MQTT发送数据
石云升
365 0
使用JMeter模拟设备通过MQTT发送数据
开发者小达人
|
12月前
|
消息中间件 移动开发 运维
《2023云原生实战案例集》——04 互联网——小七手游 MQTT构筑运营平台与游戏端的交互通道
《2023云原生实战案例集》——04 互联网——小七手游 MQTT构筑运营平台与游戏端的交互通道
开发者小达人
166 0
三分钟热度的鱼
|
2月前
|
消息中间件 网络协议 JavaScript
MQTT常见问题之微消息队列mqtt支持ipv6失败如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
48 1
三分钟热度的鱼
|
2月前
|
消息中间件 物联网 Java
MQTT常见问题之微消息队列配置失败如何解决
MQTT（Message Queuing Telemetry Transport）是一个轻量级的、基于发布/订阅模式的消息协议，广泛用于物联网（IoT）中设备间的通信。以下是MQTT使用过程中可能遇到的一些常见问题及其答案的汇总：
三分钟热度的鱼
30 4

云服务技术课堂

热门文章

最新文章

  • 1
    使用阿里云 mqtt serverless 版本超低成本快速实现 webscoket 长链接服务器
  • 2
    RocketMQ的下载与安装（全网最细保姆级别教学）
  • 3
    阿里云MQTT简介和使用流程
  • 4
    麒麟系统ARM安装rabbitmq
  • 5
    RocketMQ学习笔记
  • 6
    阿里云服务器上搭建 MQTT服务
  • 7
    RocketMQ部署文档
  • 8
    RabbitMQ入门指南(八)：MQ可靠性
  • 9
    RabbitMQ入门指南(二)：架构和管理控制台的使用
  • 10
    RabbitMQ入门指南(九)：消费者可靠性
  • 1
    阿里云OpenAPI RocketMQ 5.0的PHP收发消息文档
    52
  • 2
    爆款游戏如何借助 RocketMQ Serverless，打造流畅体验并节省 98% 成本？
    87247
  • 3
    深度剖析 RocketMQ 5.0，事件驱动：云时代的事件驱动有啥不同？
    78954
  • 4
    深度剖析 RocketMQ 5.0，IoT 消息：物联网需要什么样的消息技术？
    90866
  • 5
    深度剖析 RocketMQ 5.0，流数据库：如何实现一体化流处理？
    78565
  • 6
    深度剖析 RocketMQ 5.0，流存储：流场景的诉求是什么？
    3372
  • 7
    深度剖析 RocketMQ 5.0，消息进阶：如何支撑复杂业务消息场景？
    108198
  • 8
    【十七】RabbitMQ基础篇（延迟队列和死信队列实战）
    45
  • 9
    【十六】RabbitMQ基础篇（下载安装并基础使用，内含各种坑问题）
    33
  • 10
    深度剖析 RocketMQ 5.0，消息基础：RocketMQ 在业务消息场景的基础优势是什么？
    125309

    • 相关课程

    更多
  • 消息队列 RocketMQ 消息集成
  • 消息队列 MNS 入门课程
  • RocketMQ知识精讲与项目实战（第一阶段）
  • RocketMQ知识精讲与项目实战（第二阶段）
  • RocketMQ知识精讲与项目实战（第三阶段）
  • 微服务实战-RocketMQ Binder

    • 相关电子书

    更多
  • RocketMQ Client-GO 介绍
  • RocketMQ Prometheus Exporter 打造定制化 DevOps 平台
  • 基于 RocketMQ Prometheus Exporter 打造定制化 DevOps 平台

    • 相关实验场景

    更多
  • 钉钉群中如何接收IoT温控器数据告警通知
  • 部署RabbitMQ
  • RocketMQ运维自我实践
  • RocketMQ的常规运维实践应用
  • RocketMQ中使用Java客户端发送消息和消费的应用
  • RocketMQ一站式入门使用
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）