java - SpringAMQP - 如何捕获 ListenerExecutionFailedException？

Question

我有 3 个队列。

队列主

队列重试(3次由代码控制)

尝试 3 次后排队保存错误消息。

它工作正常...但是如果我由于任何原因发送了错误的 json 消息，例如:

{
"name":"alan","  <<<< this ," is wrong for example
"age":29,
}

我的监听器不会尝试处理该消息(我想捕获此消息，因为如果我收到错误，我将发送到错误队列)

我收到 ListenerExecutionFailedException，然后丢失了此消息。

我尝试将错误处理程序添加到我的配置中:

@Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(RabbitProperties properties, ObjectProvider<ConnectionNameStrategy> connectionNameStrategy) throws Exception {
        PropertyMapper map = PropertyMapper.get();
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory(properties, connectionNameStrategy));
        factory.setMessageConverter(messageConverter());
        **factory.setErrorHandler(errorHandler());**
        RabbitProperties.Listener listener = properties.getListener();
        if (listener != null && listener.getSimple() != null) {
            map.from(listener.getSimple()::getConcurrency).whenNonNull().to(factory::setConcurrentConsumers);
            map.from(listener.getSimple()::getMaxConcurrency).whenNonNull().to(factory::setMaxConcurrentConsumers);
            map.from(listener.getSimple()::getDefaultRequeueRejected).whenNonNull().to(factory::setDefaultRequeueRejected);

        }
        return factory;
    }

public ErrorHandler errorHandler() {
        return new ConditionalRejectingErrorHandler(new ListenerExceptionHandler());
    }

@Log4j2
public class ListenerExceptionHandler extends ConditionalRejectingErrorHandler.DefaultExceptionStrategy {

    @Override
    public boolean isFatal(Throwable t) {
        if (t instanceof ListenerExecutionFailedException) {
            ListenerExecutionFailedException lefe = (ListenerExecutionFailedException) t;
            log.error("Failed to process inbound message from queue {}, failed message={}",
                    lefe.getFailedMessage().getMessageProperties().getConsumerQueue(),
                    lefe.getFailedMessage());
        }
        return super.isFatal(t);
    }

}

我只是想是否可以在我的监听器上捕获此 ListenerExecutionFailedException ..可能吗？

因为在我的处理程序类中我不能做任何事情..只需设置日志..我无法将此消息发送到错误队列。

监听者:

@RabbitListener(queues = Queues.MAIN, concurrency = "2")
public void listenerMessage(Message message,@Header(name = "x-death", required = false) List<Map<String, ?>> xDeath) {
    log.info("ProcessMessage Received:: {}", message.getPayload());
    validateReceivedMessage(xDeath, message);
}

我尝试添加 try/catch ，例如:

try {

}catch (ListenerExecutionFailedException e){

}

没有成功。

有什么建议吗？谢谢

Answer 1

没有；消息转换异常发生在堆栈中太靠下的位置；无法使用转换失败的消息调用监听器。

此类异常被认为是致命的，以避免无限循环；您需要在原始队列上配置死信交换和路由 key ，以便将失败的消息路由到那里。