不知不觉已经跟MQTT打交道半年了，才惊醒的发现我也算从事的物联网方法(Android端)，一直以来只是单纯的使用MQTT连接、发布和接收消息等，也没在乎其Client(Android)端的源码是怎样实现的，要不是最近项目出现一个小问题困扰了很久可能也不会引发我看一看MQTT的源码实现。好啦让我们开始了解MQTT的神奇之处吧。(注：若有什么地方阐述有误，敬请指正。)

阅读本文前，默认读者已经熟知MQTT的Android端使用，Client代表客户端，Broker代表服务端，此篇源码分析主要以MQTT客户端和服务端建立连接过程为主线讲解。基础了解Mqtt报文格式等，可以参考下MQTT协议官网中文地址：

mcxiaoke.gitbooks.io/mqtt-cn/con…

org.eclipse.paho工程源码分析涉及到的类：

为方便分析源码先贴上一段工程里连接MQTT的代码：

根据上述代码可以看出，MQTT的连接是由MqttAndroidClient的connect函数而起，MqttAndroidClient对象初始化时传入了uri和mClientId，调用connect函数时传入了符合自己需求的Mqtt连接选项，因此我们先来看下MqttAndroidClient的connect函数：

这个函数会启动paho mqtt唯一一个组件MqttService，这个组件不支持跨进程调用，如果需要将MqttService放在其他进程，需要将和mqtt相关的调用全部放在同一个进程内。由于需要使用MqttService组件中的函数，需要在启动MqttService后对MqttService进行绑定。如果服务已经启动，则直接执行建立连接操作。这时候建立的连接仅仅是网络连接，不是mqtt协议连接。由上面代码可以看出，无论是MqttService是否启动并绑定，最终都是调用doConnect()方法继续执行连接操作。

直到此时出现了activityToken, connectToken, clientHandle，不要慌，我们来一个一个分析。MqttAndroidClient的connect函数时，会生成connectToken，具体生成如下:

token机制在paho mqtt实现中扮演着十分重要的角色，负责消息各种回调的实现，后面章节再单独分析paho mqtt的token机制。再来看一下clientHandle的来源：

clientHandle是serverURI + ":" + clientId+":"+contextId组合形成的字符串，contextId是应用包名。此段代码中引入了一个新的类MqttConnection，而MqttConnection代表着Mqtt的连接实例，MqttService内部使用connections记录每一个连接实例。最后了解下activityToken，我们看下storeToken(connectToken)函数：

MqttAndroidClient内部使用tokenMap记录每次调用生成的token, 将tokenNumber返回。activityToken会传入MqttConnection中，并保存于MqttConnection类中connect函数的Bundle变量resultBundle里，而resultBundle最终会被用于发送广播触发我们connect、publish、subscribe等的回调监听。这里暂时先了解这些，我们接着看执行完doConnect函数后，函数调用到了MqttService组件中的connect函数：

看到clientHandle是不是有点熟悉，上面我们讲过connections将生成的MqttConnection实例保存起来，这一步通过getConnection重新获取。接下来，代码来到了MqttConnection.connect函数中：

从上面代码以及注释中可知，这段代码主要作用就是新建了MqttAsyncClient对象，然后注册了回调函数，然后去执行connect函数，同时将状态置为正在连接状态。接下来就分析下MqttAsyncClient.connect函数：

MqttAsyncClient.connect函数的主要作用是设置了网络连接模块，设置重连回调，最后执行connectActionListener.connect函数。这段代码又引进来一个新的类ClientComms，我们先来看下ClientComms的初始化：

可以看出ClientComms是在MqttAsyncClient初始化时完成初始化的，并且将心跳的发送器pingSender和大小为10的线程池一起传入ClientComms。ClientComms类的初始化中又初始化了CommsTokenStore、CommsCallback和ClientState几个类。我们再来看下重连回调，重连代码有点多，我们只关注一下重连的回调函数即可：

自动重连的实现主要在的attemptReconnect()方法里，重连失败会继续重连直到连接成功，不过重连的间隔时间会随着重连次数增加最大到128s。最后我们再分析一下网络连接的设置createNetworkModules函数：

options实例在建立连接的过程中，我们也仅仅是设置了和连接相关的一些状态，并没有设置serverURI，故options.getServerURIS返回为null。NetworkModule为paho定义的接口，规定了网络模块需要实现的方法。目前paho定义的网络连接模块有TCPNetworkModule，SSLNetworkModule，WebsocketNetworkModule，WebSocketSecureNetworkModule，可以看下createNetworkModule根据uri使用的协议类型创建对应的NetworkModule。创建完所有的NetworkModule后，执行comms.setNetworknModule(0)，先使用第一个NetworkModule进行连接。comms是ClientComms类型的实例，在paho的实现中占有非常重要的地位，后序部分会进行分析。来看下createNetwokModule函数的实现：

