基于paho在android平台上实现MQTT Client间的简单通信 - CSDN博客

在之前的博文中，对MQTT和paho进行了简单的描述。paho为实现MQTT通信提供了接口。本篇将在android平台上，基于paho实现MQTT Client间的简单通信。broker选择公共的mosquitto broker.

（0）权限

      <uses-permissionandroid:name="android.permission.ACCESS_NETWORK_STATE"/><uses-permissionandroid:name="android.permission.INTERNET"/>

没有权限，寸步难行！

（1）引入paho lib

（2）Connection Options

paho通过MqttConnectOptions类，对broker与client间的连接进行需求设置。
例如，UserName和Password（不是所有client都可以连接某一broker，但如果无登录配置，则不用设置），Timeout时间，断开后是否自动连接，是否保留客户端的连接记录等。

      privateMqttConnectOptionsinitMqttConnectionOptions(){
    MqttConnectOptions mOptions =newMqttConnectOptions();
    mOptions.setAutomaticReconnect(false);//断开后，是否自动连接mOptions.setCleanSession(true);//是否清空客户端的连接记录。若为true，则断开后，broker将自动清除该客户端连接信息mOptions.setConnectionTimeout(60);//设置超时时间，单位为秒mOptions.setUserName("Admin");//设置用户名。跟Client ID不同。用户名可以看做权限等级mOptions.setPassword("Admin");//设置登录密码mOptions.setKeepAliveInterval(60);//心跳时间，单位为秒。即多长时间确认一次Client端是否在线mOptions.setMaxInflight(10);//允许同时发送几条消息（未收到broker确认信息）mOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);//选择MQTT版本returnmOptions;
}

（3）设置回调函数

      MqttCallbackExtended mqttCallback=newMqttCallbackExtended() {@OverridepublicvoidconnectComplete(booleanreconnect, String serverURI) {
        Log.i(TAG,"connect Complete"+ Thread.currentThread().getId());
    }@OverridepublicvoidconnectionLost(Throwable cause) {
        Log.i(TAG,"connection Lost ");
    }@OverridepublicvoidmessageArrived(String topic, MqttMessage message)throwsException {if(topic.equalsIgnoreCase("subscribe topic1")){
            Log.i(TAG,"messageArrived: "+newString(message.getPayload()));
            Message msg=newMessage();
            Bundle bindle=newBundle();
            bindle.putString("Content",newString(message.getPayload()));
            msg.what=MSG_TYPE_TO_B;
            msg.setData(bindle);
            mHandler.sendMessage(msg);
        }
    }@OverridepublicvoiddeliveryComplete(IMqttDeliveryToken token) {
        Log.i(TAG,"delivery Complete ");//即服务器成功delivery消息}

};

注：需要说明的是，不能在四个回调函数中进行UI操作。该回调函数在子线程中被调用。若掺杂UI操作，会引起MQTT连接断开，触发connectionLost回调函数。

（4）MQTT Client

      mClient=initClient("tcp://test.mosquitto.org:1883","ClientA",mqttCallback,mOptions,newString[]{"topic1","topic2","topic3"});

      privateMqttClient initClient(String serverURI, String clientId,MqttCallback callback, MqttConnectOptions options,String[] subscribeTopics){
    MqttClientclient=null;try{
        MemoryPersistence persistence =newMemoryPersistence();client=newMqttClient(serverURI,clientId,persistence);client.setCallback(callback);//设置回调函数client.connect(options);//连接brokerclient.subscribe(subscribeTopics);//设置监听的topic}catch(MqttException e) {
        e.printStackTrace();
    }returnclient;
}

创建Client时，需要指明：
（a）broker的URL：tcp://test.mosquitto.org:1883，即消息的中转站在哪儿。
（b）Client的ID：ClientA，即告诉broker我是谁。
（c）MemoryPersistence：用于存储两种消息（ 可以设置为null，但与消息的发送质量有关。建议进行设置）。
（c.1）client尚未接收完毕的消息。
（c.2）client已发送完毕，但尚未得到broker确认的消息。
（d）Client的回调函数：mqttCallback，即如果连接成功，连接断开，消息发送成功，或接收到新的消息，该如何处理。
（e）Client的连接需求：mOptions，即对连接的设置。
（f）Client所监听的Topic：new String[]{“topic1”,”topic2”,”topic3”}，即仅接收以上topic信息。

（6）发送消息

      MqttMessage msg=newMqttMessage();StringmsgStr="Hello World";
msg.setPayload(msgStr.getBytes());//设置消息内容msg.setQos(2);//设置消息发送质量，可为0,1,2.msg.setRetained(false);//服务器是否保存最后一条消息，若保存，client再次上线时，将再次受到上次发送的最后一条消息。mClient.publish("my topic",msg);//设置消息的topic，并发送。

这里需要对发送质量进行再次说明。
（0）level0，最多一次的传输。不管消息是否到达broker，只要发出，client不再关心。
（1）level1，至少一次的传输。若消息到达broker，broker会回复client一个PUBACK消息，若未收到该回复消息，或超时，client将再次发送，直至收到broker响应。
（2）level2，这是最高级别的传输。在level1的基础上，保证重复消息不会被二次接收。

注：若想清空服务器保存的最后一条消息，可发送0字节的payload，对服务器进行清空。

（7）断开与broker的连接

      try{if(mClient!=null){
        mClient.disconnect();
    }
}catch(MqttException e) {
    e.printStackTrace();
}

更多详情，可在paho官方文档中查找。
https://www.ibm.com/support/knowledgecenter/SSFKSJ_7.5.0/com.ibm.mq.javadoc.doc/WMQMQxrClasses/overview-summary.html

Demo实现效果
创建两个client，A和B，A监听TopicB，B监听TopicA，实现对话。效果如下：