MQTT之Eclipse.Paho源码(一)--建立连接

public class MqttPahoMessageHandler extends AbstractMqttMessageHandler
    implements MqttCallback, ApplicationEventPublisherAware {
  /**
   * The default completion timeout in milliseconds.
   */
   // 默认的发送完成等待超时时间
  public static final long DEFAULT_COMPLETION_TIMEOUT = 30_000L;


  // 最长发送完成超时时间，默认30秒
  private long completionTimeout = DEFAULT_COMPLETION_TIMEOUT;


  @Override
  protected void publish(String topic, Object mqttMessage, Message<?> message) {
    Assert.isInstanceOf(MqttMessage.class, mqttMessage, "The 'mqttMessage' must be an instance of 'MqttMessage'");
    try {
      // 在这里创建连接并且发送消息
      IMqttDeliveryToken token = checkConnection().publish(topic, (MqttMessage) mqttMessage);
      // 如果是非异步的，会阻塞等待一条消息发送完成（发送成功并且收到相应的回执，这个取决于qos）
      // 如果是异步的，会通过事件回调的方式t通知发送完成，非阻塞
      if (!this.async) {
        token.waitForCompletion(this.completionTimeout); // NOSONAR (sync)
      }
      else if (this.asyncEvents && this.applicationEventPublisher != null) {
        this.applicationEventPublisher.publishEvent(
            new MqttMessageSentEvent(this, message, topic, token.getMessageId(), getClientId(),
                getClientInstance()));
      }
    }
    catch (MqttException e) {
      throw new MessageHandlingException(message, "Failed to publish to MQTT in the [" + this + ']', e);
    }
  }
}

IMqttDeliveryToken token = checkConnection().publish(topic, (MqttMessage) mqttMessage);

  private synchronized IMqttAsyncClient checkConnection() throws MqttException {
    // 检查客户端对象是否是连接状态或client对象非空
    // 是 -> 将客户端关闭，释放资源
    if (this.client != null && !this.client.isConnected()) {
        this.client.setCallback(null);
        this.client.close();
        this.client = null;
    }
    // 如果客户端对象为空，获取异步客户端对象，并用创建好的连接对象进行连接操作
    if (this.client == null) {
      try {
        MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
        Assert.state(this.getUrl() != null || connectionOptions.getServerURIs() != null,
            "If no 'url' provided, connectionOptions.getServerURIs() must not be null");
        // 通过客户端工厂调用MQTTAsyncClient的构造方法创建一个客户端实例
        this.client = this.clientFactory.getAsyncClientInstance(this.getUrl(), this.getClientId());
        // 客户端实例计数器+1
        incrementClientInstance();
        // 设置回调
        this.client.setCallback(this);
        // 通过阻塞线程建立连接，保证连接建立成功后再执行其他操作
        this.client.connect(connectionOptions).waitForCompletion(this.completionTimeout);
        logger.debug("Client connected");
      }
      catch (MqttException e) {
        if (this.client != null) {
          this.client.close();
          this.client = null;
        }
        if (this.applicationEventPublisher != null) {
          this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, e));
        }
        throw new MessagingException("Failed to connect", e);
      }
    }
    return this.client;
  }

public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence,
      MqttPingSender pingSender, ScheduledExecutorService executorService) throws MqttException {
    // Count characters, surrogate pairs count as one character.
    // clientId长度计算
    int clientIdLength = 0;
    for (int i = 0; i < clientId.length() - 1; i++) {
      if (Character_isHighSurrogate(clientId.charAt(i)))
        i++;
      clientIdLength++;
    }
    // 如果长度 > 65536 处理异常
    if (clientIdLength > 65535) {
      throw new IllegalArgumentException("ClientId longer than 65535 characters");
    }
    // 检查连接Broker URL的合法性
    NetworkModuleService.validateURI(serverURI);


    this.serverURI = serverURI;
    this.clientId = clientId;


    this.persistence = persistence;
    // 创建持久化对象，如果未设置，默认使用内存持久化
    if (this.persistence == null) {
      this.persistence = new MemoryPersistence();
    }
    this.executorService = executorService;
    // 创建持久化对象的容器 -> 即HashTable
    this.persistence.open(clientId, serverURI);
    // 创建消息发送，接收处理器对象
    this.comms = new ClientComms(this, this.persistence, pingSender, this.executorService);
    // 清空内存持久容器数据
    this.persistence.close();
    this.topics = new Hashtable();


  }

  public ClientComms(IMqttAsyncClient client, MqttClientPersistence persistence, MqttPingSender pingSender, ExecutorService executorService) throws MqttException {
    // 设置当前状态为未连接状态
    this.conState = DISCONNECTED;
    this.client   = client;
    this.persistence = persistence;
    // 心跳包发送器，在创建MqttAsyncClient时创建的，类型为TimerPingSender
    this.pingSender = pingSender;
    // 心跳包发送器初始化
    this.pingSender.init(this);
    this.executorService = executorService;
    
    // 创建用于追踪发送消息的追踪器对象
    this.tokenStore = new CommsTokenStore(getClient().getClientId());
    // 创建回调对象
    this.callback   = new CommsCallback(this);
    // 初始化消息状态处理对象。
    this.clientState = new ClientState(persistence, tokenStore, this.callback, this, pingSender);
    
    callback.setClientState(clientState);
  }

