devbox
代码探险家
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Android paho Mqtt物联网网络框架_adroid mqtt 单例

作者：代码探险家 | 2024-08-20 13:23:50

赞

踩

adroid mqtt 单例

先上地址 :  

在module下的build.gradle文件内的dependencies里面添加下面远程库，Androidstudio3.0以及以上的要用

implementation代替compile
compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.0.2'

项目结构下的build.gradle要添加maven仓库

  1. maven {
  2.  url "https://repo.eclipse.org/content/repositories/paho-releases/" 
  3. }

 

由于方便在项目中的使用，小编在这里把mqtt连接做成一个单例的管理类形式，回调信息和心跳处理也做了单独的处理

实现原理:1、先登录mqtt服务器；2、想要接收哪个主题下的消息，就要先订阅哪个主题；3、想发送给某个设备，就要把对方订阅的主题当做自己的发送主题，把数据发送给对方；整个连接数据收发大概就是这个样子，也可以自己加入断开重连，检测断开机制！

上一张项目目录图

好了废话不多说，贴代码

MqttManager管理类

  1. public class MqttManager {
  2.  
  3.     public static String TAG = MqttManager.class.getSimpleName();
  4.     private static volatile MqttManager mInstance = null;
  5.     private static MqttHeartBeat heartBeat = null;//心跳处理类，此心跳仅限于和另一端保持的一个心跳机制,并非与mqtt服务器之间的一个心跳,mqtt客户端与服务器之间自带心跳
  6.  
  7.     private MqttConnectOptions conOpt = null;
  8.     private MqttClient client = null;
  9.  
  10.     public static MqttManager getInstance() {
  11.         if (mInstance == null) {
  12.             synchronized (MqttManager.class) {
  13.                 if (mInstance == null) {
  14.                     mInstance = new MqttManager();
  15.                 }
  16.             }
  17.         }
  18.         return mInstance;
  19.     }
  20.  
  21.     /**
  22.      * 创建连接
  23.      *
  24.      * @param clientId 设备ID
  25.      * @return 连接成功true   失败false
  26.      */
  27.     public boolean createConnect(String clientId) {
  28.         boolean flag = false;
  29.         String tmpDir = System.getProperty("java.io.tmpdir");
  30.         MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
  31.         try {
  32.             conOpt = new MqttConnectOptions();
  33.             conOpt.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
  34.             conOpt.setCleanSession(true);
  35.             conOpt.setPassword(MqttApi.getPassword().toCharArray());
  36.             conOpt.setUserName(MqttApi.getName());
  37.             conOpt.setConnectionTimeout(30);    //设置连接超时
  38.             conOpt.setKeepAliveInterval(30);    //设置连接空闲最大时长   单位:秒
  39.             client = new MqttClient(MqttApi.getMqtt_Url(), clientId, dataStore);
  40.             client.setCallback(new MqttCallBack());
  41.             if (client != null) {
  42.                 try {
  43.                     client.connect(conOpt);
  44.                     flag = true;
  45.                 } catch (Exception e) {
  46.                     flag = false;
  47.                 }
  48.             }
  49.         } catch (MqttException e) {
  50.             LogUtils.e(TAG, "createConnect:MqttException: " + e.toString());
  51.         }
  52.         return flag;
  53.     }
  54.  
  55.     /**
  56.      * 连接服务端
  57.      */
  58.     public void connect() {
  59.         EventBus.getDefault().post(new MqttMainEvent("onlinestatus","离线"));
  60.         new Thread(new Runnable() {
  61.             @Override
  62.             public void run() {
  63.                 boolean isflag = createConnect(MqttApi.getClintId());
  64.                 if (isflag) {
  65.                     LogUtils.e(TAG, "连接成功...");
  66.                     isflag = subScribe();//开始订阅
  67.                     if (!isflag) {
  68.                         LogUtils.e(TAG, "订阅失败...");
  69.                     } else {
  70.                         LogUtils.e(TAG, "订阅成功...");
  71.                         while (true) {
  72.                             if (!MqttApi.isBoolean) {
  73.                                 EventBus.getDefault().post(new LoginEvent("登录失败"));
  74.                                 EventBus.getDefault().post(new MqttMainEvent("login", "登录失败"));
  75.                                 LogUtils.e(TAG, "准备登录..............");
  76.                                 boolean isPublish = publish(1, JsonPut.putLogin());//发布登录
  77.                                 if (isPublish) {
  78.                                     LogUtils.e(TAG, "登录成功...");
  79.                                 } else {
  80.                                     LogUtils.e(TAG, "登录异常...");
  81.                                     connect();
  82.                                     break;
  83.                                 }
  84.                             } else {
  85.                                 if (heartBeat == null) {
  86.                                     heartBeat = new MqttHeartBeat();
  87.                                 }
  88.                                 LogUtils.e(TAG, "=======================登录成功,退出循环=====================");
  89.                                 startReconnect();
  90.                                 break;
  91.                             }
  92.                             SystemClock.sleep(10 * 1000);
  93.                         }
  94.                     }
  95.                 } else {
  96.                     LogUtils.e(TAG, "连接失败...");
  97.                     SystemClock.sleep(5000);
  98.                     connect();
  99.                 }
  100.             }
  101.         }).start();
  102.     }
  103.  
  104.     /**
  105.      * 发布数据到服务端
  106.      *
  107.      * @param code 1：登录主题，2：心跳主题，3：业务主题
  108.      * @param data 发布内容
  109.      * @return
  110.      */
  111.     public boolean publish(int code, String data) {
  112.         String topic = "";
  113.         if (code == 1) {            //登录
  114.             topic = "/lo";
  115.         } else if (code == 2) {     //心跳
  116.             topic = "/he";
  117.         } else if (code == 3) {     //业务
  118.             topic = "/bu";
  119.         }
  120.         boolean flag = false;
  121.         if (client != null && client.isConnected()) {
  122.             MqttMessage message = new MqttMessage();
  123.             message.setPayload(data.getBytes());
  124.             message.setQos(2);//@param qos的“服务质量”使用。设为0,1,2
  125.             message.setRetained(false);
  126.             try {
  127.                 client.publish(topic, message);
  128.                 flag = true;
  129.             } catch (MqttException e) {
  130.                 LogUtils.e(TAG, "发送异常..." + e.toString());
  131.                 flag = false;
  132.             }
  133.         } else {
  134.             LogUtils.e(TAG, "当前连接已断开...");
  135.         }
  136.         return flag;
  137.     }
  138.  
  139.     /**
  140.      * 订阅主题
  141.      */
  142.     public boolean subScribe() {
  143.         boolean flag = false;
  144.         if (client != null && client.isConnected()) {
  145.             try {
  146.                 client.subscribe(MqttApi.getTopicSubscribe(), 2);
  147.                 flag = true;
  148.             } catch (MqttException e) {
  149.                 flag = false;
  150.                 LogUtils.e(TAG, "订阅异常:" + e.toString());
  151.             }
  152.         }
  153.         return flag;
  154.     }
  155.  
  156.     /**
  157.      * 断开连接
  158.      */
  159.     public void disConnect() throws MqttException {
  160.         MqttApi.isBoolean = false;
  161.         conOpt = null;
  162.        
  163.         if (client != null) {
  164.             client.disconnect();
  165.             client = null;
  166.         }
  167.         LogUtils.e(TAG, "断开连接");
  168.     }
  169.  
  170.     /**
  171.      * 释放单例, 及其所引用的资源
  172.      */
  173.     public static void release() throws Exception {
  174.         if (mInstance != null) {
  175.             mInstance.disConnect();
  176.             mInstance = null;
  177.             heartBeat.disConnect();
  178.             heartBeat = null;
  179.         }
  180.     }
  181. }