可以看出，createNetwokModule函数主要是根据serverURIType来判断需要使用TCPNetworkModule，SSLNetworkModule，WebsocketNetworkModule，WebSocketSecureNetworkModule中的那种网络模块实现网络连接。

现在可以继续往下继续分析connectActionListener.connect()函数啦：

从这段代码中可以看出，连接已交给comms.connect(options, token)函数，而comms的初始化上面也提到过，ClientComms是在MqttAsyncClient初始化时完成初始化的

从comms.connect函数的代码中可以看出，最后调用了conbg.start()函数，而ConnectBG是实现了Runnable的类，并且运行在线程池中：

receiver收到消息后，响应消息的消息类型为MqttAck，由于CONACK数据包是MqttAck类型，且token不为null，故会执行clientState.notifyReceivedAck函数.

notifyReceivedAck函数中，处理各种broker返回消息，而连接消息处理最后会到connected()连接完成的方法中，该方法设置连接完成状态以及开始发送心跳。 至此，MQTT连接源码分析已完成。

现在我们回头看一下前面说的resultBundle，前面说到resultBundle最终会被用于发送广播触发我们connect、publish、subscribe等的回调监听。我们先取一处简单说明一下，前面也说到MqttConnection.connect函数中IMqttActionListener listener用于监听连接成功的回调。

简单说明下listener调用过程：listener会被传入MqttAsyncClient类里，随后又通过初始化ConnectActionListener类并保存于其成员变量userCallback中，最后是在ConnectActionListener里的onSuccess和onFailure两回调方法中调用了listener的onSuccess和onFailure两个方法，而ConnectActionListener里的onSuccess和onFailure两函数一般是CommsCallback类所调用（也会被MqttTokenAndroid类的notifyComplete函数调用，notifyComplete函数被MqttAndroidClient类的simpleAction和disconnected两方法调用，而simpleAction函数又会被连接、订阅、解除订阅、发送等调用，暂时只简单说一下这种情况）。上面代码注释中也说过CommsCallback是接收器和外部API之间的桥接。此类由Receiver调用，然后将以comms为中心的MQTT消息对象转换为外部API可理解的。 CommsReceiver接收器里轮询会调用ClientState.notifyReceivedAck((MqttAck)message);函数，该函数里有几种消息会调用notifyResult(ack, token, mex);函数，notifyResult方法对调用(CommsCallback)callback.asyncOperationComplete(token);对CommsCallback里成员变量completeQueue(Vector)进行addElement操作，而CommsCallback的run方法又是一直轮询监听completeQueue里是否有元素，有则调用handleActionComplete方法--》fireActionEvent方法--》ConnectActionListener里的onSuccess和onFailure。 大致流程参考如图：

listener的onSuccess函数中里调用了doAfterConnectSuccess(resultBundle);

连接成功后会调用MqttService.callbackToActivity()，resultBundle就作为其中一个参数被传入，接下来我们看看这个方法的实现：

callbackToActivity()方法用于发送本地广播，广播中携带resultBundle，其实包括publish、subscribe等行为不论成功失败都会调用此方法，发出一个指示行为类型及状态的本地广播。那么发送的广播是在哪接收的呢？其实前面一直没有说MqttAndroidClient类继承自BroadCastReceiver，因此我们查看其onReceive()方法：

data.getString(MqttServiceConstants.CALLBACK_ACTION)获取的就是我们前面存放在resultBundle中的action,然后根据action去调用对应的方法去回调callback的对应方法，例如：action为MESSAGE_ARRIVED_ACTION时，调用messageArrivedAction函数，如果需要监听action为MqttServiceConstants.MESSAGE_ARRIVED_ACTION的行为，则要求我们传入的callback必须为MqttCallback的实现，而如果需要监听action为MqttServiceConstants.CONNECT_EXTENDED_ACTION的行为，则要求我们传入的callback必须为MqttCallbackExtended的实现，MqttCallbackExtended是MqttCallback的子类。这里的callback就是我们建立连接前初始化MqttAndroidClient时设置的MqttCallback对象：

至此，分析完连接MQTT的源码，下一篇分析MQTT发布消息publish。

参考链接：

blog.csdn.net/rockstore/a…

blog.csdn.net/Dovar_66/ar…

...

(注：若有什么地方阐述有误，敬请指正。)