RabbitMQ重试最大尝试次数结束后如何触发功能？（Spring 集成 - RabbitMQ 监听器）答案

【问题标题】：How to trigger a functionality after RabbitMQ retry max attempts are over? (Spring Integration - RabbitMQ Listener)RabbitMQ重试最大尝试次数结束后如何触发功能？（Spring 集成 - RabbitMQ 监听器）
【发布时间】：2018-06-27 18:21:21
【问题描述】：

我想在 RabbitMQ 侦听器重试结束后触发电子邮件，如果处理过程失败仍然如此。

重试逻辑正在使用以下代码。但是如何在最大重试次数结束后触发功能（电子邮件触发）。

@Bean
public SimpleMessageListenerContainer container() {
    SimpleMessageListenerContainer container =
            new SimpleMessageListenerContainer(connectionFactory());
    container.setQueues(myQueue());
    container.setDefaultRequeueRejected(false);
    Advice[] adviceArray = new Advice[]{interceptor()};
    container.setAdviceChain(adviceArray);
    return container;
}

@Bean
public IntegrationFlow inboundFlow() {
    return IntegrationFlows.from(
            Amqp.inboundAdapter(container()))
            .log()
            .handle(listenerBeanName, listenerMethodName)
            .get();
}

@Bean
RetryOperationsInterceptor interceptor() {
    return RetryInterceptorBuilder.stateless()
            .maxAttempts(retryMaxAttempts)
            .backOffOptions(initialInterval, multiplier, maxInterval)
             //.recoverer(new RejectAndDontRequeueRecoverer()) 
            .recoverer(new CustomRejectAndRecoverer())
            .build();
}

添加CustomeRecover的代码

@Service
public class CustomRejectAndRecoverer implements MessageRecoverer {
@Autowired
private EmailGateway emailgateway;

@Override
public void recover(Message message, Throwable cause) {
    // INSERT CODE HERE.... HOW TO CALL GATEWAY
    // emailgateway.sendMail(cause);
    throw new ListenerExecutionFailedException("Retry Policy Exhausted",
            new AmqpRejectAndDontRequeueException(cause), message);
} }

【问题讨论】：

你能把它清理一下以缩小到确切的问题吗？看起来您已经得到了答案，但我认为这个问题对其他读者来说尽可能有用。

标签： rabbitmq spring-integration spring-integration-dsl spring-integration-amqp

【解决方案1】：

这正是.recoverer() 中的RetryInterceptorBuilder 的用途。

您现在在那里使用RejectAndDontRequeueRecoverer，但没有任何人阻止您通过对RejectAndDontRequeueRecoverer 的委托实现自己的MessageRecoverer，并使用发送电子邮件逻辑向某些MessageChannel 发送消息。

【讨论】：

如果我使用下面的代码，我会在网关处收到 nullpointerexception。你能帮忙找出正确的方法吗@Artem：服务公共类 CustomRejectAndRecoverer 实现 MessageRecoverer { Autowired private EmailGateway emailgateway; Override public void recover(Message message, Throwable cause) { //INSERT CODE HERE.... HOW TO CALL GATEWAY //emailgateway.sendMail(message); throw new ListenerExecutionFailedException("重试策略用尽", new AmqpRejectAndDontRequeueException(cause), message); } }
请使用该代码编辑您的问题。但是，您似乎错过了必须将您的CustomRejectAndRecoverer 声明为@Bean 的事实，否则EmailGateway 将不会被注入。
我需要在recover方法中添加'@Bean'吗？
我不明白你的意思，但是因为我看到 RetryOperationsInterceptor 被声明为 @Bean，我认为你声明 @ 不会有问题987654332@ 与CustomRejectAndRecoverer 类似。更何况你没有显示你当前的代码状态......
添加了代码。你能检查一下吗？ @Artem。在添加的代码中，我将 emailgateway 设为 null