protected ClientState(MqttClientPersistence persistence, CommsTokenStore tokenStore, 
      CommsCallback callback, ClientComms clientComms, MqttPingSender pingSender) throws MqttException {
    
    // messageId使用记录映射
    inUseMsgIds = new Hashtable();
    // 特殊消息处理队列
    pendingFlows = new Vector();
    // qos2类型消息的出站队列
    outboundQoS2 = new Hashtable();
    // qos1类型消息的出站队列
    outboundQoS1 = new Hashtable();
    // qos0类型消息的出站队列
    outboundQoS0 = new Hashtable();
    // qos2类型消息的入站队列
    inboundQoS2 = new Hashtable();
    // 初始化PingREQ对象，用于发送心跳包
    pingCommand = new MqttPingReq();
    // 针对qos2的消息，用于记录发出Rel但未收到Comp响应的计数器
    inFlightPubRels = 0;
    // 实际正在发出或接受，但未完成（针对qos1或qos2）回执处理的消息计数
    actualInFlight = 0;
    
    this.persistence = persistence;
    this.callback = callback;
    this.tokenStore = tokenStore;
    this.clientComms = clientComms;
    this.pingSender = pingSender;
    // 加载持久化的数据.
    restoreState();
  }

public IMqttToken connect(MqttConnectOptions options, Object userContext, IMqttActionListener callback)
      throws MqttException, MqttSecurityException {


    // 判断连接状态，如果是已连接/正在连接/正在断开连接/已关闭这几种状态的话进行异常处理
    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();
    // 根据不同的连接url创建网络模块
    comms.setNetworkModules(createNetworkModules(serverURI, options));
    comms.setReconnectCallback(new MqttReconnectCallback(automaticReconnect));


    // Insert our own callback to iterate through the URIs till the connect
    // succeeds
    MqttToken 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
    // connectActionListener
    if (this.mqttCallback instanceof MqttCallbackExtended) {
      connectActionListener.setMqttCallbackExtended((MqttCallbackExtended) this.mqttCallback);
    }


    comms.setNetworkModuleIndex(0);
    // 通过监听器建立连接
    connectActionListener.connect();
    return userToken;
  }

  protected NetworkModule[] createNetworkModules(String address, MqttConnectOptions options)
      throws MqttException, MqttSecurityException {


    NetworkModule[] networkModules = null;
    String[] serverURIs = options.getServerURIs();
    String[] array = null;
    if (serverURIs == null) {
      array = new String[] { address };
    } else if (serverURIs.length == 0) {
      array = new String[] { address };
    } else {
      array = serverURIs;
    }
    // 根据传入的url集合长度建立网络模块集合
    networkModules = new NetworkModule[array.length];
    for (int i = 0; i < array.length; i++) {
      // 取出对应的url和options（连接设置），进一步创建
      networkModules[i] = createNetworkModule(array[i], options);
    }
    return networkModules;
  }

  // FACTORY_SERVICE_LOADER对应NetworkModuleFactory.class和NetworkModuleService.class.getClassLoader()的集合
private static final ServiceLoader<NetworkModuleFactory> FACTORY_SERVICE_LOADER = ServiceLoader.load(
      NetworkModuleFactory.class, NetworkModuleService.class.getClassLoader());
  
  public static NetworkModule createInstance(String address, MqttConnectOptions options, String clientId)
      throws MqttException, IllegalArgumentException
  {
    try {
      URI brokerUri = new URI(address);
      applyRFC3986AuthorityPatch(brokerUri);
      // 获取到url的scheme
      String scheme = brokerUri.getScheme().toLowerCase();
      // 通过scheme匹配不同的NetWorkModule实现创建不同的网络模块
      for (NetworkModuleFactory factory : FACTORY_SERVICE_LOADER) {
        // 获取scheme，TLS对应ssl://，TCP对应tcp：//
        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);
    }
  }

