java - MQTT - Paho IMqttMessageLIstener 当监听器线程因操作而被阻止时会丢失消息

Question

我有一个简单的 MQTT 监听器，用于订阅主题并回调

MqttClient client = new MqttClient(mqttHost, MqttClient.generateClientId());
            client.connect();    
            client.subscribe("test", QUALITY_OF_SERVICE_2, new IMqttMessageListener() {
                public void messageArrived(final String s, final MqttMessage mqttMessage) {

System.out.println("Received"+mqttMessage.toString());
// Code that blocks the thread 
lock.lock();
//do something
lock.unlock();

});

假设我正在向主题测试发布 1000 条消息，但在 tomcat 上运行上述监听器将显示 < 1000 条控制台输出，表明接收器线程未获取所有发送的消息。

如果没有 lock() 代码，监听器将按预期工作并接收所有消息。

Answer 1

您不应该在 messageArrived 处理程序中执行长时间运行/阻塞的任务，因为这是在客户端的主网络循环上调用的。

如果您有长时间运行/阻塞的任务来处理消息，您应该创建一个本地队列，并使用单个本地线程(如果消息顺序很重要)或线程池(如果您想要)处理来自该队列的消息。尽快处理传入的消息。

Java 有一组内置的核心类，用于构建队列并启动线程来使用这些队列中的消息。查看java.util.concurrent包中的类。