概述
添加权限
<uses-permission android:name="android.permission.INTERNET" />
<uses-permission android:name="android.permission.WAKE_LOCK" />
<!-- 允许程序访问WiFi网络信息 -->
<uses-permission android:name="android.permission.ACCESS_WIFI_STATE" />
<uses-permission android:name="android.permission.CHANGE_WIFI_STATE" />
<!-- 允许程序获取网络状态 -->
<uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
添加依赖
/** MQTT 通信 */
implementation group: 'org.eclipse.paho', name: 'org.eclipse.paho.client.mqttv3', version: '1.2.2'
implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
implementation 'androidx.localbroadcastmanager:localbroadcastmanager:1.0.0'
implementation 'com.blankj:utilcode:1.30.0'
MQTTService 类
public class MQTTService extends Service {
public static final String TAG = MQTTService.class.getSimpleName();
private static MqttAndroidClient client;
private MqttConnectOptions conOpt;
private boolean reConnectedCode = true;
private String host = "tcp://xxx.xxx.xxx.xxx:1883";
private String userName = "username";
private String passWord = "password";
private static String myTopic = "/XXX/xxx";//要订阅的主题
private static String sendTopic = "/XXX/xxx";//要发布控制主题
private String clientId = "clientId_" + System.currentTimeMillis();//客户端标识
private IGetMessageCallBack mGetMessageCallBack;
static IGetMessageCallBack mcallBack;
@Override
public void onCreate() {
super.onCreate();
Log.e(getClass().getName(), "onCreate");
init();
}
public static void pubSatelish(String msg) {
String topic = sendTopic;
Integer qos = 0;
Boolean retained = false;
try {
if (client != null) {
client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());
}
} catch (MqttException e) {
e.printStackTrace();
}
}
public static void publish(String msg, IGetMessageCallBack callBack) {
mcallBack = callBack;
String topic = sendTopic;
Integer qos = 0;
Boolean retained = false;
if (client != null) {
try {
client.publish(topic, msg.getBytes(), qos.intValue(), retained.booleanValue());
} catch (MqttException e) {
e.printStackTrace();
}
}
}
private void init() {
// 服务器地址（协议+地址+端口号）
String uri = host;
client = new MqttAndroidClient(this, uri, clientId);
// 设置MQTT监听并且接受消息
client.setCallback(mqttCallback);
conOpt = new MqttConnectOptions();
//设置为false可以在服务器断开后不用再手动连接
conOpt.setCleanSession(true);
// 设置超时时间 单位为秒
//设置为0，防止 ERROR o.e.p.c.mqttv3.internal.ClientState - Timed out as no activity 错误
conOpt.setConnectionTimeout(10);
// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
conOpt.setKeepAliveInterval(20);
//设置自动重连
conOpt.setAutomaticReconnect(true);
// 用户名
conOpt.setUserName(userName);
// 密码
conOpt.setPassword(passWord.toCharArray()); //将字符串转换为字符串数组
// last will message
boolean doConnect = true;
String message = "{"terminal_uid":"" + clientId + ""}";
Log.e(getClass().getName(), "message是:" + message);
String topic = myTopic;
Integer qos = 0;
Boolean retained = false;
if ((!message.equals("")) || (!topic.equals(""))) {
// 最后的遗嘱
// MQTT本身就是为信号不稳定的网络设计的，所以难免一些客户端会无故的和Broker断开连接。
//当客户端连接到Broker时，可以指定LWT，Broker会定期检测客户端是否有异常。
//当客户端异常掉线时，Broker就往连接时指定的topic里推送当时指定的LWT消息。
try {
conOpt.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
} catch (Exception e) {
Log.i(TAG, "Exception Occured", e);
doConnect = false;
iMqttActionListener.onFailure(null, e);
}
}
if (doConnect) {
doClientConnection();
}
}
@Override
public void onDestroy() {
reConnectedCode = false;
// 此处解决MQTT退出异常问题
if (client != null) {
client.unregisterResources();
client.close();
}
super.onDestroy();
}
/**
* 连接MQTT服务器
*/
private void doClientConnection() {
Log.i(TAG, "连接................... ");
if (!getClientConnected() && isConnectIsNormal()) {
try {
Log.i(TAG, "连接中................... ");
client.connect(conOpt, null, iMqttActionListener);
} catch (MqttException e) {
e.printStackTrace();
}
}
}
private boolean getClientConnected() {
try {
return client.isConnected();
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
// MQTT是否连接成功
private IMqttActionListener iMqttActionListener = new IMqttActionListener() {
@Override
public void onSuccess(IMqttToken arg0) {
Log.i(TAG, "连接成功 ");
try {
// 订阅myTopic话题
Log.i(TAG, "SUB:" + myTopic);
client.subscribe(myTopic, 0);
} catch (MqttException e) {
e.printStackTrace();
Log.e(TAG, "sube：" + e.getMessage());
}
}
@Override
public void onFailure(IMqttToken arg0, Throwable arg1) {
arg1.printStackTrace();
// 连接失败，重连
}
};
// MQTT监听并且接受消息
private MqttCallback mqttCallback = new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
Log.i(TAG, "连接成功 ");
try {
// 订阅myTopic话题
Log.i(TAG, "SUB:" + myTopic);
client.subscribe(myTopic, 0);
} catch (MqttException e) {
e.printStackTrace();
Log.e(TAG, "sube：" + e.getMessage());
}
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
String str1 = new String(message.getPayload());
Log.i(TAG, "mGetMessageCallBack:" + mGetMessageCallBack);
if (mGetMessageCallBack != null) {
mGetMessageCallBack.setMessage(str1);
}
if (mcallBack != null) {
mcallBack.setMessage(str1);
}
String str2 = topic + ";qos:" + message.getQos() + ";retained:" + message.isRetained();
Log.i(TAG, "messageArrived:" + str1);
Log.i(TAG, str2);
}
@Override
public void deliveryComplete(IMqttDeliveryToken arg0) {
}
@Override
public void connectionLost(Throwable arg0) {
// 失去连接，重连
Log.i(TAG, "失去连接，重连.................");
// reConnecte();
}
};
/**
* 判断网络是否连接
*/
private boolean isConnectIsNormal() {
ConnectivityManager connectivityManager = (ConnectivityManager) this.getApplicationContext()
.getSystemService(Context.CONNECTIVITY_SERVICE);
NetworkInfo info = connectivityManager.getActiveNetworkInfo();
if (info != null && info.isAvailable()) {
String name = info.getTypeName();
Log.i(TAG, "MQTT当前网络名称：" + name);
return true;
} else {
Log.i(TAG, "MQTT 没有可用网络");
return false;
}
}
@Override
public IBinder onBind(Intent intent) {
Log.e(getClass().getName(), "onBind");
return new CustomBinder();
}
public void setIGetMessageCallBack(IGetMessageCallBack IGetMessageCallBack) {
this.mGetMessageCallBack = IGetMessageCallBack;
}
public class CustomBinder extends Binder {
public MQTTService getService() {
return MQTTService.this;
}
}
@Override
public boolean onUnbind(Intent intent) {
return super.onUnbind(intent);
}
}
消息回调接口
public interface IGetMessageCallBack {
public void setMessage(String message);
}
MQTT 连接服务类 MyServiceConnection
public class MyServiceConnection implements ServiceConnection {
private MQTTService mqttService;
private IGetMessageCallBack mIGetMessageCallBack;
@Override
public void onServiceConnected(ComponentName componentName, IBinder iBinder) {
mqttService = ((MQTTService.CustomBinder) iBinder).getService();
mqttService.setIGetMessageCallBack(mIGetMessageCallBack);
}
@Override
public void onServiceDisconnected(ComponentName componentName) {
}
public MQTTService getMqttService() {
return mqttService;
}
public void setIGetMessageCallBack(IGetMessageCallBack IGetMessageCallBack) {
mIGetMessageCallBack = IGetMessageCallBack;
}
}
初始化MQTT 绑定服务 到 Activity
// 此处申请绑定
private void initConnectTion() {
serviceConnection = new MyServiceConnection();
serviceConnection.setIGetMessageCallBack(new IGetMessageCallBack() {
@Override
public void setMessage(String message) {
//接收到消息回调
}
});
Intent intent = new Intent(this, MQTTService.class);
bindService(intent, serviceConnection, Context.BIND_AUTO_CREATE);
}
销毁服务
@Override
protected void onDestroy() {
// 要在此处进行解绑
if (serviceConnection != null) {
unbindService(serviceConnection);
}
super.onDestroy();
}
注册service
<service android:name="org.eclipse.paho.android.service.MqttService" />
<service android:name=".service.MQTTService" android:enabled="true" android:exported="true"/>
最后
以上就是长情天空为你收集整理的android 实现mqtt消息推送，以及断线重连的问题解决的全部内容，希望文章能够帮你解决android 实现mqtt消息推送，以及断线重连的问题解决所遇到的程序开发问题。
如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。
本图文内容来源于网友提供，作为学习参考使用，或来自网络收集整理，版权属于原作者所有。
发表评论 取消回复