Android 实现MQTT客户端连接服务器、订阅主题、向指定主题发布消息。

75 阅读 0 评论 50 点赞

我是靠谱客的博主懦弱玫瑰，最近开发中收集的这篇文章主要介绍Android 实现MQTT客户端连接服务器、订阅主题、向指定主题发布消息。，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

使用MQTT实现安卓客户端首先需要在Android目录下的gradle文件中添加MQTT框架的依赖：

compile 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.1'

MQTT客户端通常实现在一个Service中(Service保活详见主页文章)，在Service的onStartCommand（）方法中需要先对MQTT进行一些初始化的工作,在一些开发需求中一般都需要通过setWill()方法设置遗嘱消息，遗嘱消息的作用是当客户端意外断开连接（比如断网）的情况下，服务器会向遗嘱消息中设定的主题发送对应的遗嘱消息，从而实现通知其他客户端该客户端已离线。参数初始化如下，其中MQTT的常用端口是":1883"，MqttClient可以通过MqttCallback对一些事件进行监听，比如收到订阅的主题的消息、成功发送一条消息和客户端断开连接，在客户端断开连接的时候可以执行重新连接。


private MqttConnectOptions mqttConnectOptions;
private MqttClient mqttClient;
/**
     * 初始化mqtt客户端参数
     */
    private void initMqtt() {
        String clientId = "客户端ID"；//客户端id是每个客户端的唯一标识
        String ip = "";//服务器IP地址
        String port = "1883";
        try {
            mqttClient = new MqttClient(
                    "tcp://"+ ip + port, clientId, new MemoryPersistence());
            mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setCleanSession(false);
            mqttConnectOptions.setUserName(Constants.MQTT_USERNAME);
            mqttConnectOptions.setPassword(Constants.MQTT_PASSWORD.toCharArray());
            mqttConnectOptions.setConnectionTimeout(10);
            mqttConnectOptions.setKeepAliveInterval(20);
            mqttConnectOptions.setWill("遗嘱消息的目标主题", "遗嘱消息的内容");
            mqttClient.setCallback(new MqttCallback() {
           
                @Override
                public void connectionLost(Throwable cause) {
                    Log.e(TAG, "连接丢失");//连接丢失的时候可以在这里进行重新连接
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) {
                     Log.i(TAG, "收到消息："+message.toString());
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    Log.i(TAG, "成功发送");
                }
            });
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

MQTT客户端初始化工作完成后就可以连接MQTT服务器了，连接需要通过try catch捕捉连接异常，从而判断是否连接成功，进而进行下一步的重连操作。

 /**
     * 连接mqtt服务器
     */
    private void connectMqtt() {
        new Thread(() -> {
            try {
                if (!(mqttClient.isConnected())) {
                    mqttClient.connect(mqttConnectOptions);
                    Log.i(TAG, "连接成功！");
                }
            } catch (Exception e) {
                Log.e(TAG, e.toString());
                
            }
        }).start();
    }

服务器连接成功后就要进行主题的订阅，像这样：

 /**
     * 订阅主题
     *
     * @param topic 主题名称
     */
    private void subscribeTopic(String topic) {
        if (mqttClient != null) {
            try {
                mqttClient.subscribe(topic, 1);
                Log.i(TAG, "订阅的主题：********");
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }

subscribe的第二个参数如果是消息等级，其代表的意义是：

qos = 0 : 消息最多发送一次，在客户端离线等不可用的情况下客户端将会丢失这条消息。

qos = 1 : 消息至少发送一次，能保证客户端能收到该条消息。

qos = 2 : 消息仅传送一次。

接下来就是向指定的主题发送消息：

 private void pushMsgToTopic(String topic, String message) {
        if (mqttClient == null || !mqttClient.isConnected()) {
            return;
        }
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setPayload(message.getBytes());
        try {
            mqttClient.publish(topic, message.getBytes(), 2, true);
            Log.i(TAG, "发送一条消息：" + message);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

publish()的第三个参数跟上面的qos同理。

MQTT实现在Service中，一般通过广播与外界进行通信，比如需要向某个主题发送消息的时候，向MQTT服务中注册的广播接收器发送广播，并把消息的容通过Intent传递到MQTT服务中进行发送。在某些情况下，MQTT客户端从一个服务器断开连接到另外一个服务器的时候，原来连接的服务器会出现不发送遗嘱的情况（我也不知道为什么），这种情况需要在断开连接之前主动发送遗嘱消息，从而实现客户端在线状态的修改。