2

我在 Spring 中使用 HiveMQ MQTT 客户端来接收 MQTT 消息。

我的客户端配置如下所示

public Mqtt3AsyncClient mqtt3Client() {
    var mqtt3Client = Mqtt3Client.builder()
            .serverHost("my.host")
            .sslWithDefaultConfig()
            .serverPort(0000)
            .automaticReconnectWithDefaultConfig()
            .buildBlocking();

    mqtt3Client.connect();
    return mqtt3Client.toAsync();
}

客户端可用后，使用客户端初始化另一个 Spring Bean。它订阅了一个主题：

@PostConstruct
public void subscribeTopic() {
    mqtt3AsyncClient.subscribeWith()
            .topicFilter("topicfilter")
            .qos(MqttQos.AT_LEAST_ONCE)
            .callback(message -> {
                /*Handle message*/
            })
            .send()
            .whenComplete((mqtt3SubAck, throwable) -> {
                if (throwable != null) {
                    /*Logging*/
                } else {
                    /*Logging*/
                }
            });
}

我多次看到没有更多消息传递到我的应用程序，而我仍然能够使用客户端连接发送消息（因此当时它已连接）。

我找不到有关 HiveMQ MQTT 客户端如何处理配置的automaticReconnectWithDefaultConfig(). 谁能指出，我创建的订阅是否subscribeTopic()被重新订阅？我还找到了addSubscription()可以替换.topicFilter(..).qos(...)零件的方法。我也找不到任何信息，这是否会使订阅对连接丢失更具弹性。

我将不胜感激有关该主题的任何信息。

谢谢。

4

1 回答 1

4

目前，如果代理在重新连接的 ConAck 中报告现有会话，HiveMQ MQTT 客户端将仅继续接收订阅消息。这需要两件事 - 1）您需要在初始连接时设置 cleanSession = false，以及 2）代理需要在连接之间没有丢失会话。

对于 1) 您可以尝试将其添加到您的连接中：

client.connectWith().cleanSession(false).send();

使用 2) 将取决于代理以及连接丢失的原因是什么。如果它“只是”网络中断并且代理在后台运行正常，那么它应该可以正常工作。如果代理崩溃并重新启动，那么它将要求代理配置持久性并且能够在重新启动后重新建立会话。

实际上，在 HiveMQ MQTT 客户端的 github 项目页面上有几个关于这个问题的讨论，以及是否应该添加功能以自动重新订阅，即使在没有找到预先存在的会话的情况下也是如此。并且在相关说明中，即使在重新连接后没有找到会话，在连接丢失时完成的任何发布是否都应该自动发布。如果这些是您需要的功能，也许可以跳到那里并参与讨论:)

最后，您还可以通过在构建客户端时添加 MqttClientConnectedListener 手动执行重新订阅，然后每次自动重新连接发生时重新创建订阅。

HTH
干杯，
C

于 2019-08-07T18:40:41.943 回答