目的是為了實時監控室外廣告屏的亮度，界面，聲音，開關機等等…… 因為室外的網絡情況是隨時可變的，所以采用的MQTT協議，作為Android客戶端來說因為用MQTT發送消息太繁瑣，我們采用的是客戶端只接收命令，然后用Http進行數據反饋，這個項目近期也做完了，故記錄一下。

第一步：導入在線庫

// mqtt 包導入 implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0' implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'

第二步：聲明廣播（這個是在線庫自帶的 直接這么寫就行，不用自己寫這個service）

<service android:name="org.eclipse.paho.android.service.MqttService" />

第三步：MqttManager的全部代碼

public class MqttManager {public final static String TAG = MqttManager.class.getSimpleName();@SuppressLint("StaticFieldLeak")private static volatile MqttManager mInstance = null;private MqttCallback mCallback;public MqttClient client;private MqttConnectOptions conOpt;private Context context;public String[] topic;private MqttManager(Context context) {mCallback = new MqttCallbackBus(context);this.context = context;}public static MqttManager getInstance(Context context) {if (mInstance == null) {synchronized (MqttManager.class) {if (mInstance == null) {mInstance = new MqttManager(context);}}}return mInstance;}/*** 釋放單例, 及其所引用的資源*/public static void release() {try {if (mInstance != null) {mInstance.disConnect();mInstance = null;}} catch (Exception e) {e.printStackTrace();}}/*** 創建Mqtt 連接** @param brokerUrl Mqtt服務器地址(tcp://xxxx:1863)* @param userName 用戶名* @param password 密碼* @return*/public boolean createConnect(String brokerUrl, String userName, String password) {topic = new String[]{PUBLISH_TOPIC};String deviceId = Utils.getDeviceNo();if (client != null && client.isConnected()) {return true;}boolean flag = false;try {conOpt = new MqttConnectOptions();conOpt.setCleanSession(true); //不接收離線期間的消息 每次都是重新登陸conOpt.setAutomaticReconnect(true); //自動重連if (!TextUtils.isEmpty(password)) {conOpt.setPassword(password.toCharArray());}if (!TextUtils.isEmpty(userName)) {conOpt.setUserName(userName);}client = new MqttClient(brokerUrl, deviceId, new MemoryPersistence());client.setCallback(mCallback);flag = doConnect();client.subscribe(topic);} catch (MqttException e) {e.printStackTrace();//重連String msg = "exception_reconnect";EventBus.getDefault().post(msg);}return flag;}/*** 建立連接** @return*/private boolean doConnect() {boolean flag = false;if (client != null) {try {client.connect(conOpt);flag = true;} catch (Exception e) {e.printStackTrace();}}return flag;}public boolean publish(String topicName, int qos, byte[] payload) {boolean flag = false;try {if (client == null) {createConnect(HOST, USERNAME, PASSWORD);}if (!client.isConnected()) {this.reconnect();}MqttMessage message = new MqttMessage(payload);message.setQos(qos);client.publish(topicName, message);flag = true;} catch (MqttException e) {e.printStackTrace();}return flag;}private boolean subscribe(String topicName, int qos) {boolean flag = false;if (client != null && client.isConnected()) {try {client.subscribe(topicName, qos);flag = true;} catch (MqttException e) {e.printStackTrace();}}return flag;}private boolean subscribe(String[] topicName, int qos[]) {boolean flag = false;if (client != null && client.isConnected()) {try {client.subscribe(topicName, qos);flag = true;} catch (MqttException e) {e.printStackTrace();}}return flag;}/*** 取消連接** @throws MqttException*/public void disConnect() throws MqttException {if (client != null && client.isConnected()) {client.disconnect();}}/*** 關閉連接*/public void close() {if (client != null && client.isConnected()) {try {client.disconnect();} catch (MqttException e) {e.printStackTrace();}}}/*** 重新連接*/private void reconnect() {if (client != null && !client.isConnected()) {try {client.setCallback(mCallback);client.connect(conOpt);client.subscribe(topic);} catch (MqttException e) {e.printStackTrace();}}} }

里面用eventBus做了下重連處理，后續響應如下

boolean isHavNet = isConnectIsNomarl(); //判斷是否有網 if (isHavNet) {handler.removeMessages(4);MqttManager.getInstance(BaseApplication.getContext()).createConnect(HOST, USERNAME, PASSWORD); }else{handler.sendEmptyMessageDelayed(4, 10000);//無網等待有網再連 }

第四步：mqtt消息回調類

public class MqttCallbackBus implements MqttCallbackExtended {private Context mContext;public MqttCallbackBus(Context context) {this.mContext = context;}@Overridepublic void connectComplete(boolean reconnect, String serverURI) {Log.e("MqttCallbackBus", "MQTT_connectComplete:");//斷開連接必須重新訂閱才能收到之前訂閱的session連接的消息if(reconnect){Log.e("MqttCallbackBus_重連訂閱主題", "MQTT_connectComplete:");//這里是發送消息去重新訂閱String msg = "reconnect";EventBus.getDefault().postSticky(msg);}}@Overridepublic void connectionLost(Throwable cause) { //掉線Log.e("MqttCallbackBus>>>", "MQTT_connectionLost 掉線原因:"+cause.getMessage());cause.printStackTrace();}@Overridepublic void messageArrived(String topic, MqttMessage message) throws Exception {MqttReceivedMessage msg = (MqttReceivedMessage) message;String s = msg.toString();Log.e("MQTT_messageArrived_msg", "MQTT_msg"+topic);HashMap hs = new Gson().fromJson(s, HashMap.class);EventBus.getDefault().post(hs); //拋出收到的消息 eventbus響應做自己的業務處理}@Overridepublic void deliveryComplete(IMqttDeliveryToken token) { //（發布）publish后會執行到這里,發送狀態 token.isComplete()Log.e("MqttCallbackBus>>>", "MQTT_deliveryComplete:");} }

其中connectComplete的重連回調reconnect = true 的eventbus的響應是：

//重連 能進重連 說明必有網了 不必再else 定時重試網絡String[] topic = new String[]{PUBLISH_TOPIC};try {MqttManager.getInstance(BaseApplication.getContext()).client.subscribe(topic);} catch (MqttException e) {e.printStackTrace();}

MainActivity的使用如下：

//網絡變化監聽 首次連接mqttboolean isHavNet = isConnectIsNomarl();if (isHavNet) {handler.removeMessages(1);MqttManager.getInstance(BaseApplication.getContext()).createConnect(HOST, USERNAME, PASSWORD);} else {handler.sendEmptyMessageDelayed(1, 5000);}

MQTT全局的配置參數：

?

附贈個判斷是否有網的工具類

/*** 判斷網絡是否連接*/ private boolean isConnectIsNomarl() {ConnectivityManager connectivityManager = (ConnectivityManager) this.getApplicationContext().getSystemService(Context.CONNECTIVITY_SERVICE);@SuppressLint("MissingPermission") NetworkInfo info = connectivityManager.getActiveNetworkInfo();if (info != null && info.isConnected()) {String name = info.getTypeName();return true;} else {return false;} }

總結

以上是生活随笔為你收集整理的Android MQTT客户端的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。