我在@Service类中的方法上使用了@Retryable批注
@Service
@EnableRetry
public class PushService {
@Retryable(maxAttempts=5)
public Result pushIt(myMessage messageIn) {
...
}
}它的工作方式很简单:我直接从RabbitMQ收到一条消息,直到没有错误或尝试次数达到5次时才会得到确认,此时消息会直接到达DLQ,就像我想要的那样。
我唯一的问题是,我需要从属性文件中动态设置maxAttempts。解决方案应该是设置一个拦截器,但拥有一个拦截器的唯一事实会导致错误,例如,当我有:
@Service
@EnableRetry
public class PushService {
@Retryable(interceptor="myInterceptor")
public Result pushIt(myMessage messageIn) {
...
}
}其中,myInterceptor定义为:
@Bean
public StatefulRetryOperationsInterceptor myInterceptor() {
return RetryInterceptorBuilder.stateful().maxAttempts(5).build();
}我得到了一个无限循环,但有以下异常:
2015-04-08 07:12:10,970 GMT [SimpleAsyncTaskExecutor-1] (ConditionalRejectingErrorHandler.java:67) WARN listener.ConditionalRejectingErrorHandler: Execution of Rabbit message listener failed.
org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.wrapToListenerExecutionFailedExceptionIfNeeded(AbstractMessageListenerContainer.java:864)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:802)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:690)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$001(SimpleMessageListenerContainer.java:82)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$1.invokeListener(SimpleMessageListenerContainer.java:167)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.invokeListener(SimpleMessageListenerContainer.java:1241)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.executeListener(AbstractMessageListenerContainer.java:660)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.doReceiveAndExecute(SimpleMessageListenerContainer.java:1005)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.receiveAndExecute(SimpleMessageListenerContainer.java:989)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer.access$700(SimpleMessageListenerContainer.java:82)
at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1103)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 1
at org.springframework.amqp.rabbit.config.StatefulRetryOperationsInterceptorFactoryBean$3.getKey(StatefulRetryOperationsInterceptorFactoryBean.java:103)
at org.springframework.retry.interceptor.StatefulRetryOperationsInterceptor.invoke(StatefulRetryOperationsInterceptor.java:132)
at org.springframework.retry.annotation.AnnotationAwareRetryOperationsInterceptor.invoke(AnnotationAwareRetryOperationsInterceptor.java:118)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:653)
at com.acme.push.service.PushService$$EnhancerBySpringCGLIB$$9d503bc1.pushMessage(<generated>)
at com.acme.push.receiver.PushListener.onMessage(PushListener.java:42)
at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:799)
... 10 more我很确定我把它保持得太简单了,但我就是不知道是什么导致了这个错误,以及如何解决它,有人知道发生了什么吗?
发布于 2015-04-09 18:16:46
我最终设法在不使用@Retrayable注释的情况下获得了所需的灵活性。
我用我的延迟和最大尝试次数参数创建了一个RetryAdvice:
@Bean
public MethodInterceptor retryAdvice() {
FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(delay);
return RetryInterceptorBuilder.stateless().backOffPolicy(backOffPolicy)
.maxAttempts(maxAttempts).build();
}我将建议插入到ListenerContainer的adviceChain中
@Bean
public SimpleMessageListenerContainer replyListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(pushConnectionFactory());
container.setQueues(pushQueue());
container.setMessageListener(pushListener());
Advice[] adviceChain = new Advice[] { retryAdvice() };
container.setAdviceChain(adviceChain);
return container;
}这样,每当我的听众抛出
throw new AmqpRejectAndDontRequeueException(cause);这将导致容器以所需的延迟重试所指示的次数,之后将传播异常并在DLQ中传递消息
发布于 2015-04-08 16:48:31
不能通过注释@Retryable来使用org.springframework.amqp.rabbit.config.RetryInterceptorBuilder。
它的用途是与SimpleRabbitListenerContainerFactory类的通知链一起使用。请参阅参考文档receiving-messages和SimpleMessageListenerContainer#invokeListener签名:
@Override
protected void invokeListener(Channel channel, Message message) throws Exception {
proxy.invokeListener(channel, message);
}一旦您为通知链配置了适当的RetryInterceptor,您的注释就会变得毫无用处。
https://stackoverflow.com/questions/29508311
复制相似问题