基于开源MQTT自主接入阿里云IoT平台(Java)

简介:

本文由 GXIC 作者 wongxmig 完成,欢迎关注 IoT 开发者社区

1. 准备工作

1.1 注册阿里云账号

使用个人淘宝账号或手机号,开通阿里云账号,并通过__实名认证(可以用支付宝认证)__

1.2 免费开通IoT物联网套件

产品官网 https://www.aliyun.com/product/iot

Screen Shot 2018-06-01 at 13.53.55.png | center | 569x357

1.3 软件环境

JDK安装
编辑器 IDEA

2. 开发步骤

2.1 云端开发

1) 创建高级版产品

image.png | left | 747x253

2) 功能定义,产品物模型添加属性

添加产品属性定义

属性名 标识符 数据类型 范围
温度 temperature float -50~100
湿度 humidity float 0~100

image.png | left | 747x186

物模型对应属性上报topic

/sys/替换为productKey/替换为deviceName/thing/event/property/post

物模型对应的属性上报payload

{
    id: 123452452,
    params: {
        temperature: 26.2,
        humidity: 60.4
    },
    method: "thing.event.property.post"
}

3) 设备管理>注册设备,获得身份三元组

image.png | left | 747x188

2.2 设备端开发

我们以java程序来模拟设备,建立连接,上报数据。

1) pom.xml添加sdk依赖

<dependencies>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.1.0</version>
        </dependency>
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>23.0</version>
        </dependency>
        
</dependencies>

2) AliyunIoTSignUtil工具类


import javax.crypto.Mac;
import javax.crypto.SecretKey;
import javax.crypto.spec.SecretKeySpec;
import java.util.Arrays;
import java.util.Map;

/**
 * AliyunIoTSignUtil
 */

public class AliyunIoTSignUtil {
    public static String sign(Map<String, String> params, String deviceSecret, String signMethod) {
        //将参数Key按字典顺序排序
        String[] sortedKeys = params.keySet().toArray(new String[] {});
        Arrays.sort(sortedKeys);

        //生成规范化请求字符串
        StringBuilder canonicalizedQueryString = new StringBuilder();
        for (String key : sortedKeys) {
            if ("sign".equalsIgnoreCase(key)) {
                continue;
            }
            canonicalizedQueryString.append(key).append(params.get(key));
        }

        try {
            String key = deviceSecret;
            return encryptHMAC(signMethod,canonicalizedQueryString.toString(), key);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * HMACSHA1加密
     *
     */
    public static String encryptHMAC(String signMethod,String content, String key) throws Exception {
        SecretKey secretKey = new SecretKeySpec(key.getBytes("utf-8"), signMethod);
        Mac mac = Mac.getInstance(secretKey.getAlgorithm());
        mac.init(secretKey);
        byte[] data = mac.doFinal(content.getBytes("utf-8"));
        return bytesToHexString(data);
    }

    public static final String bytesToHexString(byte[] bArray) {

        StringBuffer sb = new StringBuffer(bArray.length);
        String sTemp;
        for (int i = 0; i < bArray.length; i++) {
            sTemp = Integer.toHexString(0xFF & bArray[i]);
            if (sTemp.length() < 2) {
                sb.append(0);
            }
            sb.append(sTemp.toUpperCase());
        }
        return sb.toString();
    }

}

3) 模拟设备IoTDemo.java代码

public class IoTDemo {

    public static String productKey = "替换productKey";
    public static String deviceName = "替换deviceName";
    public static String deviceSecret = "替换deviceSecret";
    public static String regionId = "cn-shanghai";

    //物模型-属性上报topic
    private static String pubTopic = "/sys/" + productKey + "/" + deviceName + "/thing/event/property/post";
    //高级版 物模型-属性上报payload
    private static final String payloadJson =
            "{" +
            "    \"id\": %s," +
            "    \"params\": {" +
            "        \"temperature\": %s," +
            "        \"humidity\": %s" +
            "    }," +
            "    \"method\": \"thing.event.property.post\"" +
            "}";

    private static MqttClient mqttClient;
    private static Random random = new Random();
    private static DecimalFormat df = new DecimalFormat("0.#");

    public static void main(String [] args) {

        initAliyunIoTClient();

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

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

    }

    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://" + productKey + ".iot-as-mqtt."+regionId+".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.setCleanSession(true);

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

        mqttClient.connect(connOpts);

    }


    private static void postDeviceProperties() {

        try {

            //上报数据
            String payload = String.format(payloadJson, System.currentTimeMillis(), df.format(25+random.nextFloat()*10), df.format(50+random.nextFloat()*30));

            System.out.println("post :"+payload);

            MqttMessage message = new MqttMessage(payload.getBytes("utf-8"));
            message.setQos(1);

            mqttClient.publish(pubTopic, message);

        } catch (Exception e) {

        }

    }


}

