MQTT

admin管理员组

文章数量:1794759

MQTT

Eclipse paho mqtt源码分析

• • MQTT
  • paho mqtt
  • 源码分析
  • • org.eclipse.paho.client.mqttv3.MqttClient

MQTT

MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。
MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

以上说明来源与百度：百度关于MQTT的解释说明

paho mqtt

paho mqtt是IBM根据MQTT协议编写的client jar

源码分析

org.eclipse.paho.client.mqttv3.MqttClient

MqttClient的类图，我们发现其非常简单，就是实现了IMqttClient接口类

我们主要看connect()方法，我们发现MqttClient中的connect()方法调用的是MqttAsyncClient类中的connect()方法，我们在一路像下走，会发现真正的连接方法在ClientComms类中

	/** @see IMqttClient#connect(MqttConnectOptions)*/public void connect(MqttConnectOptions options) throws MqttSecurityException, MqttException {aClient.connect(options, null, null).waitForCompletion(getTimeToWait());}

在MqttAsyncClient的connect方法中，程序会创建对应的NetworkModule来进行与服务端的连接。NetworkModule一共有4种，分别是：TCPNetworkModule、SSLNetworkModule、WebSocketSecureNetworkModule、WebSocketNetworkModule。具体选择哪种实现，是通过Uri的secheme来实现。具体进行connect连接是在ClientComms的connect方法中。

	/** (non-Javadoc)* * @see* org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(org.eclipse.paho.* client.mqttv3.MqttConnectOptions, java.lang.Object,* org.eclipse.paho.client.mqttv3.IMqttActionListener)*/public IMqttToken connect(MqttConnectOptions options, Object userContext, IMqttActionListener callback)throws MqttException, MqttSecurityException {final String methodName = "connect";if (comms.isConnected()) {throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_CONNECTED);}if (comms.isConnecting()) {throw new MqttException(MqttException.REASON_CODE_CONNECT_IN_PROGRESS);}if (comms.isDisconnecting()) {throw new MqttException(MqttException.REASON_CODE_CLIENT_DISCONNECTING);}if (comms.isClosed()) {throw new MqttException(MqttException.REASON_CODE_CLIENT_CLOSED);}if (options == null) {options = new MqttConnectOptions();}this.connOpts = options;this.userContext = userContext;final boolean automaticReconnect = options.isAutomaticReconnect();// @TRACE 103=cleanSession={0} connectionTimeout={1} TimekeepAlive={2}// userName={3} password={4} will={5} userContext={6} callback={7}log.fine(CLASS_NAME, methodName, "103",new Object[] { Boolean.valueOf(options.isCleanSession()), Integer.valueOf(options.getConnectionTimeout()),Integer.valueOf(options.getKeepAliveInterval()), options.getUserName(),((null == options.getPassword()) ? "[null]" : "[notnull]"),((null == options.getWillMessage()) ? "[null]" : "[notnull]"), userContext, callback });//创建network module 并对comms设置networkmodules，此处通过spi技术，加载所有的NetworkModuleFactory接口实现，并通过uri的scheme进行匹配，来选择对应的NetworkModuleFactory，这里我们以TCPNetworkModuleFactory为例//networkmodule有：//org.eclipse.paho.client.mqttv3.internal.TCPNetworkModuleFactory//org.eclipse.paho.client.mqttv3.internal.SSLNetworkModuleFactory//org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketNetworkModuleFactory//org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketSecureNetworkModuleFactory				comms.setNetworkModules(createNetworkModules(serverURI, options));comms.setReconnectCallback(new MqttReconnectCallback(automaticReconnect));// Insert our own callback to iterate through the URIs till the connect// succeedsMqttToken userToken = new MqttToken(getClientId());ConnectActionListener connectActionListener = new ConnectActionListener(this, persistence, comms, options,userToken, userContext, callback, reconnecting);userToken.setActionCallback(connectActionListener);userToken.setUserContext(this);// If we are using the MqttCallbackExtended, set it on the// connectActionListenerif (this.mqttCallback instanceof MqttCallbackExtended) {connectActionListener.setMqttCallbackExtended((MqttCallbackExtended) this.mqttCallback);}comms.setNetworkModuleIndex(0);connectActionListener.connect();return userToken;}

