Sha*_*gab 3 rabbitmq spring-rabbit spring-amqp spring-boot
我有以下配置
spring.rabbitmq.listener.prefetch=1
spring.rabbitmq.listener.concurrency=1
spring.rabbitmq.listener.retry.enabled=true
spring.rabbitmq.listener.retry.max-attempts=3
spring.rabbitmq.listener.retry.max-interval=1000
spring.rabbitmq.listener.default-requeue-rejected=false //I have also changed it to true but the same behavior still happens
Run Code Online (Sandbox Code Playgroud)
在我的监听器中,我抛出异常AmqpRejectAndDontRequeueException 来拒绝消息并强制兔子不要尝试重新传递它......但是兔子重新传递它3 次,然后最终将它路由到死信队列。
这是根据我提供的配置的标准行为还是我错过了什么?
您必须将重试策略配置为不为该异常重试。
你不能用属性来做到这一点,你必须自己配置重试建议。
如果您需要帮助,我稍后会发布一个示例。
requeue-rejected
位于容器级别(在堆栈上重试下方)。
编辑
@SpringBootApplication
public class So39853762Application {
public static void main(String[] args) throws Exception {
ConfigurableApplicationContext context = SpringApplication.run(So39853762Application.class, args);
Thread.sleep(60000);
context.close();
}
@RabbitListener(queues = "foo")
public void foo(String foo) {
System.out.println(foo);
if ("foo".equals(foo)) {
throw new AmqpRejectAndDontRequeueException("foo"); // won't be retried.
}
else {
throw new IllegalStateException("bar"); // will be retried
}
}
@Bean
public ListenerRetryAdviceCustomizer retryCustomizer(SimpleRabbitListenerContainerFactory containerFactory,
RabbitProperties rabbitPropeties) {
return new ListenerRetryAdviceCustomizer(containerFactory, rabbitPropeties);
}
public static class ListenerRetryAdviceCustomizer implements InitializingBean {
private final SimpleRabbitListenerContainerFactory containerFactory;
private final RabbitProperties rabbitPropeties;
public ListenerRetryAdviceCustomizer(SimpleRabbitListenerContainerFactory containerFactory,
RabbitProperties rabbitPropeties) {
this.containerFactory = containerFactory;
this.rabbitPropeties = rabbitPropeties;
}
@Override
public void afterPropertiesSet() throws Exception {
ListenerRetry retryConfig = this.rabbitPropeties.getListener().getRetry();
if (retryConfig.isEnabled()) {
RetryInterceptorBuilder<?> builder = (retryConfig.isStateless()
? RetryInterceptorBuilder.stateless()
: RetryInterceptorBuilder.stateful());
Map<Class<? extends Throwable>, Boolean> retryableExceptions = new HashMap<>();
retryableExceptions.put(AmqpRejectAndDontRequeueException.class, false);
retryableExceptions.put(IllegalStateException.class, true);
SimpleRetryPolicy policy =
new SimpleRetryPolicy(retryConfig.getMaxAttempts(), retryableExceptions, true);
ExponentialBackOffPolicy backOff = new ExponentialBackOffPolicy();
backOff.setInitialInterval(retryConfig.getInitialInterval());
backOff.setMultiplier(retryConfig.getMultiplier());
backOff.setMaxInterval(retryConfig.getMaxInterval());
builder.retryPolicy(policy)
.backOffPolicy(backOff)
.recoverer(new RejectAndDontRequeueRecoverer());
this.containerFactory.setAdviceChain(builder.build());
}
}
}
}
Run Code Online (Sandbox Code Playgroud)
注意:您当前无法配置策略以重试所有异常,“除了”此异常 - 您必须对要重试的所有异常进行分类(并且它们不能是 的超类AmqpRejectAndDontRequeueException
)。我已经打开了一个问题来支持这一点。
归档时间: |
|
查看次数: |
6879 次 |
最近记录: |