mqtt keepAlive原理详见【http://www.emqx.com/zh/blog/mqtt-keep-alive】

# 下面开始写测试代码

【pom.xml】

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>2.3.12.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.2.2</version>
</dependency>
<dependency>
    <groupId>com.alibaba.fastjson2</groupId>
    <artifactId>fastjson2</artifactId>
    <version>2.0.49</version>
</dependency>

【MyDemo7MqttCallback.java】

package com.chz.myMqttV3.demo7;
@Slf4j
public class MyDemo7MqttCallback implements MqttCallbackExtended {
private MqttClient client;
private MqttConnectOptions options;
private String[] topics;
public MyDemo7MqttCallback(MqttClient client, MqttConnectOptions options, String[] topics)
{
this.client = client;
this.options = options;
this.topics = topics;
}
@SneakyThrows
@Override
public void connectionLost(Throwable throwable) {
log.error("connectionLost", throwable);
while (!client.isConnected()) {
log.info("emqx重新连接....................................................");
client.connect(options);
Thread.sleep(1000);
}
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
log.info("messageArrived: topic={}, message={}", topic, new String(message.getPayload()));
}
@SneakyThrows
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
if( token!=null ){
MqttMessage message = token.getMessage();
String topic = token.getTopics()==null ? null : Arrays.asList(token.getTopics()).toString();
String str = message==null ? null : new String(message.getPayload());
log.info("deliveryComplete: topic={}, message={}", topic, str);
} else {
log.info("deliveryComplete: null");
}
}
@SneakyThrows
@Override
public void connectComplete(boolean reconnect, String serverURI) {
log.info("connectComplete: reconnect={}, serverURI={}", reconnect, serverURI);
if( topics.length > 0 ){
int[] qosArr = new int[topics.length];
Arrays.fill(qosArr, 2);
MyDemo7MqttMessageListener[] listeners = new MyDemo7MqttMessageListener[topics.length];
Arrays.fill(listeners, new MyDemo7MqttMessageListener());
client.subscribe(topics, qosArr, listeners);
}
}
}

【MyDemo7MqttMessageListener.java】

package com.chz.myMqttV3.demo7;
@Slf4j
public class MyDemo7MqttMessageListener implements IMqttMessageListener
{
@Override
public void messageArrived(String topic, MqttMessage message) {
log.info("messageArrived: topic={}, message={}", topic, new String(message.getPayload()));
}
}

【MyDemo7MqttClient1Test.java】

package com.chz.myMqttV3.demo7;
public class MyDemo7MqttClient1Test
{
public static void main(String[] args) throws  MqttException {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("admin");
options.setPassword("public".toCharArray());
options.setCleanSession(true);
options.setAutomaticReconnect(true);
options.setConnectionTimeout(20);
options.setKeepAliveInterval(10);
MqttClient client = new MqttClient("tcp://192.168.44.228:1883", "MyDemo7MqttClient1Test", new MemoryPersistence());
client.setCallback(new MyDemo7MqttCallback(client, options, new String[]{"device/#"}));
client.connect(options);
}
}

【MyDemo7MqttSenderTest.java】

package com.chz.myMqttV3.demo7;
public class MyDemo7MqttSenderTest
{
public static void main(String[] args) throws UnknownHostException, MqttException, InterruptedException {
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("admin");
options.setPassword("public".toCharArray());
options.setCleanSession(true);
options.setAutomaticReconnect(true);
options.setConnectionTimeout(20);
options.setKeepAliveInterval(10);
// 这里设置遗嘱消息，当broker认为本client断联时会将遗嘱消息发出，可以用来确认borker是否认为本client已经断联
options.setWill("device/1", "I am MyDemo7MqttSenderTest, I am dead!!!".getBytes(), 1, false);
MqttClient client = new MqttClient("tcp://192.168.44.228:1883", "MyDemo7MqttSenderTest", new MemoryPersistence());
client.setCallback(new MyDemo7MqttCallback(client, options, new String[]{}));
client.connect(options);
for( int i=0; true; i++ ){
String topic = "device/1";
MqttMessage mqttMessage = new MqttMessage();
mqttMessage.setRetained(true);
String msg = "I am MyDemo7MqttSenderTest, at node [192.168.44.228:1883]:" + i;
mqttMessage.setPayload(msg.getBytes(StandardCharsets.UTF_8));
client.publish(topic, mqttMessage);
System.out.println("send: " + msg);
Thread.sleep(1000L);    // 测试运行的时候在这里加入断点，卡住MqttClient内部自动发的PINGREQ消息
}
}
}

# 下面开始进行测试

1、启动【MyDemo7MqttClient1Test】。
2、在【MyDemo7MqttSenderTest】的【sleep】那一句加入断点，然后运行【MyDemo7MqttSenderTest】

等一段时间，可以看到遗嘱消息【I am MyDemo7MqttSenderTest, I am dead!!!】发出来了