guava 重试工具库 -- Retryer

2018-03-01 22:00:41   最后更新: 2018-03-02 06:08:52   访问数量:597




在我们的开发中,api 接口调用异常是经常会遇到的,任何接口都会有不同概率的异常情况,对于可以重入的接口,为了避免偶发性异常造成的服务的不可用,重试机制就非常有必要了

通常我们的接口超时都是符合正态分布的,即 99.9% 的请求超时时间都不会大于某个值,将这个值设置为请求的超时时间是比较合理的,因为即使将超时时间设置为该值的两倍或更高,按照正态分布的概率来看,请求成功率并不会有明显提高,而使用该值作为超时时间可以保证在 99.9% 的情况下请求正常返回,对于异常情况我们添加重试机制就可以大概率减少失败

但是需要注意的是,我们的调用链通常是非常长的,如果在多个层级上都添加重试,那么最终将会造成雪崩效应,最底层将受到指数扩大的请求次数,这对于底层服务的压力是非常大的,因此重试的设置需要非常谨慎,这在底层服务出现问题时尤为明显,过度的重试将会成倍加剧问题的严重性,因此,在重试机制的同时增加递增的重试间隔时间或者熔断机制也是非常有必要的,有时间的话博主将会在博客中介绍用于业务熔断组件 Hystrix,敬请期待

 

guava 提供了非常优雅、方便的重试工具类 -- retryer

 

<dependency> <groupId>com.github.rholder</groupId> <artifactId>guava-retrying</artifactId> <version>2.0.0</version> </dependency>

 

 

在这个包中,最重要的类就是 Retryer,只要我们通过一定的策略创建出 Retryer 对象,通过调用 Retryer 对象的 call 方法,即可按照既定策略执行参数中传入的 Callable 接口的 call 方法

 

Retryer 提供了构造方法,用来创建一个指定规则的 Retryer:

Retryer(@Nonnull AttemptTimeLimiter<V> attemptTimeLimiter, @Nonnull StopStrategy stopStrategy, @Nonnull WaitStrategy waitStrategy, @Nonnull BlockStrategy blockStrategy, @Nonnull Predicate<Attempt<V>> rejectionPredicate, @Nonnull Collection<RetryListener> listeners)

 

 

这个方法可以通过传入尝试时间策略、停止重试策略、重试间隔等待策略、重试阻塞策略、拒绝策略等策略来指定一个请求的重试如何进行

 

  1. attemptTimeLimiter -- 每次重试的最大超时,超过该超时则抛出 TimeoutException
  2. stopStrategy -- 停止重试的策略
  3. waitStrategy -- 重试间间隔等待策略
  4. blockStrategy -- 重试间间隔线程阻塞策略
  5. rejectionPredicate -- 拒绝尝试断言
  6. listeners -- 重试的监听者

 

public interface AttemptTimeLimiter<V> { V call(Callable<V> var1) throws Exception; }

 

这个接口主要是用来控制每次重试的超时时间

通常来说,我们会通过 AttemptTimeLimiters 类来创建这个接口的实现

 

AttemptTimeLimiters 就是一个生产 AttemptTimeLimiter 实现的工厂类,主要的方法有三个:

 

不限制时间 -- noTimeLimit

public static <V> AttemptTimeLimiter<V> noTimeLimit()

 

 

限制超时 -- AttemptTimeLimiter

public static <V> AttemptTimeLimiter<V> fixedTimeLimit(long duration, @Nonnull TimeUnit timeUnit) public static <V> AttemptTimeLimiter<V> fixedTimeLimit(long duration, @Nonnull TimeUnit timeUnit, @Nonnull ExecutorService executorService)

 

两个方法通过 duration 和 timeUnit 组合实现了对调用的接口实现总的超时控制

 

public interface StopStrategy { boolean shouldStop(Attempt var1); }

 

这个接口中只有一个方法,顾名思义,他就是用来判断是否应该停止重试

他传入了 Attempt 为参数,通过这个参数,我们可以获取到方法的返回值、抛出的异常、重试的次数等等

 

public interface Attempt<V> { V get() throws ExecutionException; boolean hasResult(); boolean hasException(); V getResult() throws IllegalStateException; Throwable getExceptionCause() throws IllegalStateException; long getAttemptNumber(); long getDelaySinceFirstAttempt(); }

 

 

我们也可以通过 StopStrategys 来生产 StopStrategy 对象

它提供了三个 static 方法:

不停止 -- neverStop

public static StopStrategy neverStop()

 

 

在一定次数后停止 -- stopAfterAttempt

public static StopStrategy stopAfterAttempt(int attemptNumber)

 

 

在一定超时后停止 -- stopAfterDelay

public static StopStrategy stopAfterDelay(long duration, @Nonnull TimeUnit timeUnit)

 

 

public interface WaitStrategy { long computeSleepTime(Attempt var1); }

 

WaitStrategy 接口只包含一个方法,顾名思义,就是间隔的时长

 

同样的,我们也可以使用 WaitStrategies 类来生产 WaitStrategy

WaitStrategies 提供了非常强大而丰富的 static 方法

 

不间隔 -- noWait

