基于paho.mqtt.java的开源Android端MQTT框架 MqttAndroid

背景

现在最常用的Android端MQTT框架是 paho.mqtt.android。paho.mqtt.android作为对 paho.mqtt.java 的包装嵌套，具体实现逻辑比较臃肿。于是想着模仿 paho.mqtt.android，实现对 paho.mqtt.java 更加简单的包装处理。

简介

对 paho.mqtt.java 库进行了的简单包装，方便Android端使用。

Github地址：https://github.com/shiyinghan/MqttAndroid

使用方法

第一步：在project的build.gradle 文件中添加JitPack依赖

allprojects {
    
    
    repositories {
    
    
        ...
        maven {
    
     url 'https://jitpack.io' }
    }
}

第二步: 在Module的build.gradle文件中添加对本库的依赖

dependencies {
    
    
    implementation 'com.github.shiyinghan:MqttAndroid:1.0.2'
}

第三步：开始使用

以下方法可直接拷贝到项目中使用参数根据实际情况修改即可

1). 构建MqttAndroidClient对象

mqttClient = new MqttAndroidClient(context, serverURI, clientId);
mqttClient.setCallback(new MqttCallbackExtended() {
    
    
	@Override
	public void connectComplete(boolean reconnect, String serverURI) {
    
    
	    //连接成功
	}
     
  	@Override
    public void connectionLost(Throwable cause) {
    
    
        //连接断开
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
    
    
		//接收订阅的消息
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
    
    
		//消息发送成功
    }
});

2). 连接Mqtt服务器

MqttConnectOptions options = new MqttConnectOptions();
//设置自动重连
options.setAutomaticReconnect(true);
//设置清除回话session
options.setCleanSession(true);
//设置连接超时时间
options.setConnectionTimeout(30);
//设置心跳包默认的发送间隔
options.setKeepAliveInterval(60);
//        //设置用户名（可选）
//        options.setUserName(userName);
//        //设置密码（可选）
//        options.setPassword(password.toCharArray());
//        //设置遗愿（可选）
//        options.setWill(topic, payload.getBytes(), qos, retained);

try {
    
    
  mqttClient.connect(options, this, new IMqttActionListener() {
    
    
       @Override
       public void onSuccess(IMqttToken asyncActionToken) {
    
    
           //连接成功
       }

       @Override
       public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
    
    
           //连接失败
       }
   });
} catch (MqttException e) {
    
    
   e.printStackTrace();
}

3). 断开连接

 try {
    
    
 	mqttClient.disconnect();
 } catch (MqttException e) {
    
    
     e.printStackTrace();
 }

4). 判断连接状态

mqttClient.isConnected()

5). 订阅主题

try {
    
    
   int qos = 0;
   mqttClient.subscribe(subscriptionTopic, qos, context, new IMqttActionListener() {
    
    
       @Override
       public void onSuccess(IMqttToken asyncActionToken) {
    
    
           //订阅成功
       }

       @Override
       public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
    
    
           //订阅失败
       }
   });
} catch (MqttException e) {
    
    
   e.printStackTrace();
}

或者添加IMqttMessageListener监听

try {
    
    
   int qos = 0;
   mqttClient.subscribe(subscriptionTopic1, qos, this,
      new IMqttActionListener() {
    
    
            @Override
            public void onSuccess(IMqttToken asyncActionToken) {
    
    
                //订阅成功
            }

            @Override
            public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
    
    
                //订阅失败
            }
        },
        new IMqttMessageListener() {
    
    
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
    
    
                //接收订阅的消息
            }
        });
} catch (MqttException e) {
    
    
   e.printStackTrace();
}

6). 取消订阅

String[] topics = {
    
    subscriptionTopic, subscriptionTopic1};
try {
    
    
    mqttClient.unsubscribe(topics, context, new IMqttActionListener() {
    
    
        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
    
    
            //取消订阅成功
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
    
    
            //取消订阅失败
        }
    });
} catch (MqttException e) {
    
    
    e.printStackTrace();
}

7). 发布消息

try {
    
    
  	MqttMessage message = new MqttMessage();
    message.setQos(0);
    message.setPayload(publishMessage.getBytes());
    mqttClient.publish(publishTopic, message, this, new IMqttActionListener() {
    
    
        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
    
    
            //发布成功
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
    
    
            //发布失败
        }
    });
} catch (MqttException e) {
    
    
    e.printStackTrace();
}

注意事项

如果打包时开启混淆的话，Android Gradle Plugin版本不能是3.6.0或者3.6.1，请使用更高版本。因为 paho.mqtt.java 中的NetworkModule使用ServiceLoader.load进行加载，但是这两个版本的Gradle Plugin，混淆打包过程有问题，会遗漏META-INF/services文件夹，导致出现“no NetworkModule installed for scheme”错误。
IMqttActionListener 回调，默认会运行在UI线程。如果想运行在自定义线程，需要在该线程中，使用Looper.prepare()和Looper.loop()创建消息队列。
执行订阅操作subscribe()时，如果设置了IMqttMessageListener监听器。接收到消息时，会执行IMqttMessageListener监听器中的messageArrived方法，而不是连接的MqttCallback中的messageArrived方法。也就是说这两个messageArrived方法是互斥的，如果想用MqttCallback监听消息接收，就不要设置IMqttMessageListener监听器。