本文主要介绍如何使用 MQTT 客户端收发 MQTT 消息,并给出示例代码供前期开发测试参考,包括资源创建、环境准备、示例代码、注意事项等。
注意:
本文给出的实例均基于 Eclipse Paho Java SDK 实现,SDK 下载请参见 MQTT 接入准备。如使用其他第三方的客户端,请适当修改。
使用 MQ 提供的 MQTT 服务,首先需要核实应用中使用的 Topic 资源是否已经创建,如果没有,请先去控制台创建 Topic,Group ID 等资源。
创建资源时需要根据需求选择对应的 Region,例如 MQTT 需要使用华北2的接入点,那么 Topic 等资源就在华北2 创建,资源创建具体请参见创建资源。
注意:MQTT 使用的多级子 Topic 不需要创建,代码里直接使用即可,没有限制。
使用 MQTT 协议来收发消息,需要根据应用平台选择合适的客户端。本示例运行在 Java 平台,使用 Eclipse Paho Java SDK 构建。首先引入 Maven 依赖,POM 文件配置如下:
<dependencies>
<dependency>
<groupId>org.eclipse.paho</groupId>
<artifactId>org.eclipse.paho.client.mqttv3</artifactId>
<version>1.2.0</version>
</dependency>
<dependency>
<groupId>commons-codec</groupId>
<artifactId>commons-codec</artifactId>
<version>1.10</version>
</dependency>
</dependencies>
<repositories>
<repository>
<id>Eclipse Paho Repo</id>
<url>https://repo.eclipse.org/content/repositories/paho-releases/</url>
</repository>
<repository>
<id>snapshots-repo</id>
<url>https://oss.sonatype.org/content/repositories/snapshots</url>
<releases>
<enabled>false</enabled>
</releases>
<snapshots>
<enabled>true</enabled>
</snapshots>
</repository>
</repositories>
本段示例代码演示如何使用 MQTT 客户端发送普通消息和 P2P 的点对点消息,其中用到的工具 MacSignature 参考下文。
扫描二维码关注公众号,回复:
3327687 查看本文章
public class MQTTSendMsg {
public static void main(String[] args) throws IOException {
/**
* 设置当前用户私有的 MQTT 的接入点。例如此处示意使用 XXX,实际使用请替换用户自己的接入点。接入点的获取方法是,在控制台创建 MQTT 实例,每个实例都会分配一个接入点域名。
*/
final String broker ="tcp://XXXX.mqtt.aliyuncs.com:1883";
/**
* 设置阿里云的 AccessKey,用于鉴权
*/
final String acessKey ="XXXXXX";
/**
* 设置阿里云的 SecretKey,用于鉴权
*/
final String secretKey ="XXXXXXX";
/**
* 发消息使用的一级 Topic,需要先在 MQ 控制台里创建
*/
final String topic ="XXXX";
/**
* MQTT 的 ClientID,一般由两部分组成,GroupID@@@DeviceID
* 其中 GroupID 在 MQ 控制台里创建
* DeviceID 由应用方设置,可能是设备编号等,需要唯一,否则服务端拒绝重复的 ClientID 连接
*/
final String clientId ="GID_XXX@@@ClientID_XXXX";
String sign;
MemoryPersistence persistence = new MemoryPersistence();
try {
final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
final MqttConnectOptions connOpts = new MqttConnectOptions();
System.out.println("Connecting to broker: " + broker);
/**
* 计算签名,将签名作为 MQTT 的 password。
* 签名的计算方法,参考工具类 MacSignature,第一个参数是 ClientID 的前半部分,即 GroupID
* 第二个参数阿里云的 SecretKey
*/
sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
connOpts.setUserName(acessKey);
connOpts.setServerURIs(new String[] { broker });
connOpts.setPassword(sign.toCharArray());
connOpts.setCleanSession(true);
connOpts.setKeepAliveInterval(90);
connOpts.setAutomaticReconnect(true);
connOpts.setMqttVersion(MQTT_VERSION_3_1_1);
sampleClient.setCallback(new MqttCallbackExtended() {
public void connectComplete(boolean reconnect, String serverURI) {
System.out.println("connect success");
//连接成功,需要上传客户端所有的订阅关系
}
public void connectionLost(Throwable throwable) {
System.out.println("mqtt connection lost");
}
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
}
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
}
});
sampleClient.connect(connOpts);
for (int i = 0; i < 10; i++) {
try {
String scontent = new Date()+"MQTT Test body" + i;
//此处消息体只需要传入 byte 数组即可,对于其他类型的消息,请自行完成二进制数据的转换
final MqttMessage message = new MqttMessage(scontent.getBytes());
message.setQos(0);
System.out.println(i+" pushed at "+new Date()+" "+ scontent);
/**
*消息发送到某个主题 Topic,所有订阅这个 Topic 的设备都能收到这个消息。
* 遵循 MQTT 的发布订阅规范,Topic 也可以是多级 Topic。此处设置了发送到二级 Topic
*/
sampleClient.publish(topic+"/notice/", message);
/**
* 如果发送 P2P 消息,二级 Topic 必须是“p2p”,三级 Topic 是目标的 ClientID
* 此处设置的三级 Topic 需要是接收方的 ClientID
*/
String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
sampleClient.publish(p2pTopic,message);
} catch (Exception e) {
e.printStackTrace();
}
}
} catch (Exception me) {
me.printStackTrace();
}
}
}
本段代码演示如何使用 MQTT 客户端订阅消息,接收普通的消息以及点对点消息。
public class MQTTRecvMsg {
public static void main(String[] args) throws IOException {
/**
* 设置当前用户私有的 MQTT 的接入点。例如此处示意使用 XXX,实际使用请替换用户自己的接入点。接入点的获取方法是,在控制台创建 MQTT 实例,每个实例都会分配一个接入点域名。
*/
final String broker ="tcp://XXXX.mqtt.aliyuncs.com:1883";
/**
* 设置阿里云的 AccessKey,用于鉴权
*/
final String acessKey ="XXXXXX";
/**
* 设置阿里云的 SecretKey,用于鉴权
*/
final String secretKey ="XXXXXXX";
/**
* 发消息使用的一级 Topic,需要先在 MQ 控制台里创建
*/
final String topic ="XXXX";
/**
* MQTT 的 ClientID,一般由两部分组成,GroupID@@@DeviceID
* 其中 GroupID 在 MQ 控制台里创建
* DeviceID 由应用方设置,可能是设备编号等,需要唯一,否则服务端拒绝重复的 ClientID 连接
*/
final String clientId ="GID_XXXX@@@ClientID_XXXXXX";
String sign;
MemoryPersistence persistence = new MemoryPersistence();
try {
final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
final MqttConnectOptions connOpts = new MqttConnectOptions();
System.out.println("Connecting to broker: " + broker);
/**
* 计算签名,将签名作为 MQTT 的 password
* 签名的计算方法,参考工具类 MacSignature,第一个参数是 ClientID 的前半部分,即 GroupID
* 第二个参数阿里云的 SecretKey
*/
sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
/**
* 设置订阅方订阅的 Topic 集合,此处遵循 MQTT 的订阅规则,可以是一级 Topic,二级 Topic,P2P 消息请订阅/p2p
*/
final String[] topicFilters=new String[]{topic+"/notice/",topic+"/p2p"};
final int[]qos={0,0};
connOpts.setUserName(acessKey);
connOpts.setServerURIs(new String[] { broker });
connOpts.setPassword(sign.toCharArray());
connOpts.setCleanSession(true);
connOpts.setKeepAliveInterval(90);
connOpts.setAutomaticReconnect(true);
sampleClient.setCallback(new MqttCallbackExtended() {
public void connectComplete(boolean reconnect, String serverURI) {
System.out.println("connect success");
//连接成功,需要上传客户端所有的订阅关系
sampleClient.subscribe(topicFilters,qos);
}
public void connectionLost(Throwable throwable) {
System.out.println("mqtt connection lost");
}
public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
}
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
}
});
//客户端每次上线都必须上传自己所有涉及的订阅关系,否则可能会导致消息接收延迟
sampleClient.connect(connOpts);
//每个客户端最多允许存在30个订阅关系,超出限制可能会丢弃导致收不到部分消息
sampleClient.subscribe(topicFilters,qos);
Thread.sleep(Integer.MAX_VALUE);
} catch (Exception me) {
me.printStackTrace();
}
}
}
上文代码用到的工具类 MacSignature.java 如下:
public class MacSignature {
/**
* @param text 要签名的文本
* @param secretKey 阿里云 MQ SecretKey
* @return 加密后的字符串
* @throws InvalidKeyException
* @throws NoSuchAlgorithmException
*/
public static String macSignature(String text, String secretKey) throws InvalidKeyException, NoSuchAlgorithmException {
Charset charset = Charset.forName("UTF-8");
String algorithm = "HmacSHA1";
Mac mac = Mac.getInstance(algorithm);
mac.init(new SecretKeySpec(secretKey.getBytes(charset), algorithm));
byte[] bytes = mac.doFinal(text.getBytes(charset));
return new String(Base64.encodeBase64(bytes), charset);
}
/**
* 发送方签名方法
*
* @param clientId MQTT ClientID
* @param secretKey 阿里云 MQ SecretKey
* @return 加密后的字符串
* @throws NoSuchAlgorithmException
* @throws InvalidKeyException
*/
public static String publishSignature(String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
return macSignature(clientId, secretKey);
}
/**
* 订阅方签名方法
*
* @param topics 要订阅的 Topic 集合
* @param clientId MQTT ClientID
* @param secretKey 阿里云 MQ SecretKey
* @return 加密后的字符串
* @throws NoSuchAlgorithmException
* @throws InvalidKeyException
*/
public static String subSignature(List<String> topics, String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
Collections.sort(topics); //以字典顺序排序
String topicText = "";
for (String topic : topics) {
topicText += topic + "\n";
}
String text = topicText + clientId;
return macSignature(text, secretKey);
}
/**
* 订阅方签名方法
*
* @param topic 要订阅的 Topic
* @param clientId MQTT ClientID
* @param secretKey 阿里云 MQ SecretKey
* @return 加密后的字符串
* @throws NoSuchAlgorithmException
* @throws InvalidKeyException
*/
public static String subSignature(String topic, String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
List<String> topics = new ArrayList<String>();
topics.add(topic);
return subSignature(topics, clientId, secretKey);
}
}