Java paho mqtt 接收消息慢

Question

Java 接收mqtt的消息，消息是每一秒发四条，当数据内容更改后，需要大约一分钟之后才能收到更改后的数据，用emqx观察 满订阅 显示队列满了，队列设置的1000，飞行窗口设置的32，消息流出几百万条 大部分都丢失了，怎么解决这个问题

Answer 1

引自免费微信小程序：皆我百晓生

针对您所描述的Java Paho MQTT接收消息慢，以及消息堆积、丢包严重的问题，我为您提供以下详细的解决方案：

1. 优化MQTT客户端配置：

   • 增加QoS级别：如果当前使用的QoS级别较低（如QoS 0），尝试提高到QoS 1或2，以确保消息至少被送达一次或保证精确送达一次。这将有助于减少消息丢失。

   MqttConnectOptions options = new MqttConnectOptions();
   options.setCleanSession(false); // 保持会话以利用QoS功能
   options.setAutomaticReconnect(true); // 自动重连以维持连接稳定
   options.setConnectionTimeout(60); // 设置合理的连接超时时间
   options.setKeepAliveInterval(30); // 调整心跳间隔，保持与服务器的有效通信
   
   mqttClient.connect(options);
   

   • 调整订阅选项：确保在订阅时正确设置了QoS级别，与发送端一致。

   mqttClient.subscribe(topic, QoS.LEAST_ONE); // 示例：使用QoS 1订阅
   

2. 优化服务器端配置：

   • 增大消息队列容量：既然您提到队列设置为1000且已满，可根据实际需求适当增大队列容量，以容纳更多未处理的消息。但请注意，这只是治标不治本，若消费速度跟不上生产速度，长期来看仍可能导致消息堆积。

   • 调整飞行窗口大小：飞行窗口（Inflight Window Size）控制了客户端同时可以处理的最大未确认消息数。将其适当增大，可允许客户端并行处理更多消息，从而提升消息处理效率。

   • 监控服务器性能：检查服务器CPU、内存、磁盘I/O等资源使用情况，确保服务器有足够的处理能力。必要时考虑升级硬件或横向扩展部署。

   • 检查网络延迟：确保客户端与服务器之间的网络连接稳定，低延迟。若存在网络波动或高延迟，可能影响消息的及时接收和处理。

3. 优化客户端消息处理逻辑：

   • 异步处理消息：确保您的消息回调方法（MqttCallback#messageArrived）内逻辑高效、非阻塞，尽可能减少处理单个消息所需的时间。如有耗时操作，应将其放入单独的线程池执行。

   mqttClient.setCallback(new MqttCallback() {
       @Override
       public void messageArrived(String topic, MqttMessage message) throws Exception {
           executorService.submit(() -> {
               // 异步处理消息逻辑
           });
       }
   
       // 其他回调方法...
   });
   

   • 批量处理消息：如果消息间关联性较小，可以考虑使用Mqtt5BlockingClient或Mqtt5AsyncClient实现批量接收和处理消息，以提高处理效率。

   // 使用Mqtt5BlockingClient示例
   Mqtt5BlockingClient blockingClient = Mqtt5Client.builder()
           .serverHost("your.server.com")
           .buildBlocking();
   
   blockingClient.toAsync().subscribeWith().topicFilter("your/topic").send()
           .thenCompose(subAck -> blockingClient.publishes(MqttGlobalPublishFilter.ALL)
                   .filter(publish -> publish.getTopic().equals("your/topic"))
                   .collectList()
                   .timeout(Duration.ofSeconds(1)) // 可调整批量接收的超时时间
                   .doOnNext(publishes -> {
                       // 批量处理消息逻辑
                   })
                   .then())
           .join();
   

4. 监控与报警：

   • 客户端监控：记录客户端接收消息的时间戳，对比发送端发布时间，以量化消息延迟情况。设置阈值触发报警，以便及时发现并排查问题。

   • 服务器端监控：通过EMQX提供的监控工具或API，实时监控消息队列长度、飞行窗口状态、消息丢弃率等关键指标，设置报警规则，以便及时介入处理。

通过上述步骤，您可以从客户端、服务器端以及消息处理逻辑等多个层面优化MQTT消息接收速度，减少消息堆积和丢包现象。请根据实际情况逐步调整各项参数，并持续监控系统性能，以达到理想的处理效果。