将清除会话标志设置为 FALSE 时，我丢失了已发布的值

Question

有人有一个合乎逻辑的解释，为什么尽管我有

clear session flage = false

，但我在未连接到经纪人时却没有收到我订阅的更新的已发布消息？

将 aore 上述标志设置为 false 后，我运行了我的应用程序，并且不断向主题发布一些值（从 PC 到代理，而不是从 Android 设备到代理）。然后我从电脑上拔下了 Android 设备的 USB 插头，并再次继续发布一些值。当我再次将 USB 插入电脑时，我希望看到我最近发布到该主题的值，但我没有收到任何值。

我发布的主题正是我订阅的主题，带有

QoS = 1

。

Answer 1

为了使消息能够为不再连接到代理的客户端排队，必须满足以下条件：

• 客户端最初必须已连接，并将

  clean session

  设置为 false
• 客户端必须以 QoS>0 进行订阅
• 发送到该主题的消息必须以 QoS>0 的方式发布
• 当客户端重新连接时，它必须具有与断开连接时相同的

  client id

• 当客户端重新连接时，必须将

  clean session

  设置为 false

我敢打赌，你在后两点中的任何一个都失败了。

Answer 2

您需要发送带有保留标志的消息，以便在重新连接时能够收到消息。 QoS 定义了代理/客户端将如何努力确保接收到消息。

请阅读此处 http://mosquitto.org/man/mqtt-7.html 有关 QOS、清理会话和保留消息的内容。

Answer 3

@ralight 的答案是正确的，并且从概念上解决了大多数容易发生错误的问题。

但是，我想在他的答案中添加 Java MQTT 客户端 Paho org.eclipse.paho.client.mqttv3 的具体细微差别。我已按照 @ralight 提到的整个步骤进行操作，但在客户端重新连接到代理时仍然没有收到 QOS >=1 消息。这里的主要问题是 Paho 客户端的初始化方式。作为参考，我将采用开源 kafka connect mqtt 连接器实现 https://github.com/johanvandevenne/kafka-connect-mqtt/blob/master/src/main/java/be/jovacon/kafka/connect/MQTTSourceTask。 java#L33

此实现存在问题，即它不会收到客户端断开连接时发送的消息，并使用 cleanSession false 重新连接并订阅 qos >= 1 （在我的测试中，它确实偶尔收到了一些消息）。

解决这个问题的方法是

• 旧代码
  • 初始化客户端
  • 连接客户端
  • 订阅
  • 代码https://github.com/johanvandevenne/kafka-connect-mqtt/blob/master/src/main/java/be/jovacon/kafka/connect/MQTTSourceTask.java#L33
• 新代码
  • 初始化客户端
  • 在客户端上设置回调（在调用之前调用 mqttClient.setCallback 以连接 messageArrived 回调，该回调将接收旧消息）
  • 连接客户端
  • 订阅
  • 代码更改https://github.com/johanvandevenne/kafka-connect-mqtt/compare/master...thehellmaker:kafka-connect-mqtt:master