MQTT 客户端收发 MQTT 消息（2）

3. MQTT 发送消息


本段示例代码演示如何使用 MQTT 客户端发送普通消息和 P2P 的点对点消息，其中用到的工具 MacSignature 参考下文。

1. public class MQTTSendMsg {
2.     public static void main(String[] args) throws IOException {
3.        /**
4.          * 设置当前用户私有的MQTT的接入点。例如此处示意使用XXX，实际使用请替换用户自己的接入点。接入点的获取方法是，在控制台申请MQTT实例，每个实例都会分配一个接入点域名。
5.          */
6.         final String broker ="tcp://XXXX.mqtt.aliyuncs.com:1883";
7.         /**
8.          * 设置阿里云的AccessKey，用于鉴权
9.          */
10.         final String acessKey ="XXXXXX";
11.         /**
12.          * 设置阿里云的SecretKey，用于鉴权
13.          */
14.         final String secretKey ="XXXXXXX";
15.         /**
16.          * 发消息使用的一级Topic，需要先在MQ控制台里申请
17.          */
18.         final String topic ="XXXX";
20.         /**
21.          * MQTT的ClientID，一般由两部分组成，GroupID@@@DeviceID
22.          * 其中GroupID在MQ控制台里申请
23.          * DeviceID由应用方设置，可能是设备编号等，需要唯一，否则服务端拒绝重复的ClientID连接
24.          */
25.         final String clientId ="GID_XXX@@@ClientID_XXXX";
26.         String sign;
27.         MemoryPersistence persistence = new MemoryPersistence();
28.         try {
29.             final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
30.             final MqttConnectOptions connOpts = new MqttConnectOptions();
31.             System.out.println("Connecting to broker: " + broker);
32.             /**
33.              * 计算签名，将签名作为MQTT的password。
34.              * 签名的计算方法，参考工具类MacSignature，第一个参数是ClientID的前半部分，即GroupID
35.              * 第二个参数阿里云的SecretKey
36.              */
37.             sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
38.             connOpts.setUserName(acessKey);
39.             connOpts.setServerURIs(new String[] { broker });
40.             connOpts.setPassword(sign.toCharArray());
41.             connOpts.setCleanSession(true);
42.             connOpts.setKeepAliveInterval(90);
43.             connOpts.setAutomaticReconnect(true);
44.             sampleClient.setCallback(new MqttCallbackExtended() {
45.                 public void connectComplete(boolean reconnect, String serverURI) {
46.                     System.out.println("connect success");
47.                     //连接成功，需要上传客户端所有的订阅关系
48.                 }
49.                 public void connectionLost(Throwable throwable) {
50.                     System.out.println("mqtt connection lost");
51.                 }
52.                 public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
53.                     System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
54.                 }
55.                 public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
56.                     System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
57.                 }
58.             });
59.             sampleClient.connect(connOpts);
60.             for (int i = 0; i < 10; i++) {
61.                 try {
62.                     String scontent = new Date()+"MQTT Test body" + i;
63.                     //此处消息体只需要传入byte数组即可，对于其他类型的消息，请自行完成二进制数据的转换
64.                     final MqttMessage message = new MqttMessage(scontent.getBytes());
65.                     message.setQos(0);
66.                     System.out.println(i+" pushed at "+new Date()+" "+ scontent);
67.                     /**
68.                      *消息发送到某个主题Topic，所有订阅这个Topic的设备都能收到这个消息。
69.                      * 遵循MQTT的发布订阅规范，Topic也可以是多级Topic。此处设置了发送到二级Topic
70.                      */
71.                     sampleClient.publish(topic+"/notice/", message);
72.                     /**
73.                      * 如果发送P2P消息，二级Topic必须是“p2p”，三级Topic是目标的ClientID
74.                      * 此处设置的三级Topic需要是接收方的ClientID
75.                      */
76.                     String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
77.                     sampleClient.publish(p2pTopic,message);
78.                 } catch (Exception e) {
79.                     e.printStackTrace();
80.                 }
81.             }
82.         } catch (Exception me) {
83.             me.printStackTrace();
84.         }
85.     }
86. }

