MQTT Java问题：接收指定数量消息后终止连接，函数提前返回求助

你的问题核心在于MQTT客户端的订阅回调是异步执行的：调用mqttClient.subscribe()后，代码会立刻走到return true，完全没等待回调里的计数达到目标值，所以函数提前返回了。

修复方案：用同步阻塞工具等待计数达标 #

可以用CountDownLatch实现阻塞等待，直到收到足够数量的匹配消息后再返回结果。修改后的代码示例如下：

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

@Override
public Boolean call() throws Exception {
    // 根据目标计数初始化CountDownLatch
    CountDownLatch latch = new CountDownLatch(this.goal);
    mqttClient.subscribe(couple.getTopic().getUrl(), (topic, message) -> {
        String payload = new String(message.getPayload());
        System.out.println(payload + "=" + couple.getMessage().getJson());
        if (payload.equals(couple.getMessage().getJson())) {
            System.out.println("Match");
            latch.countDown(); // 每匹配一次，计数器减1
            System.out.println("剩余需要匹配的数量: " + latch.getCount());
            if (latch.getCount() == 0) {
                mqttClient.unsubscribe(couple.getTopic().getUrl());
            }
        } else {
            System.out.println("Not match");
        }
    });
    
    // 阻塞等待，直到计数器归0，或者超时（可根据需求调整超时时间）
    boolean reachedGoal = latch.await(60, TimeUnit.SECONDS);
    // 如果超时还没达标，主动取消订阅
    if (!reachedGoal) {
        mqttClient.unsubscribe(couple.getTopic().getUrl());
    }
    return reachedGoal;
}

关键说明 #

• CountDownLatch的作用是让call()方法所在线程阻塞，直到回调中调用足够次数的countDown()，使计数器变为0。
• await()方法设置了超时时间，避免无限等待，超时后返回false，你可以根据业务需求调整超时时长或处理逻辑。
• 改用CountDownLatch的计数器能避免线程安全问题——回调是在MQTT客户端的独立线程中执行，和call()方法线程不同，直接操作类成员变量可能出现计数不准确的情况。

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