3. 启动运行

3.1 设备启动

Run IoTDemo.main

3.2 云端查看设备运行状态

image.png | left | 602x225

4. 项目源码

download: therometer.zip

download: aliyun-iot-demo-java.zip


<a class="anchor" id="" href="#80ifpi"></a>


相关实践学习
5分钟轻松打造应对流量洪峰的稳定商城交易系统
本实验通过SAE极速部署一个微服务电商商城,同时结合RocketMQ异步解耦、削峰填谷的能力,带大家体验面对流量洪峰仍旧稳定可靠的商城交易系统!
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
目录
相关文章
|
21天前
|
物联网
(手把手)在华为云、阿里云搭建自己的物联网MQTT消息服务器,免费IOT平台
本文介绍如何在阿里云搭建自己的物联网MQTT消息服务器,并使用 “MQTT客户端调试工具”模拟MQTT设备,接入平台进行消息收发。
527 31
|
25天前
|
物联网
如何在腾讯云等平台搭建自己的物联网MQTT服务器Broker
物联网技术及MQTT协议被广泛应用于各种场景。本文介绍物联网MQTT服务助手下载,如何搭建自己的物联网平台,并使用 “MQTT客户端调试工具”模拟MQTT设备,接入平台进行消息收发。
184 11
|
6月前
|
消息中间件 运维 安全
C5GAME 游戏饰品交易平台借助 RocketMQ Serverless 保障千万级玩家流畅体验
游戏行业蓬勃发展,作为国内领先的 STEAM 游戏饰品交易的服务平台,看 C5GAME 如何利用 RocketMQ Serverless 技术,为千万级玩家提供流畅的游戏体验,同时降低成本并提升运维效率。
406 89
C5GAME 游戏饰品交易平台借助 RocketMQ Serverless 保障千万级玩家流畅体验
|
8月前
|
消息中间件 存储 Serverless
【实践】快速学会使用阿里云消息队列RabbitMQ版
云消息队列 RabbitMQ 版是一款基于高可用分布式存储架构实现的 AMQP 0-9-1协议的消息产品。云消息队列 RabbitMQ 版兼容开源 RabbitMQ 客户端,解决开源各种稳定性痛点(例如消息堆积、脑裂等问题),同时具备高并发、分布式、灵活扩缩容等云消息服务优势。
239 2
|
9月前
|
XML 网络协议 物联网
基于surging的木舟IOT平台如何添加网络组件
【8月更文挑战第30天】在基于 Surging 的木舟 IOT 平台中添加网络组件需经历八个步骤:首先理解 Surging 及平台架构;其次明确组件需求,选择合适技术库;接着创建项目并配置;然后设计实现网络功能;再将组件集成至平台;接着进行详尽测试;最后根据反馈持续优化与维护。具体实施时应参照最新文档调整。
146 11
|
7月前
|
消息中间件 存储 Java
RocketMQ文件刷盘机制深度解析与Java模拟实现
【11月更文挑战第22天】在现代分布式系统中,消息队列(Message Queue, MQ)作为一种重要的中间件,扮演着连接不同服务、实现异步通信和消息解耦的关键角色。Apache RocketMQ作为一款高性能的分布式消息中间件,广泛应用于实时数据流处理、日志流处理等场景。为了保证消息的可靠性,RocketMQ引入了一种称为“刷盘”的机制,将消息从内存写入到磁盘中,确保消息持久化。本文将从底层原理、业务场景、概念、功能点等方面深入解析RocketMQ的文件刷盘机制,并使用Java模拟实现类似的功能。
160 3
|
8月前
|
存储 消息中间件 安全
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
【10月更文挑战第9天】本文介绍了如何利用JUC组件实现Java服务与硬件通过MQTT的同步通信(RRPC)。通过模拟MQTT通信流程,使用`LinkedBlockingQueue`作为消息队列,详细讲解了消息发送、接收及响应的同步处理机制,包括任务超时处理和内存泄漏的预防措施。文中还提供了具体的类设计和方法实现,帮助理解同步通信的内部工作原理。
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
|
8月前
|
消息中间件 存储 JSON
rabbitmq基础教程(ui,java,springamqp)
本文提供了RabbitMQ的基础教程,包括如何使用UI创建队列和交换机、Java代码操作RabbitMQ、Spring AMQP进行消息发送和接收,以及如何使用不同的交换机类型(fanout、direct、topic)进行消息路由。
94 0
rabbitmq基础教程(ui,java,springamqp)
|
9月前
|
消息中间件 弹性计算 运维
阿里云云消息队列RabbitMQ实践解决方案评测报告
阿里云云消息队列RabbitMQ实践解决方案评测报告
162 9
|
8月前
|
消息中间件 前端开发 Java
java高并发场景RabbitMQ的使用
java高并发场景RabbitMQ的使用
229 0