Apache Camel MQTT - не возникает исключение, когда брокер сообщений не работает

Question

В настоящее время я работаю с Apache Camel и его компонентом MQTT. У меня есть маршрут, использующий сообщения от посредника (Apache ActiveMQ artemis) и еще один отправляющий ему сообщения. Проблема заключается в том, что не возникает исключение, когда посредник сообщений недоступен. Более того, все сообщения, которые не отправляются, хранятся в памяти в ожидании возможного перезапуска сервера, что вызывает переполнение памяти. Я не знаю, связано ли это с самим протоколом MQTT или с конфигурацией конечной точки.

Вот моя конфигурация:

    MQTTEndpoint mqttEndpoint = null;

    mqttEndpoint = (MQTTEndpoint) mqttComponent.createEndpoint(MQTT_BROKER);

    mqttEndpoint.getConfiguration().setHost(properties.getBrokerAddress());     
    mqttEndpoint.getConfiguration().setPublishTopicName(publishTopicName);
    //mqttEndpoint.getConfiguration().setSubscribeTopicNames(subscribreTopicNames);
    mqttEndpoint.getConfiguration().setUserName(properties.getBrokerUsername());
    mqttEndpoint.getConfiguration().setPassword(properties.getBrokerPassword());
    mqttEndpoint.getConfiguration().setSslContext(createSSLContext());
    mqttEndpoint.getConfiguration().setByDefaultRetain(false);
    mqttEndpoint.getConfiguration().setQualityOfService(QoS.AT_MOST_ONCE.toString());
    mqttEndpoint.getConfiguration().setConnectAttemptsMax(1);
    mqttEndpoint.getConfiguration().setConnectWaitInSeconds(5);
    mqttEndpoint.getConfiguration().setReconnectBackOffMultiplier(1);
    mqttEndpoint.getConfiguration().setDisconnectWaitInSeconds(3);      
    mqttEndpoint.setCamelContext(camelCtx);

Answer 1

Так что это правильное поведение для уровня QOS, который вы установили. Вы устанавливаете флаг QOS на QoS.AT_MOST_ONCE.toString(), Это известно как QOS уровня 2.

Небольшое резюме QOS 2 - только один раз

Этот уровень гарантирует, что сообщение будет доставлено только один раз. Если есть проблемы с сетью, и она не может доставить ее, сообщение останется в очереди клиента, пока доставка не будет возможна. Это самый медленный уровень QOS, поскольку он требует 4 сообщений.

1. Отправитель отправляет сообщение и ожидает подтверждения (PUBREC)

2. Получатель отправляет сообщение PUBREC

3. Если отправитель не получает подтверждение (PUBREC), он повторно отправит сообщение с установленным флагом DUP.

4. Когда отправитель получает подтверждающее сообщение PUBREC, он затем отправляет сообщение об освобождении сообщения (PUBREL).
5. Если отправитель не получает PUBREL, он отправит сообщение PUBREC
6. Когда получатель получает сообщение PUBREL, он может теперь переслать сообщение любым подписчикам.
7. Затем получатель отправляет завершенную публикацию (PUBCOMP) .
8. Если отправитель не получит сообщение PUBCOMP, он отправит сообщение PUBREL.
9. Когда отправитель получает PUBCOMP, процесс завершается, и он может удалить сообщение из исходящей очереди.

Смотрите эту запись в блоге для получения дополнительной информации.

Наиболее важной частью является то, что в вашем случае приемник недоступен, поэтому протокол MQTT для QOS 2 не может быть завершен.