Java-MQTT客户端监控连接状态事件

约定不等于承诺〃 2021-06-24 13:57 481阅读 0赞

MQTT客户端监控连接状态事件

项目采用[*paho*.*mqtt*.*java*][paho_._mqtt_._java]客户端，需要监控连接状态变更事件，以进行异常维测和处理。

代码中提供了MqttCallback接口如下：

org.eclipse.paho.client.mqttv3-1.2.0-sources.jar!\\org\\eclipse\\paho\\client\\mqttv3\\MqttClient.java

public void setCallback(MqttCallback callback) {
       aClient.setCallback(callback);
    }

其中MqttCallback 接口定义如下：

public interface MqttCallback {
 /** 连接断连
 * This method is called when the connection to the server is lost.
 *
 * @param cause the reason behind the loss of connection.
 */
 public void connectionLost(Throwable cause);
 
 /** 消息收到回调的缺省回调，如果subscribe(String topicFilter, int qos, IMqttMessageListener messageListener)指定了 IMqttMessageListener 会覆盖该回调，也即：使用IMqttMessageListener 回调，MqttCallback 中的messageArrived回调无响应了
 * This method is called when a message arrives from the server.
 *
 * 
 * This method is invoked synchronously by the MQTT client. An
 * acknowledgment is not sent back to the server until this
 * method returns cleanly.
 * 
 * If an implementation of this method throws an <code>Exception</code>, then the
 * client will be shut down. When the client is next re-connected, any QoS
 * 1 or 2 messages will be redelivered by the server.
 * 
 * Any additional messages which arrive while an
 * implementation of this method is running, will build up in memory, and
 * will then back up on the network.
 * 
 * If an application needs to persist data, then it
 * should ensure the data is persisted prior to returning from this method, as
 * after returning from this method, the message is considered to have been
 * delivered, and will not be reproducible.
 * 
 * It is possible to send a new message within an implementation of this callback
 * (for example, a response to this message), but the implementation must not
 * disconnect the client, as it will be impossible to send an acknowledgment for
 * the message being processed, and a deadlock will occur.
 *
 * @param topic name of the topic on the message was published to
 * @param message the actual message.
 * @throws Exception if a terminal error has occurred, and the client should be
 * shut down.
 */
 public void messageArrived(String topic, MqttMessage message) throws Exception;
 
 /** 消息publish完成回调
 * Called when delivery for a message has been completed, and all
 * acknowledgments have been received. For QoS 0 messages it is
 * called once the message has been handed to the network for
 * delivery. For QoS 1 it is called when PUBACK is received and
 * for QoS 2 when PUBCOMP is received. The token will be the same
 * token as that returned when the message was published.
 *
 * @param token the delivery token associated with the message.
 */
 public void deliveryComplete(IMqttDeliveryToken token);
 }

可以通过connectionLost回调进行断连提示，但是重连的状态回调呢？

（说明：可以通过mqttClient.connect(mqttConnectOptions);中的

mqttConnectOptions.setAutomaticReconnect(true); // 自动重连

设置自动重连）

哈哈。[*paho*.*mqtt*.*java*][paho_._mqtt_._java]用了一个很有意思且巧妙的方法处理，如下：

/**
     * Extension of {@link MqttCallback} to allow new callbacks
     * without breaking the API for existing applications.
     * Classes implementing this interface can be registered on
     * both types of client: {@link IMqttClient#setCallback(MqttCallback)}
     * and {@link IMqttAsyncClient#setCallback(MqttCallback)}
     */
    public interface MqttCallbackExtended extends MqttCallback {
       
       /**
        * Called when the connection to the server is completed successfully.
        * @param reconnect If true, the connection was the result of automatic reconnect.
        * @param serverURI The server URI that the connection was made to.
        */
       public void connectComplete(boolean reconnect, String serverURI);
    
    }
    
    使用时调用：mqttClient.setCallback(callback); 传入的callback采用 MqttCallbackExtended 的实现类就可以得到 connectComplete回调了。具体实现可以详见代码。
    一句话：MqttCallbackExtended 应该是后期扩展的接口，为了保持接口不变及已有代码的功能，就对原有MqttCallback接口进行了继承扩展，内部实现检测到传入的callback也是MqttCallbackExtended 的实例，就会回调 connectComplete 接口了。
    // If we are using the MqttCallbackExtended, set it on the
    // connectActionListener
    if (this.mqttCallback instanceof MqttCallbackExtended) {
       connectActionListener.setMqttCallbackExtended((MqttCallbackExtended) this.mqttCallback);
    }

