作者：俏巴
概述
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、设备端订阅情况