云消息队列 MQTT 版：客户端连接常见问题

使用安卓的Demo，返回连接失败报错。

问题描述

• 资源填写正确，且网络可以连通。

• AK（AccessKey ID）、SK（AccessKey Secret）填写正确且已授权。

可能原因

接入点格式不正确，正确格式为“tcp://Endpoint:1883”或“tls://Endpoint:8883”。

为什么客户端能连接上，消费端一订阅就马上断连？

为什么控制台显示的订阅关系数比实际使用的高很多？

当cleanSession设置为false时，客户端离线后订阅关系也会一直保留，不会被清理。

若您不需要保留离线订阅关系，您可以使用订阅关系清理功能，自动清理离线客户端的订阅关系。

客户端一直断线重连，并且返回“invalid param”错误。

可能原因：消息体超过限制，服务端拒绝连接。云消息队列 MQTT 版中，消息体不能超过64 KB。

业务侧无改动，客户端和服务端连接突然断开。

可能原因：

• 设备关机、网络不通导致客户端连接断开。

• 若您使用的是Token鉴权方式，Token到期后连接会断开，重新建立连接时客户端未获取新的Token也会导致连接失败。

客户端心跳时间在哪里设置？

调用MqttConnectOptions类。

String clientId = mqttTools.getClientId();
mqttConnectOptions = new MqttConnectOptions();
// 阿里云账号AccessKey拥有所有API的访问权限，建议您使用RAM用户进行API访问或日常运维。
// 强烈建议不要把AccessKey ID保存到工程代码里，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。
// 本示例以将AccessKey保存在环境变量为例说明。
mqttConnectOptions.setUserName("Signature|" + System.getenv("MQTT_AK_ENV") + "|" + instanceId);
mqttConnectOptions.setPassword(mqttTools.macSignature(clientId).toCharArray());
mqttConnectOptions.setCleanSession(true);
// 客户端心跳超时时间，默认90秒。
mqttConnectOptions.setKeepAliveInterval(90);
mqttConnectOptions.setAutomaticReconnect(true);
mqttConnectOptions.setMqttVersion(MQTT_VERSION_3_1_1);
mqttConnectOptions.setConnectionTimeout(5000);
mqttLogger.info("MQTT is connect setting={},clientId={}", JSON.toJSONString(mqttConnectOptions), clientId);
mqttClient = new MqttClient("tcp://" + endPoint + ":1883", clientId, memoryPersistence);
mqttClient.setTimeToWait(5000);
mqttClient.setCallback(this.getMqttCallbackExtended());
mqttClient.connect(mqttConnectOptions);

客户端连接时出现“valid owner failed”报错

设备偶尔出现“connection reset by peer”错误。

客户端ID（ClientID）重复时会互踢，导致服务端断开连接，而客户端继续往TCP连接中发送数据，会触发对端Socket发送RST报文。

请确保您设置的客户端ID全局唯一。更多信息，请参见客户端限制。

如何配置MQTT开源Java SDK的客户端自动连接

开启SDK客户端自动重连功能

mqttConnectOptions = new MqttConnectOptions();
mqttConnectOptions.setAutomaticReconnect(true);

将autoReconnect参数设置为true后，云消息队列 MQTT 版的Java SDK客户端会自动重连。

mqttClient.setCallback(new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                /**
                 * 客户端连接成功后就需要尽快订阅的Topic。
                 */
                System.out.println("connect success");
            }

            @Override
            public void connectionLost(Throwable throwable) {
                throwable.printStackTrace();
            }

            @Override
            public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                /**
                 * 消费消息的回调接口，需要确保该接口不抛异常，该接口运行返回即代表消息消费成功。
                 * 消费消息需要保证在规定时间内完成，如果消费耗时超过服务端约定的超时时间，对于可靠传输的模式，服务端可能会重试推送，业务需要做好幂等去重处理。
                 * 超时时间请参见使用限制。
                 */
                System.out.println(
                    "receive msg from topic " + s + " , body is " + new String(mqttMessage.getPayload()));

            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                System.out.println("send msg succeed topic is : " + iMqttDeliveryToken.getTopics()[0]);
            }
});

这样客户端的回调中的connectComplete方法被调用时就代表连接成功或重连成功，业务方无需关心连接情况。这时候重新建立的连接与之前的连接已经不是一个，因此之前客户端的内部连接上的消息发送订阅等行为会失效。

此时需要重新订阅Topic，这样业务侧在新的内部连接中可以继续发送订阅消息。

@Override
public void connectComplete(boolean reconnect, String serverURI) {
    /**
    * 客户端连接成功后就需要尽快订阅需要的Topic。
    */
    System.out.println("connect success");
    executorService.submit(new Runnable() {
        @Override
        public void run() {
            try {
                final String topicFilter[] = {mq4IotTopic};
                final int[] qos = {qosLevel};
                mqttClient.subscribe(topicFilter, qos);
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    });
}

业务侧手动维护客户端连接

mqttConnectOptions = new MqttConnectOptions();
mqttConnectOptions.setAutomaticReconnect(false);

当autoReconnect参数设置为false时，云消息队列 MQTT 版的客户端断开后不会自动重连。

当连接断开后，业务侧会抓取到异常，需要手动将旧的连接关闭，然后重新建立新的连接。