接口回调类MqttCallBack

  1. public class MqttCallBack implements MqttCallback {
  2.  
  3.     public MqttCallBack() {}
  4.  
  5.     /**
  6.      * 连接异常断开回调此方法，可在此方法内做重连机制
  7.      */
  8.     @Override
  9.     public void connectionLost(Throwable cause) {
  10.         /*SystemClock.sleep(10000);
  11.         MqttManager.getInstance().connect();*/
  12.     }
  13.  
  14.     @Override
  15.     public void messageArrived(String topic, MqttMessage message) throws Exception {
  16.         String data = new String(message.getPayload(), "UTF-8");
  17.         Object object = JSON.parse(data);
  18.     }
  19.  
  20.     @Override
  21.     public void deliveryComplete(IMqttDeliveryToken token) {
  22.         LogUtils.d(getClass().getSimpleName(), "token:" + token.toString());
  23.     }
  24. }

常量类MqttApi

  1. public class MqttApi {
  2.  
  3.     private static boolean DEBUG = true;
  4.  
  5.     /**
  6.      * 服务端IP
  7.      */
  8.     public static String getMqtt_Url() {
  9.         if (DEBUG) {
  10.             return "tcp://mqtt.inks:1883";
  11.         }
  12.         return "tcp://192.168.1.123:61613";
  13.     }
  14.  
  15.     /**
  16.      * 用户名
  17.      */
  18.     public static String getName() {
  19.         if (DEBUG) {
  20.             return "admin";//testuser
  21.         }
  22.         return "admin";
  23.     }
  24.  
  25.     /**
  26.      * 密码
  27.      */
  28.     public static String getPassword() {
  29.         if (DEBUG) {
  30.             return "shmqttfree";//sha256
  31.         }
  32.         return "password";
  33.     }
  34.  
  35.     /**
  36.      * 订阅主题
  37.      * app接收这个主题的消息
  38.      */
  39.     public static String getTopicSubscribe() {
  40.         if (DEBUG) {
  41.             return "0000000000000";
  42.         }
  43.         return "111111111111111111";
  44.     }
  45.  
  46.     /**
  47.      * 发布主题
  48.      * app发布消息的主题
  49.      */
  50.     public static String getTopicPublisher(String topic) {
  51.         if (DEBUG) {
  52.             return "/" + topic;
  53.         }
  54.         return "/" + topic;
  55.     }
  56.  
  57.     /**
  58.      * 设备ID
  59.      *
  60.      * @return
  61.      */
  62.     public static String getClintId() {
  63.         return "00000000";
  64.     }
  65.  
  66.     /**
  67.      * 标记是否登录
  68.      **/
  69.     public static volatile boolean isBoolean = false;
  70. }

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/代码探险家/article/detail/1007394
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号