作者:俏巴
概述
MQTT协议是基于PUB/SUB的异步通信模式,不适用于服务端同步控制设备端返回结果的场景。前面博客基于开源Java MQTT Client的阿里云物联网平台RRPC功能测试介绍了RRPC原理,使用Open API实现服务端的调用实现。本文介绍如何使用NET SDK在本地实现RRPC的服务端调用。
Step By Step
1、创建产品与设备
参考链接 准备工作部分。
2、设备端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 IoTRRPCForSysTopicDemo {
<span class="hljs-comment">// 设备三元组信息</span>
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> productKey = <span class="hljs-string">"a16MX******"</span>;
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> deviceName = <span class="hljs-string">"RRPC******"</span>;
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> deviceSecret = <span class="hljs-string">"Ayq069ifWO7WOmwKyswNLv6E********"</span>;
public <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> regionId = <span class="hljs-string">"cn-shanghai"</span>;
<span class="hljs-comment">// RRPC 系统Topic</span>
private <span class="hljs-keyword">static</span> <span class="hljs-built_in">String</span> subTopic = <span class="hljs-string">"/sys/"</span> + productKey + <span class="hljs-string">"/"</span> + deviceName+ <span class="hljs-string">"/rrpc/request/+"</span>;
private <span class="hljs-keyword">static</span> MqttClient mqttClient;
public <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> main(<span class="hljs-built_in">String</span> [] args) {
initAliyunIoTClient();
<span class="hljs-comment">// RRPC订阅Topic</span>
<span class="hljs-keyword">try</span> {
mqttClient.subscribe(subTopic);
} <span class="hljs-keyword">catch</span> (MqttException e) {
e.printStackTrace();
}
mqttClient.setCallback(<span class="hljs-keyword">new</span> MqttCallback() {
@Override
public <span class="hljs-keyword">void</span> connectionLost(Throwable cause) {
System.out.println(<span class="hljs-string">"connectionLost:"</span> + cause.getMessage());
}
@Override
public <span class="hljs-keyword">void</span> messageArrived(<span class="hljs-built_in">String</span> topic, MqttMessage message) throws Exception {
System.out.println(<span class="hljs-string">"message: "</span> + <span class="hljs-keyword">new</span> <span class="hljs-built_in">String</span>(message.getPayload()));
System.out.println(<span class="hljs-string">"topic:"</span> + topic);
<span class="hljs-comment">// 根据RRPC请求消息Topic,构建RRPC响应消息Topic</span>
<span class="hljs-built_in">String</span> responseTopic = topic.replace(<span class="hljs-string">"request"</span>,<span class="hljs-string">"response"</span>);
MqttMessage message1 = <span class="hljs-keyword">new</span> MqttMessage(<span class="hljs-string">"resonse demo"</span>.getBytes(<span class="hljs-string">"utf-8"</span>));
mqttClient.publish(responseTopic, message1);
}
@Override
public <span class="hljs-keyword">void</span> deliveryComplete(IMqttDeliveryToken token) {
}
});
}
<span class="hljs-comment">/***
* 初始化Client
*/</span>
private <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> initAliyunIoTClient() {
<span class="hljs-keyword">try</span> {
<span class="hljs-built_in">String</span> clientId = <span class="hljs-string">"java"</span> + System.currentTimeMillis();
<span class="hljs-built_in">Map</span><<span class="hljs-built_in">String</span>, <span class="hljs-built_in">String</span>> params = <span class="hljs-keyword">new</span> HashMap<>(<span class="hljs-number">16</span>);
params.put(<span class="hljs-string">"productKey"</span>, productKey);
params.put(<span class="hljs-string">"deviceName"</span>, deviceName);
params.put(<span class="hljs-string">"clientId"</span>, clientId);
<span class="hljs-built_in">String</span> timestamp = <span class="hljs-built_in">String</span>.valueOf(System.currentTimeMillis());
params.put(<span class="hljs-string">"timestamp"</span>, timestamp);
<span class="hljs-comment">// cn-shanghai</span>
<span class="hljs-built_in">String</span> targetServer = <span class="hljs-string">"tcp://"</span> + productKey + <span class="hljs-string">".iot-as-mqtt."</span> + regionId + <span class="hljs-string">".aliyuncs.com:1883"</span>;
<span class="hljs-built_in">String</span> mqttclientId = clientId + <span class="hljs-string">"|securemode=3,signmethod=hmacsha1,timestamp="</span> + timestamp + <span class="hljs-string">"|"</span>;
<span class="hljs-built_in">String</span> mqttUsername = deviceName + <span class="hljs-string">"&"</span> + productKey;
<span class="hljs-built_in">String</span> mqttPassword = AliyunIoTSignUtil.sign(params, deviceSecret, <span class="hljs-string">"hmacsha1"</span>);
connectMqtt(targetServer, mqttclientId, mqttUsername, mqttPassword);
} <span class="hljs-keyword">catch</span> (Exception e) {
System.out.println(<span class="hljs-string">"initAliyunIoTClient error "</span> + e.getMessage());
}
}
public <span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> connectMqtt(<span class="hljs-built_in">String</span> url, <span class="hljs-built_in">String</span> clientId, <span class="hljs-built_in">String</span> mqttUsername, <span class="hljs-built_in">String</span> mqttPassword) throws Exception {
MemoryPersistence persistence = <span class="hljs-keyword">new</span> MemoryPersistence();
mqttClient = <span class="hljs-keyword">new</span> MqttClient(url, clientId, persistence);
MqttConnectOptions connOpts = <span class="hljs-keyword">new</span> MqttConnectOptions();
<span class="hljs-comment">// MQTT 3.1.1</span>
connOpts.setMqttVersion(<span class="hljs-number">4</span>);
connOpts.setAutomaticReconnect(<span class="hljs-literal">false</span>);
connOpts.setCleanSession(<span class="hljs-literal">true</span>);
connOpts.setUserName(mqttUsername);
connOpts.setPassword(mqttPassword.toCharArray());
connOpts.setKeepAliveInterval(<span class="hljs-number">60</span>);
mqttClient.connect(connOpts);
}
}
3、云端Code
- 3.1 SDK安装:
Install-Package aliyun-net-sdk-iot -Version 7.4.0
- 3.2 NET Code
using System;
using System.Text;
using Aliyun.Acs.Core;
using Aliyun.Acs.Core.Exceptions;
using Aliyun.Acs.Core.Profile;
using Aliyun.Acs.Iot.Model.V20180120;
namespace IoTRRPCDemo
{
<span class="hljs-class"><span class="hljs-keyword">class</span> <span class="hljs-title">Program</span>
</span>{
<span class="hljs-keyword">static</span> <span class="hljs-keyword">void</span> Main(string[] args)
{
<span class="hljs-comment">// ak,sk https://yq.aliyun.com/articles/693979</span>
IClientProfile profile = DefaultProfile.GetProfile(<span class="hljs-string">"cn-shanghai"</span>, <span class="hljs-string">"LTAIOZZg********"</span>, <span class="hljs-string">"v7CjUJCMk7j9aKduMAQLjy********"</span>);
DefaultAcsClient client = <span class="hljs-keyword">new</span> DefaultAcsClient(profile);
client.SetConnectTimeoutInMilliSeconds(<span class="hljs-number">8000</span>);<span class="hljs-comment">// 设置 Client 连接超时时间,默认是5S,RRPC响应超过5S是必须设置</span>
<span class="hljs-keyword">var</span> request = <span class="hljs-keyword">new</span> RRpcRequest();
request.Timeout = <span class="hljs-number">8000</span>;
<span class="hljs-comment">// 设置设备三元组信息</span>
request.DeviceName = <span class="hljs-string">"RRPC******"</span>;
request.RequestBase64Byte = <span class="hljs-string">"5raI5oGv5LiL6KGM5rWL6K+V"</span>; <span class="hljs-comment">// 消息下行测试 base64编码内容</span>
request.ProductKey = <span class="hljs-string">"a16MX******"</span>;
<span class="hljs-keyword">try</span>
{
<span class="hljs-keyword">var</span> response = client.GetAcsResponse(request);
Console.WriteLine(Encoding.Default.GetString(response.HttpResponse.Content));
Console.ReadKey();
}
<span class="hljs-keyword">catch</span> (ServerException e)
{
Console.WriteLine(<span class="hljs-string">"ServerException:"</span>);
Console.WriteLine(e);
Console.ReadKey();
}
<span class="hljs-keyword">catch</span> (ClientException e)
{
Console.WriteLine(<span class="hljs-string">"ClientException: "</span>);
Console.WriteLine(e);
Console.ReadKey();
}
}
}
}
4、测试情况
- 4.1 设备端
- 4.2 云端