标签: spring-retry

是否可以根据错误状态代码在spring retry(https://github.com/spring-projects/spring-retry)中设置RetryPolicy ？例如我要重试上HttpServerErrorException与HttpStatus.INTERNAL_SERVER_ERROR状态码,其是503因此,应忽略所有其他错误代码- [500 - 502]和[504 - 511].

我必须使用一个项目,我不能使用Java-Config for Spring,但我必须使用XML-Config.现在我正在寻找与Java-Config中的@EnableRetry相当的XML-Config.

我希望这条线能够运作.

@Retryable(SQLException.class)
public void saveOrUpdate(Entity entity) 

我正在使用Spring-Retry进行一些数据库操作.在SQLRecoverableException我重试三次(这假设导致异常的任何内容是非瞬态的,如果它失败了三次),在SQLTransientException我无限期重试(程序无法做任何事情而无法访问数据库,所以它也可以继续重试,直到用户决定重新启动服务器),并且在任何其他异常时我都不会重试.我使用指数退避策略,基本重试为100毫秒,最大重试次数为30,000毫秒.

private static final int MAX_RECOVERABLE_RETRIES = 3;
private static final long INITIAL_INTERVAL = 100;
private static final long MAX_INTERVAL = 30 * 1000;
private static final double MULTIPLIER = 2.0;

public static RetryTemplate databaseTemplate() {
    RetryTemplate template = new RetryTemplate();
    ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();
    Map<Class<? extends Throwable>, RetryPolicy> policyMap = new HashMap<>();
    NeverRetryPolicy baseException = new NeverRetryPolicy();
    SimpleRetryPolicy recoverablePolicy = new SimpleRetryPolicy();
    recoverablePolicy.setMaxAttempts(MAX_RECOVERABLE_RETRIES);
    AlwaysRetryPolicy transientPolicy = new AlwaysRetryPolicy();
    policyMap.put(Exception.class, baseException);
    policyMap.put(SQLRecoverableException.class, recoverablePolicy); …

我正在尝试使用@Retryable调用 REST 模板的方法。如果由于通信错误而返回错误，我想重试，否则我只想在调用时抛出异常。

当 ApiException 发生时，而不是被 @Retryable 抛出和忽略，我收到了ExhaustedRetryException一个关于没有找到足够的“可恢复的”（即@Recover方法）的抱怨。

我想我会看看是否仅仅拥有可恢复的方法可能会让它快乐并且仍然按预期执行。没那么多。它没有抛出异常，而是调用了可恢复的方法。

@Retryable(exclude = ApiException include = ConnectionException, maxAttempts = 5, backoff = @Backoff(multiplier = 2.5d, maxDelay = 1000000L, delay = 150000L))
Object call(String domainUri, ParameterizedTypeReference type, Optional<?> domain = Optional.empty(), HttpMethod httpMethod = HttpMethod.POST) throws RestClientException {

    RequestEntity request = apiRequestFactory.createRequest(domainUri, domain, httpMethod)
    log.info "************************** Request Entity **************************"
    log.info "${request.toString()}"
    ResponseEntity response

    try {

        response = restTemplate.exchange(request, type)
        log.info "************************** Response Entity **************************"
        log.info "${response.toString()}"

    } …

我有一个宁静的服务，使用Spring Cloud Feign客户端调用外部服务

@FeignClient(name = "external-service", configuration = FeignClientConfig.class)
public interface ServiceClient {

    @RequestMapping(value = "/test/payments", method = RequestMethod.POST)
    public void addPayment(@Valid @RequestBody AddPaymentRequest addPaymentRequest);

    @RequestMapping(value = "/test/payments/{paymentId}", method = RequestMethod.PUT)
    public ChangePaymentStatusResponse updatePaymentStatus(@PathVariable("paymentId") String paymentId,
            @Valid @RequestBody PaymentStatusUpdateRequest paymentStatusUpdateRequest);

}

最近3个月，我在日志文件中发现3-4次以下失败：

json.ERROR_RESPONSE_BODY：连接拒绝执行POST http：// external-service / external / payments json.message：发送付款添加付款失败出于其他原因：{ERROR_RESPONSE_BODY =连接拒绝执行POST http：// external-service / external / payments，EVENT = ADD_PAYMENT_FAILURE，TRANSACTION_ID = XXXXXXX} {} json.EVENT：ADD_PAYMENT_FAILURE json.stack_trace：feign.RetryableException：连接拒绝执行POST http：// external-service / external / payments 在feign.FeignException.errorExecuting（FeignException.java:67）在feign.SynchronousMethodHandler.executeAndDecode（SynchronousMethodHandler.java:104）在feign.SynchronousMethodHandler.invoke（SynchronousMethodHandler.java:76）在feign.ReflectiveFeign $ FeignInvocationHandler.invoke（Reflective） Java：103）

是否可以在Feign客户端上添加Spring Retry。我想用什么来注释addPayment操作

@Retryable(value …

我正在尝试实现一个基于Spring Boot的Kafka消费者,它具有一些非常强大的消息传递保证,即使在出现错误的情况下也是如此.

• 必须按顺序处理来自分区的消息,
• 如果消息处理失败,应暂停特定分区的消耗,
• 应该通过退避重试处理,直到成功为止.

我们当前的实施满足以下要求:

  @Bean
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRetryTemplate(retryTemplate());

    final ContainerProperties containerProperties = factory.getContainerProperties();
    containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);
    containerProperties.setErrorHandler(errorHandler());

    return factory;
  }

  @Bean
  public RetryTemplate retryTemplate() {

    final ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
    backOffPolicy.setInitialInterval(1000);
    backOffPolicy.setMultiplier(1.5);

    final RetryTemplate template = new RetryTemplate();
    template.setRetryPolicy(new AlwaysRetryPolicy());    
    template.setBackOffPolicy(backOffPolicy);

    return template;
  }

  @Bean
  public ErrorHandler errorHandler() {
    return new SeekToCurrentErrorHandler();
  }

但是,在这里,记录永远被消费者锁定.在某些时候,处理时间将超过max.poll.interval.ms,服务器将分区重新分配给其他消费者,从而创建一个副本.

假设max.poll.interval.ms等于5分钟(默认)并且故障持续30分钟,这将导致消息被处理ca. 6次.

另一种可能性是在N次重试(例如3次尝试)之后通过使用将消息返回到队列SimpleRetryPolicy.然后,将重播该消息(感谢SeekToCurrentErrorHandler)并且处理将从头开始,最多再次尝试5次.这导致延迟形成一系列例如

10 secs …

我已经用KafkaHandler. 我的消费者应该消费事件，然后针对每个事件向其他服务发送 REST 请求。我只想在该 REST 服务关闭时重试。否则，我可以忽略失败的事件。

我的容器工厂配置如下：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, MyCustomEvent>
  kafkaListenerContainerFactory() {

  ConcurrentKafkaListenerContainerFactory<String, MyCustomEvent> factory =
    new ConcurrentKafkaListenerContainerFactory<>();

  factory.setConsumerFactory(consumerFactory());
  factory.setStatefulRetry(true);
  factory.setRetryTemplate(retryTemplate());
  factory.setConcurrency(3);

  ContainerProperties containerProperties = factory.getContainerProperties();
  containerProperties.setAckOnError(false);
  containerProperties.setAckMode(AckMode.RECORD);
  containerProperties.setErrorHandler(new SeekToCurrentErrorHandler());

  return factory;
}

我ExceptionClassifierRetryPolicy用于设置异常和相应的重试策略。

重试后一切看起来都很好。当我得到一个时它会重试，当我得到一个时ConnectException它会忽略IllegalArgumentException。

然而，在IllegalArgumentException场景中，SeekToCurrentErrorHandler返回到未处理的偏移量（因为它寻找未处理的消息，包括失败的消息），最终立即重试失败的消息。消费者不断地来回并重试百万次。

如果我有机会了解哪个记录失败了SeekToCurrentErrorHandler，那么我将创建一个自定义实现SeekToCurrentErrorHandler来检查失败的消息是否可重试（通过使用该thrownException字段）。如果它不可重试，那么我会将它从列表中删除records以寻找回来。

关于如何实现此功能的任何想法？

注：enable.auto.commit设为false，auto.offset.reset设为earliest。

谢谢！

我在类@Retryable中的方法上使用注释@Service

@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) …

我有这个测试:

@RunWith(MockitoJUnitRunner.class)
public class myServiceTest {

@InjectMocks
myService subject;

private myService spy;

@Before
public void before() {
    spy = spy(subject);
}

@Test
public void testing() {

    when(spy.print2()).thenThrow(new RuntimeException()).thenThrow(new RuntimeException()).thenReturn("completed");
    spy.print1();
    verify(spy, times(3)).print2();
}

然后我有:

@Service("myService")
public class myService extends myAbstractServiceClass {


public String print1() {
    String temp = "";
    temp = print2();
    return temp;
}

 @Retryable
 public String print2() {
     return "completed";
 }
}

然后我有这个接口(我的abstractService实现):

public interface myServiceInterface {

    @Retryable(maxAttempts = 3)
    String print1() throws RuntimeException;

    @Retryable(maxAttempts = 3)
    String …

我有以下方法，即@Retryable：

    @Retryable(value = HttpClientErrorException.class)
    public Integer callToExternalService(DateTime start, DateTime end) throws MyException

我的问题是，每次方法重试时是​​否可以修改输入参数，因为我需要使用不同的值进行休息调用。对于这种情况，是否有类似的选项@Recover？