可以谓之为： “扩展模式”吧！

补充说明一点：MqttCallback 中的messageArrived回调是消息收到回调的缺省回调，如果subscribe(String topicFilter, int qos, IMqttMessageListener messageListener)指定了 IMqttMessageListener 会覆盖该回调，也即：使用IMqttMessageListener 回调，MqttCallback 中的messageArrived回调无响应了

/* (non-Javadoc)
     * @see org.eclipse.paho.client.mqttv3.IMqttClient#subscribe(java.lang.String, int)
     */
    public void subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException {
       this.subscribe(new String[] {topicFilter}, new int[] {qos}, new IMqttMessageListener[] {messageListener});
    }

具体原因：详见代码。

org.eclipse.paho.client.mqttv3-1.2.0-sources.jar!\\org\\eclipse\\paho\\client\\mqttv3\\internal\\CommsCallback.java

protected boolean deliverMessage(String topicName, int messageId, MqttMessage aMessage) throws Exception
    {     
       boolean delivered = false;
       
       Enumeration keys = callbacks.keys();
       while (keys.hasMoreElements()) {
          String topicFilter = (String)keys.nextElement();
          if (MqttTopic.isMatched(topicFilter, topicName)) {
             aMessage.setId(messageId);
             ((IMqttMessageListener)(callbacks.get(topicFilter))).messageArrived(topicName, aMessage);
             delivered = true;
          }
       }
       
       /* if the message hasn't been delivered to a per subscription handler, give it to the default handler */
       if (mqttCallback != null && !delivered) {
          aMessage.setId(messageId);
          mqttCallback.messageArrived(topicName, aMessage);
          delivered = true;
       }
       
       return delivered;
    }

OK。验证如下：

taskClient.getMqttClient().setCallback(new MqttCallbackExtended() {
        @Override
        public void connectComplete(boolean reconnect, String serverURI) {
            AlLog.getLogger().warn("TaskClient connectComplete reconnect={} serverURI={}", reconnect, serverURI);
        }
    
        @Override
        public void connectionLost(Throwable cause) {
            AlLog.getLogger().error("TaskClient connectionLost " + taskClient.isConnected() + " cause:", cause);
    
        }
    
        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            AlLog.getLogger().debug("TaskClient messageArrived message:{}", message);
        }
    
        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            AlLog.getLogger().debug("TaskClient deliveryComplete token:{}", token);
        }
    });

日志：

02-28 14:45:09.668 E/29252.64,MQTT Ping: devXXX SubscribeTask$1.connectionLost.41: TaskClient connectionLost false cause:  
02-28 14:45:10.738 W/29252.63,MQTT Call: devXXX SubscribeTask$1.connectComplete.63: TaskClient connectComplete reconnect=true serverURI=tcp://XXX.XXX.XXX.XXX:1883  
02-28 14:45:26.514 W/29252.35,Timer-0 SubscribeTask$3.run.93: TimerTask taskClient isConnected true

[paho_._mqtt_._java]: http://www.baidu.com/link?url=lqBZgwumCOlZ3SUySZZaGS55nNM3IzA5zj_uj3SV_ONEb0T3h-KExdAxROMdX9NH1SjBZssynsfKMgvtGrEtAq