MQTT协议特点

MQTT是一个由IBM主导开发的物联网传输协议，它被设计用于轻量级的发布/订阅式消息传输，旨在为低带宽和不稳定的网络环境中的物联网设备提供可靠的网络服务。它的核心设计思想是开源、可靠、轻巧、简单，具有以下主要的几项特性：

1. 非常小的通信开销（最小的消息大小为 2 字节）；
2. 支持各种流行编程语言（包括C，Java，Ruby，Python 等等）且易于使用的客户端；
3. 支持发布 / 预定模型，简化应用程序的开发；
4. 提供三种不同消息传递等级，让消息能按需到达目的地，适应在不稳定工作的网络传输需求

应项目要求使用MQTT协议实现客户端与服务端通信。Android端使用MqttAndroidClient实现MQTT通信：MqttAndroidClient

MqttAndroidClient配置

1.添加依赖

在项目根目录下的build.gradle中添加：

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

在app目录下的build.gradle中添加：

dependencies {
    implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'
    implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
}

2.权限声明

在AndroidManifest.xml中添加：

    <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_PHONE_STATE" />
    <uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.INTERNET" />
    <uses-permission android:name="android.permission.ACCESS_WIFI_STATE" />

3.配置服务

在AndroidManifest.xml中添加：

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

功能实现

一、简单实现

    private String clientId = "ExampleAndroidClient";
    private MqttAndroidClient mqttAndroidClient;
    private String host = "tcp://xx.xx.xx.xx";//MQTT服务器地址
    private MqttConnectOptions conOpt;
    private String name;//MQTT账号
    private String password;//MQTT密码
    private String topic;//订阅的标题


    //MQTT连接
    private void setMqtt() {
        clientId = clientId + System.currentTimeMillis();
        mqttAndroidClient = new MqttAndroidClient(this,host,clientId);
        conOpt = new MqttConnectOptions();
        // 清除缓存
        conOpt.setCleanSession(true);
        // 自动重连
        conOpt.setAutomaticReconnect(true);
        // 设置超时时间，单位：秒
        conOpt.setConnectionTimeout(10);
        // 心跳包发送间隔，单位：秒
        conOpt.setKeepAliveInterval(20);
        // 用户名
        conOpt.setUserName(name);
        // 密码
        conOpt.setPassword(password.toCharArray());
        mqttAndroidClient.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                Log.d(TAG, "connectionLost: 连接断开");
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                Log.d(TAG, "消息到达" + message.getPayload());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {

            }
        });
        try {
            //进行连接
            mqttAndroidClient.connect(conOpt, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.d(TAG, "onSuccess: 连接成功");
                    try {
                        //连接成功后订阅主题
                        mqttAndroidClient.subscribe(topic, 2);

                    } catch (MqttException e) {
                        e.printStackTrace();
                    }

                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.d(TAG, "onFailure: 连接失败");
                }
            });

        } catch (MqttException e) {
            e.printStackTrace();
        }
    }


    public void publishMessage(String payload) {
        try {
            if (mqttAndroidClient.isConnected() == false) {
                mqttAndroidClient.connect();
            }

            MqttMessage message = new MqttMessage();
            message.setPayload(payload.getBytes());
            message.setQos(0);
            mqttAndroidClient.publish(topic, message, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.i(TAG, "publish succeed!");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.i(TAG, "publish failed!");
                }
            });
        } catch (MqttException e) {
            Log.e(TAG, e.toString());
            e.printStackTrace();
        }
    }

二、通过service实现

1.新建MqttService

public class MqttService extends Service {

    public static final String TAG = MqttService.class.getSimpleName();

    private static MqttAndroidClient client;
    private MqttConnectOptions conOpt;
    private IGetMessageCallBack mIGetMessageCallBack;

    private String host = "tcp://xx.xx.xx.xx";
    private String userName = "admin";
    private String passWord = "password";
    private String topic;
    private String clientId = "AndroidClient_";

    @Override
    public void onCreate() {
        super.onCreate();
        init();
    }