1. public class MQTTRecvMsg {
2.    public static void main(String[] args) throws IOException {
3.        /**
4.          * 设置当前用户私有的MQTT的接入点。例如此处示意使用XXX，实际使用请替换用户自己的接入点。接入点的获取方法是，在控制台申请MQTT实例，每个实例都会分配一个接入点域名。
5.          */
6.         final String broker ="tcp://XXXX.mqtt.aliyuncs.com:1883";
7.         /**
8.          * 设置阿里云的AccessKey，用于鉴权
9.          */
10.         final String acessKey ="XXXXXX";
11.         /**
12.          * 设置阿里云的SecretKey，用于鉴权
13.          */
14.         final String secretKey ="XXXXXXX";
15.         /**
16.          * 发消息使用的一级Topic，需要先在MQ控制台里申请
17.          */
18.         final String topic ="XXXX";
20.         /**
21.          * MQTT的ClientID，一般由两部分组成，GroupID@@@DeviceID
22.          * 其中GroupID在MQ控制台里申请
23.          * DeviceID由应用方设置，可能是设备编号等，需要唯一，否则服务端拒绝重复的ClientID连接
24.          */
25.         final String clientId ="GID_XXXX@@@ClientID_XXXXXX";
26.         String sign;
27.         MemoryPersistence persistence = new MemoryPersistence();
28.         try {
29.             final MqttClient sampleClient = new MqttClient(broker, clientId, persistence);
30.             final MqttConnectOptions connOpts = new MqttConnectOptions();
31.             System.out.println("Connecting to broker: " + broker);
32.             /**
33.              * 计算签名，将签名作为MQTT的password
34.              * 签名的计算方法，参考工具类MacSignature，第一个参数是ClientID的前半部分，即GroupID
35.              * 第二个参数阿里云的SecretKey
36.              */
37.             sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
38.             /**
39.              * 设置订阅方订阅的Topic集合，此处遵循MQTT的订阅规则，可以是一级Topic，二级Topic，P2P消息请订阅/p2p
40.              */
41.             final String[] topicFilters=new String[]{topic+"/notice/",topic+"/p2p"};
42.             final int[]qos={0,0};
43.             connOpts.setUserName(acessKey);
44.             connOpts.setServerURIs(new String[] { broker });
45.             connOpts.setPassword(sign.toCharArray());
46.             connOpts.setCleanSession(true);
47.             connOpts.setKeepAliveInterval(90);
48.             connOpts.setAutomaticReconnect(true);
49.             sampleClient.setCallback(new MqttCallbackExtended() {
50.                 public void connectComplete(boolean reconnect, String serverURI) {
51.                     System.out.println("connect success");
52.                     //连接成功，需要上传客户端所有的订阅关系
53.                     sampleClient.subscribe(topicFilters,qos);
54.                 }
55.                 public void connectionLost(Throwable throwable) {
56.                     System.out.println("mqtt connection lost");
57.                 }
58.                 public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
59.                     System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
60.                 }
61.                 public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
62.                     System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
63.                 }
64.             });
65.             //客户端每次上线都必须上传自己所有涉及的订阅关系，否则可能会导致消息接收延迟
66.             sampleClient.connect(connOpts);
67.             //每个客户端最多允许存在30个订阅关系，超出限制可能会丢弃导致收不到部分消息
68.             sampleClient.subscribe(topicFilters,qos);
69.             Thread.sleep(Integer.MAX_VALUE);
70.         } catch (Exception me) {
71.             me.printStackTrace();
72.         }
73.     }
74. }

1. public class MacSignature {
2.     /**
3.      * @param text      要签名的文本
4.      * @param secretKey 阿里云MQ SecretKey
5.      * @return 加密后的字符串
6.      * @throws InvalidKeyException
7.      * @throws NoSuchAlgorithmException
8.      */
9.     public static String macSignature(String text, String secretKey) throws InvalidKeyException, NoSuchAlgorithmException {
10.         Charset charset = Charset.forName("UTF-8");
11.         String algorithm = "HmacSHA1";
12.         Mac mac = Mac.getInstance(algorithm);
13.         mac.init(new SecretKeySpec(secretKey.getBytes(charset), algorithm));
14.         byte[] bytes = mac.doFinal(text.getBytes(charset));
15.         return new String(Base64.encodeBase64(bytes), charset);
16.     }
17.     /**
18.      * 发送方签名方法
19.      *
20.      * @param clientId  MQTT ClientID
21.      * @param secretKey 阿里云MQ SecretKey
22.      * @return 加密后的字符串
23.      * @throws NoSuchAlgorithmException
24.      * @throws InvalidKeyException
25.      */
26.     public static String publishSignature(String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
27.         return macSignature(clientId, secretKey);
28.     }
29.     /**
30.      * 订阅方签名方法
31.      *
32.      * @param topics    要订阅的Topic集合
33.      * @param clientId  MQTT ClientID
34.      * @param secretKey 阿里云MQ SecretKey
35.      * @return 加密后的字符串
36.      * @throws NoSuchAlgorithmException
37.      * @throws InvalidKeyException
38.      */
39.     public static String subSignature(List<String> topics, String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
40.         Collections.sort(topics); //以字典顺序排序
41.         String topicText = "";
42.         for (String topic : topics) {
43.             topicText += topic + "\n";
44.         }
45.         String text = topicText + clientId;
46.         return macSignature(text, secretKey);
47.     }
48.     /**
49.      * 订阅方签名方法
50.      *
51.      * @param topic     要订阅的Topic
52.      * @param clientId  MQTT ClientID
53.      * @param secretKey 阿里云MQ SecretKey
54.      * @return 加密后的字符串
55.      * @throws NoSuchAlgorithmException
56.      * @throws InvalidKeyException
57.      */
58.     public static String subSignature(String topic, String clientId, String secretKey) throws NoSuchAlgorithmException, InvalidKeyException {
59.         List<String> topics = new ArrayList<String>();
60.         topics.add(topic);
61.         return subSignature(topics, clientId, secretKey);
62.     }
63. }