android paho框架,Android 开发笔记 04 篇：Mqtt 框架 - Paho Java
官方文档：Eclipse Paho Java Client
API：Package org.eclipse.paho.client.mqttv3
Paho 基础知识
Paho Java 客户端是用 Java 编写的 MQTT 客户端库，用于开发在 JVM 或其他 Java 兼容平台(例如 Android )上运行的应用程序 Paho Java 客户端提供了两个 API：MqttAsyncClient 提供了一个完全异步的 API，该活动通过注册的回调通知完成活动。 MqttClient 是 MqttAsyncClient 周围的同步包装器，其中函数与应用程序同步显示。
常用类和接口说明：
IMqttActionListener：异步操作完成后，将通知此接口的实现。
MqttCallback：使与客户端相关的异步事件发生时通知应用程序。
MqttAsyncClient：轻量级客户端，用于使用允许操作在后台运行的非阻塞方法与MQTT服务器通信。
MqttConnectOptions：包含用于控制客户端如何连接到服务器的选项集。
MqttMessage：MQTT消息包含应用程序有效负载和指定如何传递消息的选项。该消息包括表示为byte []的“有效负载”(消息的主体)。
Paho Java 实践
S1：添加依赖
implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.2'
S2：引入权限
S3：添加工具类(我自制)
/**
* Paho Java Client 工具类
* @author zhanghairong
* @version 2018-2-18
*/
public class MyMqttUtil {
// Log TAG
private final String TAG = "MyMqttUtil";
// MyMqttUtil 上下文环境
private Context mContext;
// MyMqttUtil 单例
private static MyMqttUtil mMyMqttUtilInstance;
// MqttAsyncClient 要连接的服务器
private final String serverURI = "tcp://xxxx:1883";
// MqttAsyncClient ID
private final String clientID = "123456";
// MqttAsyncClient 用于连接服务器的用户名
private final String userName = "root";
// MqttAsyncClient 用于连接服务器的密码
private final String password = "root";
// 断开连接的主题
private String Topic_DisCon = "Topic_DisCon";
// 数据持久化设置
private MqttDefaultFilePersistence dataStore;
// 客户端如何连接到服务器的选项集
private MqttConnectOptions mMqttConnectOptions;
// MqttAsyncClient 实例
private MqttAsyncClient mMqttAsyncClient;
// 私有构造
private MyMqttUtil(Context context) {
this.mContext = context;
initMqttClient();
}
// 单例模式
public static MyMqttUtil getInstance(Context context) {
if (mMyMqttUtilInstance == null) {
synchronized (MyMqttUtil.class) {
if (mMyMqttUtilInstance == null) {
mMyMqttUtilInstance = new MyMqttUtil(context);
}
}
}
return mMyMqttUtilInstance;
}
// 公共接口IMqttActionListener：异步操作完成后，将通知此接口的实现。在此处监听连接结果
private IMqttActionListener myIMqttActionListener = new IMqttActionListener() {
@Override// 操作成功完成后，将调用此方法。
public void onSuccess(IMqttToken asyncActionToken) {
Log.d(TAG, "MQTT 连接成功");
}
@Override// 操作失败时将调用此方法。
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
Log.d(TAG, "MQTT 连接失败");
}
};
// 公共接口MqttCallback：使与客户端相关的异步事件发生时通知应用程序。
private MqttCallback mMqttCallback = new MqttCallback() {
@Override// 与服务器的连接丢失时，将调用此方法。
public void connectionLost(Throwable cause) {
Log.d(TAG, "与服务器的连接丢失,尝试重连");
}
@Override// 从服务器收到消息时，将调用此方法。
public void messageArrived(String topic, MqttMessage message) throws Exception {
Log.d(TAG, "从服务器收到消息 topic 是：" + topic + " message 是：" + new String(message.getPayload()));
}
@Override// 在完成消息传递并收到所有确认后调用。
public void deliveryComplete(IMqttDeliveryToken token) {
Log.d(TAG, "完成消息传递并收到所有确认");
}
};
// 初始化配置以及连接服务器
public void initMqttClient(){
try {
// MqttConnectOptions：包含用于控制客户端如何连接到服务器的选项集。
mMqttConnectOptions = new MqttConnectOptions();
// 设置MQTT版本。默认操作是连接版本3.1.1，如果失败则回落到3.1。分别使用MQTT_VERSION_3_1_1或MQTT_VERSION_3_1选项，可以选择特定的版本3.1.1或3.1，而不会回退。
mMqttConnectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
// 设置连接超时值。该值(以秒为单位)定义了客户端将等待与MQTT服务器建立网络连接的最大时间间隔。默认超时为30秒。值为0将禁用超时处理，这意味着客户端将等待，直到成功建立网络连接或失败。
mMqttConnectOptions.setConnectionTimeout(10);
/*
设置客户端和服务器是否应该记住重新启动和重新连接之间的状态。
如果设置为false，则客户端和服务器都将在客户端，服务器和连接重新启动期间保持状态。维持状态：
即使重新启动客户端，服务器或连接，消息传递也将可靠地满足指定的QOS。
服务器会将订阅视为持久订阅。
如果设置为true，则客户端和服务器将不会在客户端，服务器或连接的重新启动期间保持状态。这表示
如果重新启动客户端，服务器或连接，则无法维持到指定QOS的消息传递
服务器会将订阅视为非持久订阅
*/
mMqttConnectOptions.setCleanSession(true);
/*
设置“保持活动”间隔。该值以秒为单位，定义了发送或接收消息之间的最大时间间隔。
它使客户端能够检测服务器是否不再可用，而不必等待TCP / IP超时。
客户端将确保在每个有效期内，至少有一条消息在网络上传播。
在这段时间内，如果没有与数据相关的消息，则客户端会发送一个很小的“ ping”消息，服务器将予以确认。
值为0将禁用客户端中的keepalive处理。
*/
mMqttConnectOptions.setKeepAliveInterval(10);
// 设置用于连接的用户名。
mMqttConnectOptions.setUserName(userName);
// 设置用于连接的密码。
mMqttConnectOptions.setPassword(password.toCharArray());
/*
设置连接的“最后的遗嘱”(LWT)。如果此客户端意外断开与服务器的连接，则服务器将使用提供的详细信息向自身发布消息。
topic -要发布到的主题。
payload -消息的字节有效负载。
qos -以(0、1或2)发布消息的服务质量。
retained -是否保留消息。
*/
mMqttConnectOptions.setWill(Topic_DisCon, "close".getBytes(), 1, false);
/*
设置如果连接断开，客户端是否将自动尝试重新连接到服务器。
如果设置为false，则在连接丢失的情况下，客户端将不会尝试自动重新连接到服务器。
如果设置为true，则在连接断开的情况下，客户端将尝试重新连接到服务器。它最初将等待1秒钟，然后再尝试重新连接，对于每次失败的重新连接尝试，延迟都会加倍，直到2分钟为止，此时延迟将保持2分钟。
*/
mMqttConnectOptions.setAutomaticReconnect(true);
dataStore = new MqttDefaultFilePersistence(System.getProperty("java.io.tmpdir"));
mMqttAsyncClient = new MqttAsyncClient(serverURI, clientID, dataStore);
mMqttAsyncClient.setCallback(mMqttCallback);// 设置一个回调侦听器以用于异步发生的事件。
} catch (MqttException e) {
e.printStackTrace();
}
}
// 连接服务器
public void connectServer(){
try {
if((mMqttAsyncClient != null) && (!mMqttAsyncClient.isConnected())){
mMqttAsyncClient.connect(mMqttConnectOptions, mContext, myIMqttActionListener);// 使用指定的选项连接到MQTT服务器
}
} catch (MqttException e) {
e.printStackTrace();
}
}
// 订阅主题
public void subTopic(String subTopic, int qos) {
if (mMqttAsyncClient != null && mMqttAsyncClient.isConnected()) {
try {
mMqttAsyncClient.subscribe(subTopic, qos);
Log.d(TAG, "订阅主题：" + subTopic + " 成功");
} catch (MqttException e) {
e.printStackTrace();
}
}
else {
Log.d(TAG, "服务器未连接");
}
}
// 推送数据
public void pubMessage(String pubTopic, String message){
if (mMqttAsyncClient != null && mMqttAsyncClient.isConnected()) {
try {
MqttMessage mqttMessage = new MqttMessage(message.getBytes());
mMqttAsyncClient.publish(pubTopic, mqttMessage);
Log.d(TAG, "推送主题：" + pubTopic + "数据 " + message + " 成功");
} catch (MqttException e) {
e.printStackTrace();
}
}
else {
Log.d(TAG, "服务器未连接");
}
}
// 断开连接
public void disConnectServer(){
if (mMqttAsyncClient != null && mMqttAsyncClient.isConnected()) {
try {
mMqttAsyncClient.disconnect();
Log.d(TAG, "服务器断开成功");
} catch (MqttException e) {
e.printStackTrace();
}
}
else {
Log.d(TAG, "服务器未连接");
}
}
}
S4：测试
btn_connect.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
MyMqttUtil.getInstance(Sample_MQTT.this).connectServer();
}
});
btn_publish.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
MyMqttUtil.getInstance(Sample_MQTT.this).pubMessage("Topic_Pub", "123456");
}
});
btn_subscribe.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
MyMqttUtil.getInstance(Sample_MQTT.this).subTopic("Topic_Sub", 1);
}
});
btn_disconnect.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
MyMqttUtil.getInstance(Sample_MQTT.this).disConnectServer();
}
});