添加权限

 	<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"/>