如何用Eclipse Paho同步获取MQTT保留消息或默认值？求非超时方案

我可以通过以下Eclipse Paho代码获取MQTT保留消息：

client.subscribe(myRetainedMessageTopic, (topic, message) -> {
  this.state = message.toString();
  client.unsubscribe(topic);
});

但该方案存在两个问题：

• 无法保证myRetainedMessageTopic主题存在保留消息
• 若无消息则会一直等待监听器执行完成，无法及时处理默认情况

我尝试了临时方案，通过硬编码睡眠等待：

AtomicBoolean myTopicPresent = new AtomicBoolean(false);
client.subscribe(myRetainedMessageTopic, (topic, message) -> {
  this.state = message.toString();
  myTopicPresent.set(true);
});
Thread.sleep(500);
client.unsubscribe(myRetainedMessageTopic);
if(!myTopicPresent.get()) {
  this.state = DEFAULT_STATE;
  client.publish(myRetainedMessageTopic, retainedMqttMessageWithDefaultState);
}

但这种方案不够规范，是否存在无需随机超时的更优同步实现方式？

最优同步实现方案 #

可以利用Java标准并发工具类实现可控的同步等待，彻底规避Thread.sleep()的不确定性，推荐两种成熟方案：

方案1：使用CountDownLatch #

CountDownLatch专为单事件或多事件的同步等待场景设计，逻辑清晰且易于维护：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

// 初始化计数器为1，对应"收到保留消息"这一个事件
CountDownLatch latch = new CountDownLatch(1);
AtomicBoolean hasRetainedMsg = new AtomicBoolean(false);

client.subscribe(myRetainedMessageTopic, (topic, message) -> {
  this.state = message.toString();
  hasRetainedMsg.set(true);
  latch.countDown(); // 收到消息后触发计数器减1，唤醒等待线程
});

try {
    // 等待最多2秒（可根据业务场景调整超时时间）
    boolean isMsgReceivedInTime = latch.await(2, TimeUnit.SECONDS);
    client.unsubscribe(myRetainedMessageTopic);
    
    // 超时未收到消息，或收到但标记异常时，设置默认状态并发布默认保留消息
    if (!isMsgReceivedInTime || !hasRetainedMsg.get()) {
        this.state = DEFAULT_STATE;
        client.publish(myRetainedMessageTopic, retainedMqttMessageWithDefaultState);
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    // 根据业务需求处理中断异常，比如重置状态或记录日志
}

方案2：使用CompletableFuture #

Java 8+提供的CompletableFuture更适合异步结果的同步获取，代码更简洁：

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

CompletableFuture<String> retainedMsgFuture = new CompletableFuture<>();

client.subscribe(myRetainedMessageTopic, (topic, message) -> {
    retainedMsgFuture.complete(message.toString());
    client.unsubscribe(topic);
});

try {
    // 等待最多2秒获取结果，超时则抛出TimeoutException
    String receivedState = retainedMsgFuture.get(2, TimeUnit.SECONDS);
    this.state = receivedState;
} catch (Exception e) {
    // 超时、中断或执行异常，均判定为未收到保留消息
    this.state = DEFAULT_STATE;
    client.publish(myRetainedMessageTopic, retainedMqttMessageWithDefaultState);
    client.unsubscribe(myRetainedMessageTopic);
}

方案核心优势 #

• 可控超时：不再依赖固定的500ms延迟，而是根据业务允许的最大等待时间设置（比如2秒），既避免无限等待，也不会因网络波动漏收消息
• 规范并发：基于Java标准并发工具实现，代码可读性、可维护性更强，符合工业级开发规范
• 及时处理：无论是否收到保留消息，都能在指定时间内进入后续业务逻辑，不会阻塞线程

内容的提问来源于stack exchange，提问作者Martin