    private void init() {
        clientId = clientId + System.currentTimeMillis();
        client = new MqttAndroidClient(this, host, clientId);
        conOpt = new MqttConnectOptions();
        client.setCallback(mqttCallback);
        // 清除缓存
        conOpt.setCleanSession(true);
        // 自动重连
        conOpt.setAutomaticReconnect(true);
        // 设置超时时间，单位：秒
        conOpt.setConnectionTimeout(10);
        // 心跳包发送间隔，单位：秒
        conOpt.setKeepAliveInterval(20);
        conOpt.setUserName(userName);
        conOpt.setPassword(passWord .toCharArray());
        doClientConnection();
    }

    @Override
    public IBinder onBind(Intent intent) {
        return new CustomBinder();
    }

    public class CustomBinder extends Binder {
        public MqttService getService(){
            return MqttService.this;
        }
    }

    public void setIGetMessageCallBack(IGetMessageCallBack iGetMessageCallBack) {
        this.mIGetMessageCallBack = iGetMessageCallBack;
    }

    /** 连接MQTT服务器 */
    private void doClientConnection() {
        if (!client.isConnected() && isConnectIsNormal()) {
            try {
                client.connect(conOpt, null, iMqttActionListener);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

    }

    public static void publishMessage(String payload) {
        try {
            if (client.isConnected() == false) {
                client.connect();
            }

            MqttMessage message = new MqttMessage();
            message.setPayload(payload.getBytes());
            message.setQos(0);
            client.publish(topic, message,null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.i(TAG, "publish succeed!");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.i(TAG, "publish failed!");
                }
            });
        } catch (MqttException e) {
            Log.e(TAG, e.toString());
            e.printStackTrace();
        }
    }

    // MQTT是否连接成功
    private IMqttActionListener iMqttActionListener = new IMqttActionListener() {
        @Override
        public void onSuccess(IMqttToken asyncActionToken) {
            Log.d(TAG, "onSuccess: 连接成功");
            try {
                //连接成功后订阅主题
                client.subscribe(topic, 2);

            } catch (MqttException e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
            exception.printStackTrace();
        }
    };

    // MQTT监听并且接受消息
    private MqttCallback mqttCallback = new MqttCallback() {
        @Override
        public void connectionLost(Throwable cause) {
            Log.d(TAG, "connectionLost: 连接断开");
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            String str = new String(message.getPayload());
            if (mIGetMessageCallBack != null){
                mIGetMessageCallBack.setMessage(str);
            }
            Log.d(TAG, "消息到达");
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {

        }
    };

    /** 判断网络是否连接 */
    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 void onDestroy() {
        stopSelf();
        try {
            client.disconnect();
        } catch (MqttException e) {
            e.printStackTrace();
        }
        super.onDestroy();
    }

2.新建IGetMessageCallBack,接收服务端发送的消息

public interface IGetMessageCallBack {
    public void setMessage(String message);
}

3.新建MyServiceConnection

public class MyServiceConnection implements ServiceConnection {

    private MqttService mqttService;
    private IGetMessageCallBack iGetMessageCallBack;

    @Override
    public void onServiceConnected(ComponentName name, IBinder service) {
        mqttService =  ((MqttService.CustomBinder)service).getService();
        mqttService.setIGetMessageCallBack(iGetMessageCallBack);
    }

    @Override
    public void onServiceDisconnected(ComponentName name) {

    }

    public MqttService getMqttService() {
        return mqttService;
    }

    public void setIGetMessageCallBack(IGetMessageCallBack iGetMessageCallBack) {
        this.iGetMessageCallBack = iGetMessageCallBack;
    }

4.使用

public class MainActivity extends AppCompatActivity implements IGetMessageCallBack {


    private MyServiceConnection myServiceConnection;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        myServiceConnection = new MyServiceConnection();
        myServiceConnection.setIGetMessageCallBack(this);
        startMqttService();
        //向服务器发送消息
        //MqttService.publishMessage("Hello!");
    }

    
    private void startMqttService() {
        Intent intent = new Intent(this, MqttService.class);
        bindService(intent, myServiceConnection, Context.BIND_AUTO_CREATE);
    }

    @Override
    public void setMessage(String message) {
        Log.d(TAG, "获取的消息" + message);
    }
}