Android Mqtt协议（连接、发布、订阅）实例、亲测可用

￡神魔★判官ぃ 2023-01-06 11:38 230阅读 0赞

实现步骤：

1、项目根目录build.gradle中添加：

allprojects {
        repositories {
            ...
            maven{ url "https://repo.eclipse.org/content/repositories/paho-snapshots/"}
        }
    }

2、App目录的build.gradle中添加：

dependencies {
        ...
        //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'
    }

3、mainfests清单文件中添加权限：

<uses-permission android:name="android.permission.WAKE_LOCK" />
        <uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
        <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
        <uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" />
        <uses-permission android:name="android.permission.INTERNET" />

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

4、定义工具类MqttManager.java

import android.content.Context;
    import android.os.Handler;
    import android.os.Message;
    import android.util.Log;
    
    import org.eclipse.paho.android.service.MqttAndroidClient;
    import org.eclipse.paho.client.mqttv3.IMqttActionListener;
    import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    import org.eclipse.paho.client.mqttv3.IMqttToken;
    import org.eclipse.paho.client.mqttv3.MqttCallback;
    import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    import org.eclipse.paho.client.mqttv3.MqttMessage;
    
    /**
     * Mqtt工具
     */
    public class MqttManager {
    
        private Context mContext;
        private Handler mHandler;
        private static MqttManager mMqttManager;
        private MqttAndroidClient mMqttAndroidClient;
        private MqttConnectOptions mMqttConnectOptions;
        //MQTT相关配置
        //服务器地址（协议+地址+端口号）
        private final static String HOST = "tcp://119.168.210.120:1883";
        //用户名
        private final static String USERNAME = "test";
        //密码
        private final static String PASSWORD = "12345678";
        //发布主题
        private final static String PUBLISH_TOPIC = "/r";
        //订阅主题
        private final static String RESPONSE_TOPIC = "/b";
        //服务质量,0最多一次，1最少一次，2只一次
        private final static int QUALITY_OF_SERVICE = 0;
        //重连hand
        public final static int HANDLE_CONNECT_FAILED = 0x1000;
        public final static int HANDLE_CONNECT_BROKEN = 0x1001;
    
        /**
         * 构造函数
         * @param context
         */
        private MqttManager(Context context) {
            this.mContext = context;
            init();
        }
    
        /**
         * 单例模式
         */
        public static MqttManager getInstance(Context context) {
            if (mMqttManager == null) {
                mMqttManager = new MqttManager(context);
            }
            return mMqttManager;
        }
    
        /**
         * 初始化
         */
        private void init() {
            mMqttConnectOptions = new MqttConnectOptions();
            mMqttConnectOptions.setCleanSession(true); //设置是否清除缓存
            mMqttConnectOptions.setConnectionTimeout(10); //设置超时时间，单位：秒
            mMqttConnectOptions.setKeepAliveInterval(30); //设置心跳包发送间隔，单位：秒
            mMqttConnectOptions.setUserName(USERNAME); //设置用户名
            mMqttConnectOptions.setPassword(PASSWORD.toCharArray()); //设置密码
            // 异常断开连接前发送的信息
            String message = "{\"Dced\":\"" + 1 + "\"}";
            String topic = PUBLISH_TOPIC;
            Integer qos = QUALITY_OF_SERVICE;
            Boolean retained = false;
            try {
                //mMqttConnectOptions.setWill(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         * MQTT是否连接成功的监听
         */
        private IMqttActionListener iMqttActionListener = new IMqttActionListener() {
            @Override
            public void onSuccess(IMqttToken arg0) {
                Log.i("--->mqtt", "连接成功 ");
                try {
                    //订阅主题，参数：主题、服务质量
                    mMqttAndroidClient.subscribe(PUBLISH_TOPIC, QUALITY_OF_SERVICE);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
    
            @Override
            public void onFailure(IMqttToken arg0, Throwable arg1) {
                arg1.printStackTrace();
                Log.i("--->mqtt", "onFailure 连接失败:" + arg1.getMessage());
                sendHandlerMsg(HANDLE_CONNECT_FAILED, 0, null);
            }
        };
    
        /**
         * 订阅主题的回调
         */
        private MqttCallback mqttCallback = new MqttCallback() {
            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                Log.i("--->mqtt", "收到消息： " + new String(message.getPayload()) + "\tToString:" + message.toString());
                //收到其他客户端的消息后，响应给对方告知消息已到达或者消息有问题等
                //response("message arrived:"+message);
            }
    
            @Override
            public void deliveryComplete(IMqttDeliveryToken arg0) {
                Log.i("--->mqtt", "deliveryComplete");
            }
    
            @Override
            public void connectionLost(Throwable arg0) {
                Log.i("--->mqtt", "连接断开");
                sendHandlerMsg(HANDLE_CONNECT_BROKEN, 0, null);
            }
        };
    
        /**
         * 建立mqtt连接，连接MQTT服务器
         */
        public boolean connect(String clientId, Handler handler){
            this.mHandler = handler;
            //创建Mqtt客户端
            mMqttAndroidClient = new MqttAndroidClient(mContext, HOST, clientId);
            mMqttAndroidClient.setCallback(mqttCallback); //设置订阅消息的回调
            //建立连接
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        mMqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);
                    } catch (Exception e) {
                        Log.e("--->mqtt","connect: " + e.getMessage());
                        e.printStackTrace();
                    }
                }
            }).start();
    
    
    
    
            /*try {
                mMqttAndroidClient.connect(mMqttConnectOptions, null, iMqttActionListener);
            } catch (Exception e) {
                Log.e("--->mqtt","connect: " + e.getMessage());
                e.printStackTrace();
                return false;
            }*/
            return true;
        }
    
        /**
         * 发布消息
         */
        public void publish(String message) {
            String topic = PUBLISH_TOPIC;
            Integer qos = QUALITY_OF_SERVICE;
            Boolean retained = false;
            try {
                if (mMqttAndroidClient != null && mMqttAndroidClient.isConnected()) {
                    //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
                    mMqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
                } else {
                    Log.i("--->mqtt","mqttAndroidClient is Null or is not connected");
                }
            } catch (Exception e) {
                Log.i("--->mqtt","publish MqttException:" + e.getMessage());
                e.printStackTrace();
            }
        }
    
        /**
         * 响应订阅消息
         */
        public void response(String message) {
            String topic = RESPONSE_TOPIC;
            Integer qos = QUALITY_OF_SERVICE;
            Boolean retained = false;
            try {
                //参数分别为：主题、消息的字节数组、服务质量、是否在服务器保留断开连接后的最后一条消息
                mMqttAndroidClient.publish(topic, message.getBytes(), qos.intValue(), retained.booleanValue());
            } catch (Exception e) {
                Log.i("--->mqtt","publish:" + e.getMessage());
                e.printStackTrace();
            }
        }
    
        /**
         * 断开链接
         */
        public void disconnect() {
            try {
                if (mMqttAndroidClient != null) {
                    mMqttAndroidClient.unregisterResources();
                    mMqttAndroidClient.disconnect();
                    mMqttAndroidClient = null;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 发送handler消息
         */
        private void sendHandlerMsg(int arg1, int arg2, Object obj){
            if(mHandler == null){
                return;
            }
            Message message = new Message();
            message.arg1 = arg1;
            message.arg2 = arg2;
            message.obj = obj;
            mHandler.sendMessage(message);
        }
    
    }

5、Activity.java中使用

private MqttManager m_mqttManager;

/**
         * 初始化mqtt协议
         */
        private void initMqttManager(){
            m_mqttManager = MqttManager.getInstance(this);
        }

m_mqttManager.connect("gardstest097", m_handler);

m_mqttManager.publish("test");

完！！！