基于开源Java MQTT Client的阿里云物联网平台RRPC功能测试-阿里云开发者社区

开发者社区> Java技术进阶> 正文

基于开源Java MQTT Client的阿里云物联网平台RRPC功能测试

简介: MQTT协议是基于PUB/SUB的异步通信模式,不适用于服务端同步控制设备端返回结果的场景。物联网平台基于MQTT协议制定了一套请求和响应的同步机制,无需改动MQTT协议即可实现同步通信。物联网平台提供API给服务端,设备端只需要按照固定的格式回复PUB消息,服务端使用API,即可同步获取设备端的响应结果。

概述

MQTT协议是基于PUB/SUB的异步通信模式,不适用于服务端同步控制设备端返回结果的场景。物联网平台基于MQTT协议制定了一套请求和响应的同步机制,无需改动MQTT协议即可实现同步通信。物联网平台提供API给服务端,设备端只需要按照固定的格式回复PUB消息,服务端使用API,即可同步获取设备端的响应结果。RRPC:Revert-RPC。RPC(Remote Procedure Call)采用客户机/服务器模式,用户不需要了解底层技术协议,即可远程请求服务。RRPC则可以实现由服务端请求设备端并能够使设备端响应的功能。本文主要基于开源Java MQTT Client,分别针对系统Topic和自定义Topic,演示阿里云物联网平台RRPC的实现。

RRPC原理

_

  • 1、物联网平台收到来自用户服务器的RRPC调用,下发一条RRPC请求消息给设备。消息体为用户传入的数据,Topic为物联网平台定义的Topic,其中含有唯一的RRPC消息ID。
  • 2、设备收到下行消息后,按照指定Topic格式(包含之前云端下发的唯一的RRPC消息ID)回复一条RRPC响应消息给云端,云端提取出Topic中的消息ID,和之前的RRPC请求消息匹配上,然后回复给用户服务器。
  • 3、如果调用时设备不在线,云端会给用户服务器返回设备离线的错误;如果设备没有在超时时间内(8秒内)回复RRPC响应消息,云端会给用户服务器返回超时错误。

更多原理介绍可以参考阿里云官方文档


实验测试

准备工作

1、创建产品和设备

参考:阿里云物联网平台Qucik Start 创建产品和设备部分。

2、创建自定义Topic

_

3、开源SDK的使用

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

4、服务端RRpc API的调用:调用该接口向指定设备发送请求消息,并同步返回响应。

这里我们使用Open API Explorer快速测试调用,RRPC测试地址

_


系统Topic测试

参数:

RRPC调用的系统Topic格式如下:

  • RRPC请求消息Topic:/sys/${YourProductKey}/${YourDeviceName}/rrpc/request/${messageId}
  • RRPC响应消息Topic:/sys/${YourProductKey}/${YourDeviceName}/rrpc/response/${messageId}
  • RRPC订阅Topic:/sys/${YourProductKey}/${YourDeviceName}/rrpc/request/+

1、设备端Code

import com.alibaba.taro.AliyunIoTSignUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.util.HashMap;
import java.util.Map;

public class IoTRRPCForSysTopic {

    // 设备三元组信息
    public static String productKey = "******";
    public static String deviceName = "******";
    public static String deviceSecret = "********";
    public static String regionId = "cn-shanghai";

    // RRPC 系统Topic
    private static String subTopic = "/sys/" + productKey + "/" + deviceName+ "/rrpc/request/+";
    private static MqttClient mqttClient;

    public static void main(String [] args) {

        initAliyunIoTClient();

        // RRPC订阅Topic
        try {
            mqttClient.subscribe(subTopic);
        } catch (MqttException e) {
            e.printStackTrace();
        }

        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("connectionLost:" + cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("message: " + new String(message.getPayload()));

                // 根据RRPC请求消息Topic,构建RRPC响应消息Topic
                String responseTopic = topic.replace("request","response");
                MqttMessage message1 = new MqttMessage("resonse demo".getBytes("utf-8"));
                mqttClient.publish(responseTopic,message1);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
//                System.out.println("IMqttDeliveryToken: " + token);
            }
        });
    }

    /***
     * 初始化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://" + 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);
    }
}

2、服务端下发消息

_

3、设备端订阅情况

_


自定义Topic测试

参数

RRPC调用自定义Topic的格式如下:

  • RRPC请求消息Topic:/ext/rrpc/${messageId}/${topic}
  • RRPC响应消息Topic:/ext/rrpc/${messageId}/${topic}
  • RRPC订阅Topic:/ext/rrpc/+/${topic}

1、设备端Code

import com.alibaba.taro.AliyunIoTSignUtil;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import java.util.HashMap;
import java.util.Map;

public class IoTRRPCForPersonalTopic {

    // 设备三元组信息
    public static String productKey = "******";
    public static String deviceName = "******";
    public static String deviceSecret = "********";
    public static String regionId = "cn-shanghai";

    // RRPC 自定义Topic
    private static String personalTopic = "/" + productKey + "/" + deviceName + "/user/rrpcdemo";
    private static String subTopic = "/ext/rrpc/+" + personalTopic;
    private static MqttClient mqttClient;

    public static void main(String [] args){

        initAliyunIoTClient();

        try {
            mqttClient.subscribe(subTopic);
        } catch (MqttException e) {
            e.printStackTrace();

            System.out.println(e.getMessage());
        }

        mqttClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                System.out.println("connectionLost");
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("topic: " + topic);
                System.out.println("message: " + new String(message.getPayload()));

                // 设置响应 Topic 及 message
                MqttMessage msg = new MqttMessage("demo".getBytes());
                mqttClient.publish(topic,msg);
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
//                System.out.println("IMqttDeliveryToken: " + token);
            }
        });
    }

    /***
     * 初始化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://" + productKey + ".iot-as-mqtt."+regionId+".aliyuncs.com:1883";

            // 从云端下发自定义格式Topic的RRPC调用命令到设备端时,设备端必须在进行MQTT CONNECT协议设置时,在clientId中增加ext=1参数。
            String mqttclientId = clientId + "|securemode=3,signmethod=hmacsha1,timestamp=" + timestamp + ",ext=1|";
            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);
    }
}

注意

从云端下发自定义格式Topic的RRPC调用命令到设备端时,设备端必须在进行MQTT CONNECT协议设置时,在clientId中增加ext=1参数。

2、服务端下发消息

_

3、设备端订阅情况

_


参考链接

调用系统Topic
调用自定义Topic

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
Java技术进阶
使用钉钉扫一扫加入圈子
+ 订阅

Java技术进阶成长,课程资料,案例解析,实战经验全都有!

官网链接