声明:原文出处已在文末标出,本人出于学习,对其做了整理,收集干货,不作商业用途!

一.MQTT介绍

1.简介

MQTT(message queuing telemetry transport)是IBM开发的即时通讯协议,是一种发布/订阅极其轻量级的消息传输协议,专门为网络受限设备、低宽带以及高延迟和不可靠的网络而设计的。由于以上轻量级的特点,是实现智能家居的首选传输协议,相比于XMPP,更加轻量级而且占用宽带低。

2.特点

a.由于采用发布/订阅的消息模式,可以提供一对多的消息发布
b.轻量级,网络开销小
c.对负载内容会有屏蔽的消息传输
d.有三种消息发布质量(Qos):
qos=0:“至多一次”,这一级别会发生消息丢失或重复,消息发布依赖于TCP/IP网络
qos=1:“至少一次”,确保消息到达,但消息重复可能会发生
qos=2:“只有一次”,确保消息到达一次
e.通知机制,异常中断时会通知双方

3.原理

MQTT协议有三种身份:发布者、代理、订阅者,发布者和订阅者都为客户端,代理为服务器,同时消息的发布者也可以是订阅者(为了节约内存和流量发布者和订阅者一般都会定义在一起)。
MQTT传输的消息分为主题(Topic,可理解为消息的类型,订阅者订阅后,就会收到该主题的消息内容(payload))和负载(payload,可以理解为消息的内容)两部分。

二.MQTT通用使用

通用的使用方式的重连机制在安卓系统中需要自己去编写,下一篇会详细讲解阿里专门针对Android客户端的实现方式MqttAndroidClient

1.集成

1.build.grade中导入

implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.0'
          

2.添加权限

<uses-permission android:name="android.permission.INTERNET" />
          <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
          <uses-permission android:name="android.permission.WAKE_LOCK" />
          

2.使用

直接上代码,以下就是发布者和订阅者一体的实现方式,自己添加了重连机制,对应的注释也比较清晰,如果对于MQTT入门最好看下阿里MQTT的官方文档

public class MQTTManager {
          
              private String broker = "";//固定配置
              private String secretKey = "";//固定配置
              private String acessKey = "";//固定配置
          
              private String topic = "";//自己定义
              private String groupId = "";//自己定义
              private String clientId = "";//自己定义
          
              private MqttClient mqttClient;
              private volatile static MQTTManager manager;
          
              private int[] qos = {0, 0};//订阅个数就是数组的长度
          
              private ScheduledExecutorService reconnectPool;//重连线程池
          
              public static MQTTManager getInstance() {
                  if (manager == null) {
                      synchronized (MQTTManager.class) {
                          if (manager == null)
                              manager = new MQTTManager();
                      }
                  }
                  return manager;
              }
          
              public MQTTManager() {
                  clientId = String.format("%s@@@%s", groupId, MQTTCons.Sep_SEND);//这是根据自己需求定义的clientId
              }
          
              /**
               * 发送信息
               * @param msg
               */
              public void sendMessage(String msg) {
                  MqttMessage message = new MqttMessage(msg.getBytes());
                  message.setQos(0);
                  try {
                      if (mqttClient != null)
                          mqttClient.publish(topic, message);
                  } catch (MqttException e) {
                      e.printStackTrace();
                      TVLog.i("MqttException-sendMQTT-" + e);
                  }
              }
          
              /**
               * 开启MQTT连接和订阅
               */
              public void startSendMQTT() {
                  try {
                      closeMQTT();//断开和关闭连接的操作,由于我们需求需要,切换用户要重新创建新的连接,一般应用中基本都会始终订阅一条
                      MemoryPersistence persistence = new MemoryPersistence();
                      mqttClient = new MqttClient(broker, clientId, persistence);
                      final MqttConnectOptions connOpts = new MqttConnectOptions();
                      String sign = MacSignature.macSignature(clientId.split("@@@")[0], secretKey);
                      connOpts.setUserName(acessKey);
                      connOpts.setServerURIs(new String[]{broker});
                      connOpts.setPassword(sign.toCharArray());
                      connOpts.setCleanSession(true);
                      connOpts.setKeepAliveInterval(20);
                      connOpts.setConnectionTimeout(10);
                      connOpts.setMqttVersion(MQTT_VERSION_3_1_1);
                      connOpts.setAutomaticReconnect(false);//禁用自带重连机制,用于TV端会出现不稳定性,所以自己写了重连
                      mqttClient.setCallback(new MqttCallbackExtended() {
                          public void connectComplete(boolean reconnect, String serverURI) {
                              TVLog.i("Send connect success" + topic);
                              closeReconnectTask();
                              subscribeFilter();//mqtt每次连接成功都得订阅Topic
                          }
          
                          public void connectionLost(Throwable throwable) {
                              TVLog.i("mqtt connection lost");
                              startReconnectTask();
                          }
          
                          public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
                              TVLog.i("messageArrived:" + topic + "------" + new String(mqttMessage.getPayload()));
                          }
          
                          public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                              try {
                                  TVLog.i("deliveryComplete:" + iMqttDeliveryToken.getMessage().toString());
                              } catch (MqttException e) {
                                  e.printStackTrace();
                              }
                          }
                      });
                      mqttClient.connect(connOpts);
                  } catch (Exception me) {
                      me.printStackTrace();
                  }
              }
          
              /**
               * 订阅Topic
               */
              private void subscribeFilter() {
                  String registerTopic = "";//自定义
                  String controlTopic = "";//自定义,作为示例订阅了两个
                  String[] topicFilters = new String[]{registerTopic, controlTopic};
          
                  try {
                      mqttClient.subscribe(topicFilters, qos);
                  } catch (MqttException e) {
                      e.printStackTrace();
                  }
              }
          
              private synchronized void startReconnectTask() {//开启重连任务
                  if (reconnectPool != null) return;
                  reconnectPool = Executors.newScheduledThreadPool(1);
                  reconnectPool.scheduleAtFixedRate(new Runnable() {
                      @Override
                      public void run() {
                          try {
                              if (mqttClient == null || mqttClient.isConnected()) return;
                              mqttClient.reconnect();
                              TVLog.d("reconnectSendMQTT" + topic);
                          } catch (Exception e) {
                              e.printStackTrace();
                          }
                      }
                  }, 0, 5 * 1000, TimeUnit.MILLISECONDS);
              }
          
              public synchronized void closeReconnectTask() {//程序销毁的时候也记得关闭
                  if (reconnectPool != null) {
                      reconnectPool.shutdownNow();
                      reconnectPool = null;
                  }
              }
          
              public void closeMQTT() {//在程序销毁的时候也记得调用
                  try {
                      closeReconnectTask();
                      if (mqttClient != null) {
                          mqttClient.disconnect();
                          mqttClient.close();
                          mqttClient = null;
                          TVLog.d("closeSendMQTT" + topic);
                      }
                  } catch (MqttException e) {
                      e.printStackTrace();
                  }
              }
          
          }
          

注意:mqttClient.disconnect()和mqttClient.connect(connOpts)都为耗时操作,网络差的时候会阻塞主线程,可以新开线程或者直接用线程池ScheduledExecutorService管理

结语

转载:https://www.jianshu.com/p/a371c6ac076b