String[] hostArray = new String[]{"ssl://127.0.0.1:8883","tcp://127.0.0.1:1883"};
mqttConnectOptions.setServerURIs(hostArray);

public void connect(MqttConnectOptions options, MqttToken token) throws MqttException {
    synchronized (conLock) {
      if (isDisconnected() && !closePending) {
       // 将连接状态置为CONNECTING
        conState = CONNECTING;
        // 首次连接mqtt时，发送Connect控制报文。
        conOptions = options;
                MqttConnect connect = new MqttConnect(client.getClientId(),
                        conOptions.getMqttVersion(),
                        conOptions.isCleanSession(),
                        conOptions.getKeepAliveInterval(),
                        conOptions.getUserName(),
                        conOptions.getPassword(),
                        conOptions.getWillMessage(),
                        conOptions.getWillDestination());


                this.clientState.setKeepAliveSecs(conOptions.getKeepAliveInterval());
                this.clientState.setCleanSession(conOptions.isCleanSession());
                this.clientState.setMaxInflight(conOptions.getMaxInflight());


        tokenStore.open();
        // 创建一个后台线程进行连接操作
        ConnectBG conbg = new ConnectBG(this, token, connect, executorService);
        // 线程开启
        conbg.start();
      }
      else {
        if (isClosed() || closePending) {
          throw new MqttException(MqttException.REASON_CODE_CLIENT_CLOSED);
        } else if (isConnecting()) {
          throw new MqttException(MqttException.REASON_CODE_CONNECT_IN_PROGRESS);
        } else if (isDisconnecting()) {
          throw new MqttException(MqttException.REASON_CODE_CLIENT_DISCONNECTING);
        } else {
          throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_CONNECTED);
        }
      }
    }
  }

public void run() {
      Thread.currentThread().setName(threadName);
      MqttException mqttEx = null;
      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 send
        tokenStore.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];
        networkModule.start();
        // 创建线程用于消息接受
        receiver = new CommsReceiver(clientComms, clientState, tokenStore, networkModule.getInputStream());
        // 线程执行
        receiver.start("MQTT Rec: "+getClient().getClientId(), executorService);
        // 创建线程用于消息发送
        sender = new CommsSender(clientComms, clientState, tokenStore, networkModule.getOutputStream());
        // 线程执行
        sender.start("MQTT Snd: "+getClient().getClientId(), executorService);
        // 执行回调线程
        callback.start("MQTT Call: "+getClient().getClientId(), executorService);
        // 发送Conn消息
        internalSend(conPacket, conToken);
      } catch (MqttException ex) {


        mqttEx = ex;
      } catch (Exception ex) {


        mqttEx =  ExceptionHelper.createMqttException(ex);
      }
      // 如果有异常，断开连接
      if (mqttEx != null) {
        shutdownConnection(conToken, mqttEx);
      }
    }

  public void start() throws IOException, MqttException {
    try {
     
      SocketAddress sockaddr = new InetSocketAddress(host, port);
      // 创建socket对象
      socket = factory.createSocket();
      // 通过socket进行连接
      socket.connect(sockaddr, conTimeout*1000);
      socket.setSoTimeout(1000);
    }
    catch (ConnectException ex) {
      throw new MqttException(MqttException.REASON_CODE_SERVER_CONNECT_ERROR, ex);
    }
  }

protected void notifyReceivedAck(MqttAck ack) throws MqttException {
    if (token == null) {
        // ....
     } else if (ack instanceof MqttConnack) {
      // 收到connack类型的消息，如果返回码是0，即为成功。
      int rc = ((MqttConnack) ack).getReturnCode();
      if (rc == 0) {
        synchronized (queueLock) {
          if (cleanSession) {
            clearState();
            // Add the connect token back in so that users can be  
            // notified when connect completes.
            tokenStore.saveToken(token,ack);
          }
          // 初始化计数器为0
          inFlightPubRels = 0;
          actualInFlight = 0;
          // 将出站数据通过类型不同划分到发送队列和特殊消息队列中进行等待发送
          restoreInflightMessages();
          // 通知心跳发送器向broker发送心跳，并监控broker活跃
          connected();
        }
      } else {
        mex = ExceptionHelper.createMqttException(rc);
        throw mex;
      }
      clientComms.connectComplete((MqttConnack) ack, mex);
      notifyResult(ack, token, mex);
      tokenStore.removeToken(ack);


      // Notify the sender thread that there maybe work for it to do now
      synchronized (queueLock) {
        queueLock.notifyAll();
      }
    } else {
      notifyResult(ack, token, mex);
      releaseMessageId(ack.getMessageId());
      tokenStore.removeToken(ack);
    }
    checkQuiesceLock();
  }