序
本文主要研究一下resilience4j的Retry
Retry
resilience4j-retry-0.13.0-sources.jar!/io/github/resilience4j/retry/Retry.java
/**
* A Retry instance is thread-safe can be used to decorate multiple requests.
* A Retry.
*/
public interface Retry {
/**
* Returns the ID of this Retry.
*
* @return the ID of this Retry
*/
String getName();
/**
* Creates a retry Context.
*
* @return the retry Context
*/
Retry.Context context();
/**
* Returns the RetryConfig of this Retry.
*
* @return the RetryConfig of this Retry
*/
RetryConfig getRetryConfig();
/**
* Returns an EventPublisher can be used to register event consumers.
*
* @return an EventPublisher
*/
EventPublisher getEventPublisher();
/**
* Creates a Retry with a custom Retry configuration.
*
* @param name the ID of the Retry
* @param retryConfig a custom Retry configuration
*
* @return a Retry with a custom Retry configuration.
*/
static Retry of(String name, RetryConfig retryConfig){
return new RetryImpl(name, retryConfig);
}
/**
* Creates a Retry with a custom Retry configuration.
*
* @param name the ID of the Retry
* @param retryConfigSupplier a supplier of a custom Retry configuration
*
* @return a Retry with a custom Retry configuration.
*/
static Retry of(String name, Supplier<RetryConfig> retryConfigSupplier){
return new RetryImpl(name, retryConfigSupplier.get());
}
/**
* Creates a retryable supplier.
*
* @param retry the retry context
* @param supplier the original function
* @param <T> the type of results supplied by this supplier
*
* @return a retryable function
*/
static <T> Supplier<T> decorateSupplier(Retry retry, Supplier<T> supplier){
return () -> {
Retry.Context context = retry.context();
do try {
T result = supplier.get();
context.onSuccess();
return result;
} catch (RuntimeException runtimeException) {
context.onRuntimeError(runtimeException);
} while (true);
};
}
/**
* Creates a retryable callable.
*
* @param retry the retry context
* @param supplier the original function
* @param <T> the type of results supplied by this supplier
*
* @return a retryable function
*/
static <T> Callable<T> decorateCallable(Retry retry, Callable<T> supplier){
return () -> {
Retry.Context context = retry.context();
do try {
T result = supplier.call();
context.onSuccess();
return result;
} catch (RuntimeException runtimeException) {
context.onRuntimeError(runtimeException);
} while (true);
};
}
/**
* Creates a retryable runnable.
*
* @param retry the retry context
* @param runnable the original runnable
*
* @return a retryable runnable
*/
static Runnable decorateRunnable(Retry retry, Runnable runnable){
return () -> {
Retry.Context context = retry.context();
do try {
runnable.run();
context.onSuccess();
break;
} catch (RuntimeException runtimeException) {
context.onRuntimeError(runtimeException);
} while (true);
};
}
//......
}
- 这个类定义了一些工厂方法，最后new的是RetryImpl
- 还定义了decorate开头的方法，包装retry的逻辑
- retry逻辑是包装在一个循环里头，先执行业务代码，如果成功调用context.onSuccess()，跳出循环，如果失败捕获RuntimeException，然后调用context.onRuntimeError
RetryConfig
resilience4j-retry-0.13.0-sources.jar!/io/github/resilience4j/retry/RetryConfig.java
public class RetryConfig {
public static final int DEFAULT_MAX_ATTEMPTS = 3;
public static final long DEFAULT_WAIT_DURATION = 500;
public static final IntervalFunction DEFAULT_INTERVAL_FUNCTION = (numOfAttempts) -> DEFAULT_WAIT_DURATION;
public static final Predicate<Throwable> DEFAULT_RECORD_FAILURE_PREDICATE = (throwable) -> true;
private int maxAttempts = DEFAULT_MAX_ATTEMPTS;
private IntervalFunction intervalFunction = DEFAULT_INTERVAL_FUNCTION;
// The default exception predicate retries all exceptions.
private Predicate<Throwable> exceptionPredicate = DEFAULT_RECORD_FAILURE_PREDICATE;
//......
}
- 该配置主要有3个参数，一个是maxAttempts，一个是exceptionPredicate，一个是intervalFunction
Retry.Context
resilience4j-retry-0.13.0-sources.jar!/io/github/resilience4j/retry/Retry.java
interface Context {
/**
* Records a successful call.
*/
void onSuccess();
/**
* Handles a checked exception
*
* @param exception the exception to handle
* @throws Throwable the exception
*/
void onError(Exception exception) throws Throwable;
/**
* Handles a runtime exception
*
* @param runtimeException the exception to handle
*/
void onRuntimeError(RuntimeException runtimeException);
}
- 这个接口定义了onSuccess、onError、onRuntimeError
RetryImpl.ContextImpl
resilience4j-retry-0.13.0-sources.jar!/io/github/resilience4j/retry/internal/RetryImpl.java
public final class ContextImpl implements Retry.Context {
private final AtomicInteger numOfAttempts = new AtomicInteger(0);
private final AtomicReference<Exception> lastException = new AtomicReference<>();
private final AtomicReference<RuntimeException> lastRuntimeException = new AtomicReference<>();
private ContextImpl() {
}
public void onSuccess() {
int currentNumOfAttempts = numOfAttempts.get();
if(currentNumOfAttempts > 0){
succeededAfterRetryCounter.increment();
Throwable throwable = Option.of(lastException.get()).getOrElse(lastRuntimeException.get());
publishRetryEvent(() -> new RetryOnSuccessEvent(getName(), currentNumOfAttempts, throwable));
}else{
succeededWithoutRetryCounter.increment();
}
}
public void onError(Exception exception) throws Throwable{
if(exceptionPredicate.test(exception)){
lastException.set(exception);
throwOrSleepAfterException();
}else{
failedWithoutRetryCounter.increment();
publishRetryEvent(() -> new RetryOnIgnoredErrorEvent(getName(), exception));
throw exception;
}
}
public void onRuntimeError(RuntimeException runtimeException){
if(exceptionPredicate.test(runtimeException)){
lastRuntimeException.set(runtimeException);
throwOrSleepAfterRuntimeException();
}else{
failedWithoutRetryCounter.increment();
publishRetryEvent(() -> new RetryOnIgnoredErrorEvent(getName(), runtimeException));
throw runtimeException;
}
}
private void throwOrSleepAfterException() throws Exception {
int currentNumOfAttempts = numOfAttempts.incrementAndGet();
Exception throwable = lastException.get();
if(currentNumOfAttempts >= maxAttempts){
failedAfterRetryCounter.increment();
publishRetryEvent(() -> new RetryOnErrorEvent(getName(), currentNumOfAttempts, throwable));
throw throwable;
}else{
waitIntervalAfterFailure(currentNumOfAttempts, throwable);
}
}
private void throwOrSleepAfterRuntimeException(){
int currentNumOfAttempts = numOfAttempts.incrementAndGet();
RuntimeException throwable = lastRuntimeException.get();
if(currentNumOfAttempts >= maxAttempts){
failedAfterRetryCounter.increment();
publishRetryEvent(() -> new RetryOnErrorEvent(getName(), currentNumOfAttempts, throwable));
throw throwable;
}else{
waitIntervalAfterFailure(currentNumOfAttempts, throwable);
}
}
private void waitIntervalAfterFailure(int currentNumOfAttempts, Throwable throwable) {
// wait interval until the next attempt should start
long interval = intervalFunction.apply(numOfAttempts.get());
publishRetryEvent(()-> new RetryOnRetryEvent(getName(), currentNumOfAttempts, throwable, interval));
Try.run(() -> sleepFunction.accept(interval))
.getOrElseThrow(ex -> lastRuntimeException.get());
}
}
- throwOrSleepAfterRuntimeException方法会根据重试次数判断，如果超出则抛lastRuntimeException，如果不超出则调用waitIntervalAfterFailure
- waitIntervalAfterFailure则通过sleepFunction来进行延时
小结
- resilience4j的Retry沿用了该组件的一贯风格，通过decorate方法来织入重试的逻辑
- 重试的逻辑就是一个while true循环，先执行业务方法，如果成功则调用Retry.Context的onSuccess方法，然后跳出循环，如果失败的话，只捕获RuntimeException，然后调用Retry.Context的onRuntimeError
- onRuntimeError会先判断该异常是否需要重试，如果不需要则直接抛出原有异常，需要重试的话，则numOfAttempts.incrementAndGet()，如果超出限制则抛出异常，没有超出限制则根据配置的重试间隔进行sleep，然后onRuntimeError返回继续下一循环重试。
