阅读979 返回首页    go windows

MQTT 客户端收发 MQTT 消息__Java 接入示例_MQTT 接入(物联)_消息队列 MQ-阿里云

本文主要介绍如何使用 MQTT 客户端收发 MQTT 消息，并给出示例代码供前期开发测试参考，包括资源申请、环境准备、示例代码、注意事项等。

注意：

本文给出的实例均基于 Eclipse Paho Java SDK 实现，SDK 下载请参见 MQTT 接入准备。如使用其他第三方的客户端，请适当修改。

1. 资源申请

使用 MQ 提供的 MQTT 服务，首先需要核实应用中使用的 Topic 资源是否已经申请，如果没有，请先去控制台申请 Topic，Group ID 等资源。

申请资源时需要根据需求选择对应的 Region，例如 MQTT 需要使用华北2的接入点，那么 Topic 等资源就在华北2 申请，资源申请具体请参见申请 MQ 资源。

注意：MQTT 使用的多级子 Topic 不需要申请，代码里直接使用即可，没有限制。

2. 环境准备

使用 MQTT 协议来收发消息，需要根据应用平台选择合适的客户端。本示例运行在 Java 平台，使用 Eclipse Paho Java SDK 构建。首先引入 Maven 依赖，POM 文件配置如下：

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>org.eclipse.paho</groupId>
  4.         <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
  5.         <version>1.0.2</version>
  6.     </dependency>
  7. </dependencies>
  8. <repositories>
  9.         <repository>
  10.             <id>Eclipse Paho Repo</id>
  11.             <url>https://repo.eclipse.org/content/repositories/paho-releases/</url>
  12.         </repository>
  13.         <repository>
  14.             <id>snapshots-repo</id>
  15.             <url>https://oss.sonatype.org/content/repositories/snapshots</url>
  16.             <releases>
  17.                 <enabled>false</enabled>
  18.             </releases>
  19.             <snapshots>
  20.                 <enabled>true</enabled>
  21.             </snapshots>
  22.         </repository>
  23.  </repositories>

3. MQTT 发送消息

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

  1. public class MQTTSendMsg {
  2.     public static void main(String[] args) throws IOException {
  3.         /**
  4.          * 设置MQTT的接入点，请根据应用所在环境选择合适的Region，不支持跨Region访问
  5.          */
  6.         final String broker ="tcp://mqtt-test.cn-qingdao.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(false);
  42.             connOpts.setKeepAliveInterval(100);
  43.             sampleClient.setCallback(new MqttCallback() {
  44.                 public void connectionLost(Throwable throwable) {
  45.                     System.out.println("mqtt connection lost");
  46.                     throwable.printStackTrace();
  47.                     while(!sampleClient.isConnected()){
  48.                         try {
  49.                             sampleClient.connect(connOpts);
  50.                         } catch (MqttException e) {
  51.                             e.printStackTrace();
  52.                         }
  53.                         try {
  54.                             Thread.sleep(1000);
  55.                         } catch (InterruptedException e) {
  56.                             e.printStackTrace();
  57.                         }
  58.                     }
  59.                 }
  60.                 public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
  61.                     System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
  62.                 }
  63.                 public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
  64.                     System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
  65.                 }
  66.             });
  67.             sampleClient.connect(connOpts);
  68.             for (int i = 0; i < 10; i++) {
  69.                 try {
  70.                     String scontent = new Date()+"MQTT Test body" + i;
  71.                     final MqttMessage message = new MqttMessage(scontent.getBytes());
  72.                     message.setQos(0);
  73.                     System.out.println(i+" pushed at "+new Date()+" "+ scontent);
  74.                     /**
  75.                      *消息发送到某个主题Topic，所有订阅这个Topic的设备都能收到这个消息。
  76.                      * 遵循MQTT的发布订阅规范，Topic也可以是多级Topic。此处设置了发送到二级Topic
  77.                      */
  78.                     sampleClient.publish(topic+"/notice/", message);
  79.                     /**
  80.                      * 如果发送P2P消息，二级Topic必须是“p2p”，三级Topic是目标的ClientID
  81.                      * 此处设置的三级Topic需要是接收方的ClientID
  82.                      */
  83.                     String p2pTopic =topic+"/p2p/GID_mqttdelay3@@@DEVICEID_001";
  84.                     sampleClient.publish(p2pTopic,message);
  85.                 } catch (Exception e) {
  86.                     e.printStackTrace();
  87.                 }
  88.             }
  89.         } catch (Exception me) {
  90.             me.printStackTrace();
  91.         }
  92.     }
  93. }

4. MQTT 接收消息

本段代码演示如何使用 MQTT 客户端订阅消息，接收普通的消息以及点对点消息。

  1. public class MQTTRecvMsg {
  2.    public static void main(String[] args) throws IOException {
  3.         /**
  4.          * 设置MQTT的接入点，请根据应用所在环境选择合适的Region，不支持跨Region访问
  5.          */
  6.         final String broker ="tcp://mqtt-test.cn-qingdao.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消息不需要显式订阅
  40.              */
  41.             final String[] topicFilters=new String[]{topic+"/notice/"};
  42.             final int[]qos={0};
  43.             connOpts.setUserName(acessKey);
  44.             connOpts.setServerURIs(new String[] { broker });
  45.             connOpts.setPassword(sign.toCharArray());
  46.             connOpts.setCleanSession(false);
  47.             connOpts.setKeepAliveInterval(100);
  48.             sampleClient.setCallback(new MqttCallback() {
  49.                 public void connectionLost(Throwable throwable) {
  50.                     System.out.println("mqtt connection lost");
  51.                     throwable.printStackTrace();
  52.                     while(!sampleClient.isConnected()){
  53.                         try {
  54.                             sampleClient.connect(connOpts);
  55.                             sampleClient.subscribe(topicFilters,qos);
  56.                         } catch (MqttException e) {
  57.                             e.printStackTrace();
  58.                         }
  59.                         try {
  60.                             Thread.sleep(1000);
  61.                         } catch (InterruptedException e) {
  62.                             e.printStackTrace();
  63.                         }
  64.                     }
  65.                 }
  66.                 public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
  67.                     System.out.println("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
  68.                 }
  69.                 public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
  70.                     System.out.println("deliveryComplete:" + iMqttDeliveryToken.getMessageId());
  71.                 }
  72.             });
  73.             sampleClient.connect(connOpts);
  74.             sampleClient.subscribe(topicFilters,qos);
  75.             Thread.sleep(Integer.MAX_VALUE);
  76.         } catch (Exception me) {
  77.             me.printStackTrace();
  78.         }
  79.     }
  80. }

上文代码用到的工具类 MacSignature.java 如下：

  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. }

最后更新：2016-11-29 09:51:55

  上一篇：go MQ 客户端收发 MQTT 消息__Java 接入示例_MQTT 接入(物联)_消息队列 MQ-阿里云
  下一篇：go SSL 方式接入示例__Java 接入示例_MQTT 接入(物联)_消息队列 MQ-阿里云