Android上简化使用Mqtt client

关于Mqtt

Mqtt3.1.1中文文档
这是一个客户端服务端架构的发布/订阅模式的消息传输协议；轻巧、开放、简单、规范，易于实现，国内很多关于Mqtt的文章都是讲推送的，其实Mqtt远远不止这个功能，具体详情可自行文档查阅。

场景说明

在Android上，往往具体的业务逻辑要回到Activity中，而Mqtt连接则是建立在service中（考虑到内存资源及退出应用后续操作），同时Mqtt的消息透传回调也是在service中。
通常做法是要写很多类（通常是Boardcast）来实现确保service同Activity组件之间的通信，就像Eclipse paho的Android service一样。

改进方案

本文基于Eclipse的paho框架的java client端，在Android上引入EventBus来做事件分发，简化了代码量，同时方便实现。
以下是我写的一个简单的工具操作类：

public class MqttManager {
    // 单例
    private static MqttManager mInstance = null;
    // 回调
    private MqttCallback mCallback;
    // Private instance variables
    private MqttClient client;
    private MqttConnectOptions conOpt;
    private boolean clean = true; 
    private MqttManager() {
        mCallback = new MqttCallbackBus();
    }
    public static MqttManager getInstance() {
        if (null == mInstance) {
            mInstance = new MqttManager();
        }
        return mInstance;
    }
    /**     * 释放单例, 及其所引用的资源     */
    public static void release() {
        try {
            if (mInstance != null) {
                mInstance.disConnect();
                mInstance = null;
            }
        } catch (Exception e) {
        }
    }
    /**
     * 创建Mqtt 连接
     *
     * @param brokerUrl Mqtt服务器地址(tcp://xxxx:1863)
     * @param userName  用户名
     * @param password  密码
     * @param clientId  clientId
     * @return
     */
    public boolean creatConnect(String brokerUrl, String userName, String password, String clientId) {
        boolean flag = false;
        String tmpDir = System.getProperty("java.io.tmpdir");
        MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);
        try {
            // Construct the connection options object that contains connection parameters
            // such as cleanSession and LWT
            conOpt = new MqttConnectOptions();
            conOpt.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
            conOpt.setCleanSession(clean);
            if (password != null) {
                conOpt.setPassword(password.toCharArray());
            }
            if (userName != null) {
                conOpt.setUserName(userName);
            }
            // Construct an MQTT blocking mode client
            client = new MqttClient(brokerUrl, clientId, dataStore);
            // Set this wrapper as the callback handler
            client.setCallback(mCallback);
            flag = doConnect();
        } catch (MqttException e) {
            Logger.e(e.getMessage());
        }
        return flag;
    }
    /**
     * 建立连接
     *
     * @return
     */
    public boolean doConnect() {
        boolean flag = false;
        if (client != null) {
            try {
                client.connect(conOpt);
                Logger.d("Connected to " + client.getServerURI() + " with client ID " + client.getClientId());
                flag = true;
            } catch (Exception e) {
            }
        }
        return flag;
    }
    /**
     * Publish / send a message to an MQTT server
     *
     * @param topicName the name of the topic to publish to
     * @param qos       the quality of service to delivery the message at (0,1,2)
     * @param payload   the set of bytes to send to the MQTT server
     * @return boolean
     */
    public boolean publish(String topicName, int qos, byte[] payload) {
        boolean flag = false;
        if (client != null && client.isConnected()) {
            Logger.d("Publishing to topic \"" + topicName + "\" qos " + qos); 
           // Create and configure a message
            MqttMessage message = new MqttMessage(payload);            message.setQos(qos);
            // Send the message to the server, control is not returned until
            // it has been delivered to the server meeting the specified
            // quality of service.
            try {
                client.publish(topicName, message);
                flag = true;
            } catch (MqttException e) {
            }
        }
        return flag;
    }
    /**
     * Subscribe to a topic on an MQTT server
     * Once subscribed this method waits for the messages to arrive from the server
     * that match the subscription. It continues listening for messages until the enter key is
     * pressed.
     *
     * @param topicName to subscribe to (can be wild carded)
     * @param qos       the maximum quality of service to receive messages at for this subscription
     * @return boolean
     */
    public boolean subscribe(String topicName, int qos) {
        boolean flag = false;
        if (client != null && client.isConnected()) {
            // Subscribe to the requested topic
            // The QoS specified is the maximum level that messages will be sent to the client at.
            // For instance if QoS 1 is specified, any messages originally published at QoS 2 will
            // be downgraded to 1 when delivering to the client but messages published at 1 and 0
            // will be received at the same level they were published at.
            Logger.d("Subscribing to topic \"" + topicName + "\" qos " + qos);
            try {
                client.subscribe(topicName, qos);
                flag = true;
            } catch (MqttException e) {
            }
        }
        return flag;
    }
    /**
     * 取消连接
     *
     * @throws MqttException
     */
    public void disConnect() throws MqttException {
        if (client != null && client.isConnected()) {  
          client.disconnect();
        }
    }
}

在Mqtt Callback方法中发送Evenet事件：

public class MqttCallbackBus implements MqttCallback {
    @Override
    public void connectionLost(Throwable cause) {
        Logger.e(cause.getMessage());
    }
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        Logger.d(topic + "====" + message.toString());
        EventBus.getDefault().post(message);
    }
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
    }
}

问题

1. 由于EventBus不支持跨进程通信，所以当在service是独立进程时，无法在主进程中接收到EventBus分发的事件。
   可以考虑通过广播做中转，即在Mqtt的回调中发送一个广播，在广播中进行事件分发。

• 连接等操作是较费时的，为了避免ANR，应在子线程中完成操作，也可以考虑使用RxJava来简化。

Demo

https://github.com/LichFaker/MqttClientAndroid