Какие AmqpEvent или AmqpException обрабатывать при сбое монопольного потребителя

Question

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

Теперь я выяснил, что мне нужно обрабатывать ListenerContainerConsumerFailedEvent но я предполагаю, что реализация ApplicationListener для этого события не будет гарантировать, что я получаю это событие из-за исключительного исключения потребителя.Я мог бы хотеть проверить Throwable события, или событие дальнейшие проверки.

Какой подкласс AmqpException или какие дополнительные проверки я должен выполнить, чтобы гарантировать, что исключение получено из-за исключительного доступа потребителя?

Answer 1

Логика в реализациях контейнера слушателя выглядит следующим образом:

if (e.getCause() instanceof ShutdownSignalException
            && e.getCause().getMessage().contains("in exclusive use")) {
        getExclusiveConsumerExceptionLogger().log(logger,
                "Exclusive consumer failure", e.getCause());
        publishConsumerFailedEvent("Consumer raised exception, attempting restart", false, e);
    }

Итак, мы действительно генерируем событие ListenerContainerConsumerFailedEvent, и вы можете отследить сообщение о причине, как мы это делаем в рамках, но наС другой стороны, вы можете просто ввести свой собственный ConditionalExceptionLogger:

/**
 * Set a {@link ConditionalExceptionLogger} for logging exclusive consumer failures. The
 * default is to log such failures at WARN level.
 * @param exclusiveConsumerExceptionLogger the conditional exception logger.
 * @since 1.5
 */
public void setExclusiveConsumerExceptionLogger(ConditionalExceptionLogger exclusiveConsumerExceptionLogger) {

и поймать такую ​​исключительную ситуацию там.

Также вы можете использовать RabbitUtils.isExclusiveUseChannelClose(cause) в своем коде:

/**
 * Return true if the {@link ShutdownSignalException} reason is AMQP.Channel.Close
 * and the operation that failed was basicConsumer and the failure text contains
 * "exclusive".
 * @param sig the exception.
 * @return true if the declaration failed because of an exclusive queue.
 */
public static boolean isExclusiveUseChannelClose(ShutdownSignalException sig) {