MQTT

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 处理程序中执行较长的 running/blocking 任务，因为这是在客户端的主网络循环上调用的。

如果您有很长的 running/blocking 任务来处理一条消息，您应该创建一个本地队列并使用单个本地线程（如果消息顺序很重要）或线程池来处理来自该队列的消息如果您想尽快处理传入的消息。

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

MQTT - Paho IMqttMessageLIstener 在侦听器线程因操作而被阻塞时错过消息

MQTT - Paho IMqttMessageLIstener misses messages when the listener thread is blocked for an operation

java

paho