Обнаружение и восстановление после удаления учетных данных в Spring AMQP

Для такого сценария контейнер прослушивателя выдает ListenerContainerConsumerFailedEvent. Вы можете прослушать это, проверить его reason и исключение и решить stop() контейнер и сделать что-то еще, что вам нужно. Затем start() снова использовать брокера с новыми учетными данными.

На стороне RabbitTemplate нужно просто try...catch вызвать и проанализировать исключение по той же причине.

Это не то, что я пробовал до сих пор, но это мое лучшее ощущение, как справиться с состоянием ACCESS_REFUSED. Вы действительно ничего не можете сделать по этому вопросу с точки зрения CachingConnectionFactory.

ОБНОВЛЕНИЕ

Мое приложение выглядит так:

spring.rabbitmq.username=test
spring.rabbitmq.password=test
spring.rabbitmq.template.retry.enabled=true
spring.rabbitmq.template.retry.initial-interval=1ms
logging.level.org.springframework.retry=DEBUG

@SpringBootApplication
public class So49155945Application {

    public static void main(String[] args) {
        ConfigurableApplicationContext applicationContext = SpringApplication.run(So49155945Application.class, args);
        RabbitTemplate rabbitTemplate = applicationContext.getBean(RabbitTemplate.class);

        try {
            rabbitTemplate.convertAndSend("foo", "foo");
        }
        catch (AmqpException e) {
            System.err.println("Error during sending: " + e.getCause().getCause().getMessage());
        }
    }

}

И это то, что у меня есть в консоли, когда я запускаю это приложение для этого несуществующего пользователя:

Error during sending: ACCESS_REFUSED - Login was refused using authentication mechanism PLAIN. For details see the broker logfile.

ОБНОВЛЕНИЕ 2

Что я нашел также, как мы можем сделать этот реквизит:

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.template.mandatory=true

Затем добавьте rabbitTemplate.setConfirmCallback(), и наше отклоненное сообщение для асинхронной отправки будет отклонено. Однако это по-прежнему асинхронный обратный вызов, аналогичный упомянутому ChannelListener. С точки зрения Spring AMQP действительно нечего делать. Все является асинхронной природой протокола AMQP и может действительно нуждаться в каком-то «быстром отказе» из клиентской библиотеки Rabbit.

Пожалуйста, задайте такой вопрос в группе rabbitmq-users Google. Это место, где тусуются инженеры RabbitMQ.

ОБНОВЛЕНИЕ 3

В качестве решения для таких событий на Брокере может использоваться Плагин обмена событиями. Конкретные события user.deleted или user.password.changed генерируются Брокером.

После долгих экспериментов и отладки я принял предложение Артема Билана и принял Плагин обмена событиями RabbitMQ.

Так что теперь, вместо того, чтобы пытаться отслеживать события ShutdownSignalException и ListenerContainerConsumerFailedEvent в коде Spring и Rabbit, между SimpleMessageListenerContainer, с одной стороны, и RabbitTemplate, с другой, я просто подписываюсь на обмен и позволяю своему новому @RabbitListener уведомлять меня о проблемах с учетными данными. Это без каких-либо других движущихся частей или объявлений bean-компонентов, без каких-либо проблем с синхронизацией или заблокированных потоков, и, как правило, следует за потоком автоконфигурации, а не борется с ним.

Все, что мне сейчас нужно, это:

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.endpoint.RefreshEndpoint;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Component;

import static org.springframework.amqp.core.ExchangeTypes.TOPIC;

@Component
public class ReuathenticationListener {

    private static Logger log = LoggerFactory.getLogger(ReuathenticationListener.class);

    @Autowired private RabbitProperties rabbitProperties;
    @Autowired private RefreshEndpoint refreshEndpoint;
    @Autowired private CachingConnectionFactory connectionFactory;

    @RabbitListener(
        id = "credential_expiry_listener",
        bindings = @QueueBinding(value = @Queue(value="credentials.expiry", autoDelete="true", durable="false"),
            exchange = @Exchange(value="amq.rabbitmq.event", type=TOPIC, internal="true", durable="true"),
            key = "user.#")
    )
    public void expiryHandler(final MessageHeaders headers) {
        final String key = (String) headers.get("amqp_receivedRoutingKey");
        // See: https://www.rabbitmq.com/event-exchange.html
        if (!key.equals("user.deleted") &&
            !key.equals("user.authentication.failure")) {
            return;
        }

        final String failedName = (String) headers.get("name");
        final String prevUsername = rabbitProperties.getUsername();

        if (!failedName.equals(prevUsername)) {
            log.debug("Ignore expiry of unrelated user: " + failedName);
            return;
        }

        log.info("Refreshing Rabbit credentials...");
        refreshEndpoint.refresh();
        log.info("Refreshed username: '" + prevUsername + "' => '" + rabbitProperties.getUsername() + "'");

        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.resetConnection();

        log.info("CachingConnectionFactory reset, reconnection should now begin.");
    }
}