我们来看createNetworkModules方法的实现，它本身并没有什么复杂的逻辑。主要是构建参数，并调用createNetworkModule方法，在createNetworkModule方法种，它有调用了NetworkModuleService类中的createInstance方法。

public static NetworkModule createInstance(String address, MqttConnectOptions options, String clientId)throws MqttException, IllegalArgumentException{try {URI brokerUri = new URI(address);//应用RFC3986的权限补丁applyRFC3986AuthorityPatch(brokerUri);String scheme = brokerUri.getScheme().toLowerCase();//NetworkModule是通过Factory来进行创建的，我们在看FACTORY_SERVICE_LOADER属性，发现它用到了SPI机制，什么是SPI呢，我们会在后面的博客中进行讲解for (NetworkModuleFactory factory : FACTORY_SERVICE_LOADER) {if (factory.getSupportedUriSchemes().contains(scheme)) {return factory.createNetworkModule(brokerUri, options, clientId);}}/** To throw an IllegalArgumentException exception matches the previous behavior of* MqttConnectOptions.validateURI(String), but it would be nice to provide something more meaningful.*/throw new IllegalArgumentException(brokerUri.toString());} catch (URISyntaxException e) {throw new IllegalArgumentException(address, e);}}

在这里我们分析一下ClientComms的connect方法，在此方法中并不会真证进行tcp/ip的连接，具体的连接是通过内部类（ConnectBG）启动线程的方式进行创建。我们可以发现在paho mqtt中，基本上都是通过线程进行启动的。在线程中会同时启动：CommsReceiver（数据接收）、CommsSender（发送）、CommsCallback（回调）。

	private class ConnectBG implements Runnable {ClientComms 	clientComms = null;MqttToken 		conToken;MqttConnect 	conPacket;private String threadName;ConnectBG(ClientComms cc, MqttToken cToken, MqttConnect cPacket, ExecutorService executorService) {clientComms = cc;conToken 	= cToken;conPacket 	= cPacket;threadName = "MQTT Con: "+getClient().getClientId();}void start() {if (executorService == null) {new Thread(this).start();} else {executorService.execute(this);}}public void run() {Thread.currentThread().setName(threadName);final String methodName = "connectBG:run";MqttException mqttEx = null;//@TRACE 220=>log.fine(CLASS_NAME, methodName, "220");try {// Reset an exception on existing delivery tokens.// This will have been set if disconnect occurred before delivery was// fully processed.MqttDeliveryToken[] toks = tokenStore.getOutstandingDelTokens();for (MqttDeliveryToken tok : toks) {tok.internalTok.setException(null);}// Save the connect token in tokenStore as failure can occur before sendtokenStore.saveToken(conToken,conPacket);// Connect to the server at the network level e.g. TCP socket and then// start the background processing threads before sending the connect// packet.NetworkModule networkModule = networkModules[networkModuleIndex];//这里真正进行tcp/ip的连接//启动网络模块，进行网络连接networkModule.start();//数据的接收，启动完成后，负责从broker接收消息receiver = new CommsReceiver(clientComms, clientState, tokenStore, networkModule.getInputStream());receiver.start("MQTT Rec: "+getClient().getClientId(), executorService);//数据发送，启动完成后，负责发送消息到brokersender = new CommsSender(clientComms, clientState, tokenStore, networkModule.getOutputStream());sender.start("MQTT Snd: "+getClient().getClientId(), executorService);//回调服务，在连接时，指定的Callback服务callback.start("MQTT Call: "+getClient().getClientId(), executorService);//发送一条遗言消息，如果在连接客户端代码中有设置的话internalSend(conPacket, conToken);} catch (MqttException ex) {//@TRACE 212=connect failed: unexpected exceptionlog.fine(CLASS_NAME, methodName, "212", null, ex);mqttEx = ex;} catch (Exception ex) {//@TRACE 209=connect failed: unexpected exceptionlog.fine(CLASS_NAME, methodName, "209", null, ex);mqttEx =  ExceptionHelper.createMqttException(ex);}if (mqttEx != null) {//如果发现异常，进行连接的shutdown操作shutdownConnection(conToken, mqttEx);}}}

以上就是paho mqtt连接MQTT服务器的流程，代码其实非常简单。

本文标签： MQTT