开源客户端首次连接配置

开源客户端Java SDK下载地址

paho.mqtt.java

SDK版本

SDK依赖如下，建议使用最新版本。

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.5</version>
</dependency>

初始化客户端

创建MqttClient

final String brokerUrl = properties.getProperty("brokerUrl");
final MemoryPersistence memoryPersistence = new MemoryPersistence();
final String topic = properties.getProperty("topic");
final int qosLevel = Integer.parseInt(properties.getProperty("qos"));
final MqttClient mqttClient = new MqttClient(brokerUrl, recvClientId, memoryPersistence);
mqttClient.setTimeToWait(3000L);

mqttClient.setCallback(new MqttCallbackExtended() {
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        //连接成功回调，可以开始订阅Topic。
    }

    @Override
    public void connectionLost(Throwable throwable) {
        // 连接断开回调，建议打印日志方便定位问题。
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
        // 收到消息的回调，此次不要阻塞，且不能在这里同步发送消息，否则可能会导致死锁卡住，以及心跳无法发送导致断链。
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        // 成功发送消息到服务端。
    }
});

初始化连接项

MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
.....
connOpts.setCleanSession(cleanSession);
connOpts.setKeepAliveInterval(60); // 心跳间隔时间，单位：秒。
connOpts.setAutomaticReconnect(true); // 务必开启自动重连。
connOpts.setMaxInflight(1000);
.....

首次连接

初始化mqttClient，开始连接服务端。

mqttClient.connect(mqttConnectOptions);

首次连接不重连问题

问题现象

如果由于网络抖动或延时等其他原因导致客户端连接服务端失败，则首次连接不能自动重连。具体报错如下：

Caused by: java.net.ConnectException: Network is unreachable (connect failed)
	at java.net.PlainSocketImpl.socketConnect(Native Method)
	at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)
	at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)
	at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)
	at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)
	at java.net.Socket.connect(Socket.java:607)
	at org.eclipse.paho.client.mqttv3.internal.TCPNetworkModule.start(TCPNetworkModule.java:74)

可能原因

客户端首次连接，wasConnected=false，所以进不了connectionLost的重连处理。

if (wasConnected && callback != null) {
    // Let the user know client has disconnected either normally or abnormally.
    callback.connectionLost(reason);
}

解决方案

首次连接建议尽可能进行尝试，直至连接成功。

for(;;) {
    try {
        mqttClient.connect(mqttConnectOptions);
        break;
    } catch (Throwable e) {
        log.error("",e); // 建议打印日志，方便定位问题。
        Thread.sleep(5000L);
    }
}

客户端自动重连

异常示例

客户端连接服务端后，如果后续由于其他原因导致连接断开，会触发connectionLost。

2023-06-20 17:10:26:972	 connectionLost clientId=XXX
已断开连接 (32109) - java.net.SocketException: Operation timed out (Read failed)
	at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:197)

异常原因源码分析

public void connectionLost(MqttException cause) {
    final String methodName = "connectionLost";
    // If there was a problem and a client callback has been set inform
    // the connection lost listener of the problem.
    try {
        if (mqttCallback != null && cause != null) {
            // @TRACE 708=call connectionLost
            log.fine(CLASS_NAME, methodName, "708", new Object[] { cause });
            mqttCallback.connectionLost(cause);
        }
        if(reconnectInternalCallback != null && cause != null){
            reconnectInternalCallback.connectionLost(cause);
        }
    } catch (java.lang.Throwable t) {
        // Just log the fact that a throwable has caught connection lost 
        // is called during shutdown processing so no need to do anything else
        // @TRACE 720=exception from connectionLost {0}
        log.fine(CLASS_NAME, methodName, "720", new Object[] { t });
    }
}

触发重连

MqttReconnectCallback

public void connectionLost(Throwable cause) {
    if (automaticReconnect) {
        // Automatic reconnect is set so make sure comms is in resting
        // state
        comms.setRestingState(true);
        reconnecting = true;
        startReconnectCycle();
    }
}

MqttReconnectActionListener

重试间隔为1 s、2 s、4 s、8 s......128 s，默认最大间隔时间为128 s。

public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
    // @Trace 502=Automatic Reconnect failed, rescheduling: {0}
    log.fine(CLASS_NAME, methodName, "502", new Object[] { asyncActionToken.getClient().getClientId() });
    if (reconnectDelay < connOpts.getMaxReconnectDelay()) {
        reconnectDelay = reconnectDelay * 2;
    }
    rescheduleReconnectCycle(reconnectDelay);
}

重连成功

connectComplete回调

2023-06-20 17:12:36:764	connect success to: tcp://xxxxxx.mqtt.aliyuncs.com:1883,reconnect=true