public static WaitStrategy noWait()

 

 

指定时间间隔 -- fixedWait

public static WaitStrategy fixedWait(long sleepTime, @Nonnull TimeUnit timeUnit)

 

 

随机间隔 -- randomWait

public static WaitStrategy randomWait(long maximumTime, @Nonnull TimeUnit timeUnit); public static WaitStrategy randomWait(long minimumTime, @Nonnull TimeUnit minimumTimeUnit, long maximumTime, @Nonnull TimeUnit maximumTimeUnit);

 

 

线性递增间隔 -- incrementingWait

public static WaitStrategy incrementingWait(long initialSleepTime, @Nonnull TimeUnit initialSleepTimeUnit, long increment, @Nonnull TimeUnit incrementTimeUnit)

 

这个方法设置了初始间隔和递增步长

 

指数递增间隔

public static WaitStrategy exponentialWait(); public static WaitStrategy exponentialWait(long maximumTime, @Nonnull TimeUnit maximumTimeUnit); public static WaitStrategy exponentialWait(long multiplier, long maximumTime, @Nonnull TimeUnit maximumTimeUnit);

 

 

斐波那切数列递增间隔 -- fibonacciWait

public static WaitStrategy fibonacciWait(); public static WaitStrategy fibonacciWait(long maximumTime, @Nonnull TimeUnit maximumTimeUnit); public static WaitStrategy fibonacciWait(long multiplier, long maximumTime, @Nonnull TimeUnit maximumTimeUnit);

 

 

一旦抛出异常则间隔 -- exceptionWait

public static <T extends Throwable> WaitStrategy exceptionWait(@Nonnull Class<T> exceptionClass, @Nonnull Function<T, Long> function)

 

他是 WaitStrategies 中最复杂的一个策略了

一旦上一次尝试抛出了 exceptionClass 及其子类的异常,则调用回调方法 function,以其返回值作为下一次尝试前的时间间隔

 

合并多个策略 -- join

public static WaitStrategy join(WaitStrategy... waitStrategies)

 

 

public interface BlockStrategy { void block(long var1) throws InterruptedException; }

 

这个策略指定了线程在本次尝试后 sleep 多少毫秒

 

BlockStrategies 可以方便的创建 BlockStrategy

他只提供了一个 static 方法,一旦指定,则线程会在间隔时间内 sleep,否则不会

public static BlockStrategy threadSleepStrategy()

 

 

@FunctionalInterface @GwtCompatible public interface Predicate<T> extends java.util.function.Predicate<T> { @CanIgnoreReturnValue boolean apply(@Nullable T var1); boolean equals(@Nullable Object var1); default boolean test(@Nullable T input) { return this.apply(input); } }

 

Predicate 接口最重要的方法是 apply 方法,返回是否需要拒绝尝试

与 AttemptTimeLimiters 类似,Predicate 通常用 Predicates 来创建

 

Predicates 提供了非常丰富的 static 方法集合,可以实现各种各样的断言,甚至是断言的与或非组合

public static <T> Predicate<T> alwaysTrue(); public static <T> Predicate<T> alwaysFalse(); public static <T> Predicate<T> isNull(); public static <T> Predicate<T> notNull(); public static <T> Predicate<T> not(Predicate<T> predicate); public static <T> Predicate<T> and(Iterable<? extends Predicate<? super T>> components); public static <T> Predicate<T> and(Predicate... components); public static <T> Predicate<T> and(Predicate<? super T> first, Predicate<? super T> second); public static <T> Predicate<T> or(Iterable<? extends Predicate<? super T>> components); public static <T> Predicate<T> or(Predicate... components); public static <T> Predicate<T> or(Predicate<? super T> first, Predicate<? super T> second); public static <T> Predicate<T> equalTo(@NullableDecl T target); public static Predicate<Object> instanceOf(Class<?> clazz); public static Predicate<Class<?>> subtypeOf(Class<?> clazz); public static <T> Predicate<T> in(Collection<? extends T> target); public static <A, B> Predicate<A> compose(Predicate<B> predicate, Function<A, ? extends B> function); public static Predicate<CharSequence> containsPattern(String pattern); public static Predicate<CharSequence> contains(Pattern pattern);

 

 

上面提到,Retryer 具有 call 方法,只要设置好重试策略,将相应方法封装为 Callable 接口的实现,传入 Retryer 的 call 方法,即可按照预定的重试策略调用对应的方法

但是,如果我们的方法不是直接执行,而是需要放入线程池中呢?Retryer 提供了 wrap 接口实现将方法的重试策略封装到一个 Callable 实现中,从而让我们可以直接通过线程池调用:

public Retryer.RetryerCallable<V> wrap(Callable<V> callable)

 

 






读书笔记      技术帖      龙潭书斋      技术分享      java      guava      retryer      重试     


1#duchao: (回复)2018-04-23 11:04:05

要是添加一些 demo 会更生动一些

2#stenenjohn: (回复)2018-09-17 22:42:30

要是能够通俗的说下那个调用具体是怎么执行的就好了,比如他的重试,是在for循环里面,然后源码具体每一步干了什么之类的,多谢

京ICP备15018585号