RabbitMQ在使用者重新启动后停止向使用者传递MQTT消息

Question

我有一只兔子。OpenShift上的6.10实例在启用MQTT插件的情况下运行。我有多个Spring Boot应用程序，使用EclipsePAHOMQTT实现，使用RabbitMQ队列中的消息。所有使用者都使用MqttDefaultFilePersistence。持久性数据被写入具有100M配额的持久性卷上的目录/tmp/mqtt。这是我为连接RabbitMQ队列而编写的代码：

    MqttConnectOptions options = new MqttConnectOptions();
    options.setUserName(username);
    options.setPassword(password.toCharArray());
    options.setCleanSession(false);
    options.setAutomaticReconnect(reconnect);
    MqttClient mqttClient = new MqttClient(serverUri, "MyConsumerApp", new MqttDefaultFilePersistence("/tmp/mqtt"));
    mqttClient.connect(options);

现在我意识到，由于某些原因，如果重新启动消费应用程序，它将不再使用任何消息。正如我在RabbitMQ管理控制台中看到的那样，队列中的消息正在被填满。RabbitMQ日志中没有错误消息。在这种情况下唯一有效的解决方案是删除RabbitMQ管理控制台中的订阅队列，然后重新启动使用者应用程序。也许有人遇到过类似的问题？有人能告诉我这是怎么回事吗？也许RabbitMQ使用的MQTT插件和保存消息状态的PAHO实现存在问题。看起来RabbitMQ在消费者断开连接后不知道在哪里恢复，所以它只是停止传递消息。有点悲观的锁定…：-）

Answer 1

您必须使用持久队列。如果队列填满是问题，那么您可以使用自动删除队列。但这有一个缺点，即一次尝试传递消息，如果没有消费者，则队列将被删除，消息将丢失。

对于你的情况，你应该看看死信交换(https://www.rabbitmq.com/dlx.html)

如果消费者在一段时间后被切断并重新连接，未处理的消息可以从死信交换服务。这样消息的顺序也不会丢失。