feat: 添加重试器, 并应用到身份配置加载中(有待测试).

本提交添加了一个 Retryer, 用于增加对失败加载身份配置时, 可进行重试.
请注意: 对于将该功能整合到身份配置的加载, 其状况有待测试.

issue #5
This commit is contained in:
LamGC 2021-10-26 19:57:20 +08:00
parent bded118c03
commit 1cdc15a05f
Signed by: LamGC
GPG Key ID: 6C5AE2A913941E1D
13 changed files with 449 additions and 12 deletions

View File

@ -0,0 +1,14 @@
package net.lamgc.oracle.sentry.common;
import com.oracle.bmc.model.BmcException;
import net.lamgc.oracle.sentry.common.retry.RetryExceptionHandler;
public class OracleBmcExceptionHandler implements RetryExceptionHandler {
@Override
public boolean handle(Exception e) {
if (e instanceof BmcException bmc) {
return bmc.getStatusCode() == -1;
}
return true;
}
}

View File

@ -0,0 +1,14 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 断言校验器.
* <p> 校验器可帮助 {@link Retryer} 检查执行结果是否符合预期, 如果不符合预期, 那么将会重新执行任务.
* @param <R> 执行结果类型.
* @author LamGC
*/
@FunctionalInterface
public interface AssertionChecker<R> {
void check(R result) throws RetryAssertException;
}

View File

@ -0,0 +1,14 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 指数退避延迟器.
* @author LamGC
*/
public final class ExponentialBackoffDelayer implements RetryDelayer {
@Override
public long nextDelayTime(int currentRetryCount) {
return (4L << currentRetryCount) * 1000;
}
}

View File

@ -0,0 +1,26 @@
package net.lamgc.oracle.sentry.common.retry;
import java.util.concurrent.TimeUnit;
/**
* 固定延迟器.
* <p> 永远以指定的延迟重试.
* @author LamGC
*/
public final class FixedTimeDelayer implements RetryDelayer {
private final long delay;
public FixedTimeDelayer(long time, TimeUnit unit) {
this(unit.toMillis(time));
}
public FixedTimeDelayer(long delay) {
this.delay = delay;
}
@Override
public long nextDelayTime(int currentRetryCount) {
return delay;
}
}

View File

@ -0,0 +1,27 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 非空检查器.
* <p> 当执行结果为 null 时断言失败.
* @param <R>
* @author LamGC
*/
public final class NonNullChecker<R> implements AssertionChecker<R> {
@SuppressWarnings("rawtypes")
private final static NonNullChecker INSTANCE = new NonNullChecker();
@SuppressWarnings("unchecked")
public static <R> NonNullChecker<R> getInstance() {
return (NonNullChecker<R>) INSTANCE;
}
private NonNullChecker() {}
@Override
public void check(Object result) throws RetryAssertException {
if (result == null) {
throw new RetryAssertException("The execution result is null.");
}
}
}

View File

@ -0,0 +1,18 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 重试断言异常.
* <p> 当指定条件失败时抛出该异常.
* <p> 跑出该异常的原因并非任务执行失败, 而是任务执行的结果与预期不符.
* @author LamGC
*/
public class RetryAssertException extends Exception {
public RetryAssertException(String message) {
super(message);
}
public RetryAssertException(String message, Throwable cause) {
super(message, cause);
}
}

View File

@ -0,0 +1,17 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 重试延迟器.
* <p> 用于决定每次重试的间隔时间, 可根据重试次数调整间隔时常, 以避免频繁执行影响性能.
* @author LamGC
*/
public interface RetryDelayer {
/**
* 获取下一次重试延迟时间.
* @param currentRetryCount 当前重试次数, 如果第一次重试失败, 则本参数为 0.
* @return 返回延迟时间.
*/
long nextDelayTime(int currentRetryCount);
}

View File

@ -0,0 +1,13 @@
package net.lamgc.oracle.sentry.common.retry;
@FunctionalInterface
public interface RetryExceptionHandler {
/**
* 处理异常, 并指示是否继续重试.
* @param e 异常对象.
* @return 如果可以继续重试, 返回 {@code true}.
*/
boolean handle(Exception e);
}

View File

@ -0,0 +1,20 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 重试失败异常.
* <p> 该异常是由于某一个任务尝试执行失败过多而抛出, 失败的原因可能是任务执行时抛出异常, 或执行结果与预期不符.
*/
public final class RetryFailedException extends Exception {
public RetryFailedException(String message) {
super(message);
}
public RetryFailedException(String message, Throwable cause) {
super(message, cause);
}
public RetryFailedException(Throwable cause) {
super(cause);
}
}

View File

@ -0,0 +1,20 @@
package net.lamgc.oracle.sentry.common.retry;
/**
* 可重试任务.
* <p> 实现该方法后, 该任务可在 {@link Retryer} 运行, 可用于一些需重试的任务.
* @param <R> 执行结果类型.
*/
@FunctionalInterface
public interface RetryableTask<R> {
/**
* 运行方法.
* <p> 当该方法抛出异常, 或经 {@link AssertionChecker} 检查认为结果与预期不符时, 将会被重新运行.
* <p> 请注意, 即使任务执行完成, {@link AssertionChecker} 认为结果与预期不符, 任务将会被重新运行, 请注意处理该情况.
* @throws Exception 当异常抛出时, 将视为执行失败, 重试器将根据设定自动重新执行.
* @return 根据需要可返回.
*/
R run() throws Exception;
}

View File

@ -0,0 +1,177 @@
package net.lamgc.oracle.sentry.common.retry;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.*;
import java.util.concurrent.*;
/**
* 重试器.
* <p> 通过重试器, 可以对某一个可能失败的任务做重试, 尽可能确保任务执行成功.
* @param <R> 任务结果类型.
* @author LamGC
*/
public final class Retryer<R> {
private final static ThreadPoolExecutor ASYNC_EXECUTOR = new ThreadPoolExecutor(
1,
Math.min(4, Math.max(1, Runtime.getRuntime().availableProcessors() / 2)),
10, TimeUnit.SECONDS,
new LinkedBlockingDeque<>(),
new ThreadFactoryBuilder()
.setNameFormat("Thread-Retryer-%d")
.build()
);
private final RetryableTask<R> task;
private final int retryNumber;
private final RetryDelayer delayer;
private final Set<AssertionChecker<R>> checkers = new HashSet<>();
private final RetryExceptionHandler handler;
private Retryer(RetryableTask<R> task, int retryNumber, RetryDelayer delayer, Set<AssertionChecker<R>> checkers, RetryExceptionHandler handler) {
this.handler = handler;
if (retryNumber < 0) {
throw new IllegalArgumentException("The number of retries is not allowed to be negative: " + retryNumber);
}
this.task = Objects.requireNonNull(task);
this.delayer = Objects.requireNonNull(delayer);
this.retryNumber = retryNumber;
if (checkers != null && !checkers.isEmpty()) {
this.checkers.addAll(checkers);
}
}
@SuppressWarnings("BusyWait")
private R execute0() throws Exception {
Exception lastException;
int currentRetryCount = 0;
do {
try {
R result = task.run();
checkResult(result);
return result;
} catch (Exception e) {
lastException = e;
if (e instanceof InterruptedException) {
break;
}
if (handler != null && !handler.handle(e)) {
break;
}
if (currentRetryCount >= retryNumber) {
break;
}
long delayTime = delayer.nextDelayTime(currentRetryCount);
if (delayTime > 0) {
try {
Thread.sleep(delayTime);
} catch (InterruptedException interrupted) {
break;
}
}
currentRetryCount ++;
}
} while (true);
throw lastException;
}
/**
* 使用 {@link AssertionChecker} 检查结果是否符合预期.
* <p> 当结果不符合检验器预期时, 检验器将抛出 {@link RetryAssertException} 来表示结果不符预期,
* {@link Retryer} 将会重试该任务.
* @param result 执行结果.
* @throws RetryAssertException 当断言检验器对结果断言失败时抛出该异常.
*/
private void checkResult(R result) throws RetryAssertException {
for (AssertionChecker<R> checker : checkers) {
checker.check(result);
}
}
/**
* 异步执行任务.
* <p> 使用线程池执行 task.
* @return 返回 Future 对象以跟踪异步执行结果.
*/
public Future<R> executeAsync() {
return ASYNC_EXECUTOR.submit(this::execute0);
}
/**
* 同步执行任务.
* @return 如果执行完成且成功, 返回执行结果.
* @throws RetryFailedException 当重试多次仍失败时抛出该异常.
*/
public R execute() throws RetryFailedException {
Future<R> future = executeAsync();
try {
return future.get();
} catch (InterruptedException | ExecutionException e) {
if (e instanceof ExecutionException exception) {
throw new RetryFailedException("Failure after a certain number of attempts.", exception);
} else {
throw new RetryFailedException(e);
}
}
}
/**
* 获取一个构建器.
* @param task 需要重试的任务.
* @param <R> 结果类型.
* @return 返回新的构造器.
*/
public static <R> Builder<R> builder(RetryableTask<R> task) {
return new Builder<>(task);
}
/**
* {@link Retryer} 构造器.
* <p> 可通过链式调用快速创建 {@link Retryer}.
* @param <R> 任务结果类型.
*/
public static class Builder<R> {
private final RetryableTask<R> task;
private RetryDelayer delayer = new FixedTimeDelayer(0);
private int retryNumber = 0;
private final Set<AssertionChecker<R>> checkers = new HashSet<>();
private RetryExceptionHandler handler = (e) -> true;
private Builder(RetryableTask<R> task) {
this.task = task;
}
public Retryer<R> create() {
return new Retryer<>(task, retryNumber, delayer, checkers, handler);
}
public Builder<R> delayer(RetryDelayer delayer) {
this.delayer = delayer;
return this;
}
public Builder<R> retryIfReturnNull() {
this.checkers.add(NonNullChecker.getInstance());
return this;
}
public Builder<R> retryNumber(int number) {
this.retryNumber = number;
return this;
}
public Builder<R> checker(AssertionChecker<R> checker) {
checkers.add(checker);
return this;
}
public Builder<R> exceptionHandler(RetryExceptionHandler handler) {
this.handler = handler == null ? (e) -> true : handler;
return this;
}
}
}

View File

@ -7,6 +7,9 @@ import com.oracle.bmc.Region;
import com.oracle.bmc.auth.AuthenticationDetailsProvider;
import com.oracle.bmc.auth.SimpleAuthenticationDetailsProvider;
import com.oracle.bmc.auth.SimplePrivateKeySupplier;
import net.lamgc.oracle.sentry.common.OracleBmcExceptionHandler;
import net.lamgc.oracle.sentry.common.retry.ExponentialBackoffDelayer;
import net.lamgc.oracle.sentry.common.retry.Retryer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -18,7 +21,8 @@ import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@ -55,23 +59,39 @@ public final class OracleAccountManager {
if (configFiles == null) {
throw new IOException("Unable to access the specified directory: " + directory.getCanonicalPath());
}
int loadedCount = 0;
AtomicInteger loadedCount = new AtomicInteger();
for (File configFile : configFiles) {
try {
OracleAccount account = loadFromConfigFile(configFile);
if (account == null) {
Retryer<OracleAccount> retryer = Retryer.builder(() -> {
OracleAccount account = loadFromConfigFile(configFile);
if (account == null) {
return null;
}
log.info("已成功加载身份配置文件.\n\tUserId: {}\n\tUsername: {}\n\tPath: {}",
account.id(),
account.name(),
configFile.getCanonicalPath());
return account;
})
.delayer(new ExponentialBackoffDelayer())
.exceptionHandler(new OracleBmcExceptionHandler())
.retryNumber(8)
.create();
Future<OracleAccount> future = retryer.executeAsync();
future.get(30, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e instanceof TimeoutException) {
continue;
} else if (e instanceof InterruptedException) {
Thread.currentThread().interrupt();
break;
}
loadedCount ++;
log.info("已成功加载身份配置文件.\n\tUserId: {}\n\tUsername: {}\n\tPath: {}",
account.id(),
account.name(),
configFile.getCanonicalPath());
} catch (Exception e) {
log.error("加载身份配置文件时发生异常.(Path: {})\n{}", configFile.getCanonicalPath(), Throwables.getStackTraceAsString(e));
log.error("加载身份配置文件时发生异常.(Path: {})\n{}", configFile.getCanonicalPath(), Throwables.getStackTraceAsString(e.getCause()));
}
}
return loadedCount;
return loadedCount.get();
}
/**

View File

@ -0,0 +1,57 @@
package net.lamgc.oracle.sentry.common.retry;
import org.junit.jupiter.api.Test;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class RetryerTest {
@Test
void successTest() throws RetryFailedException, ExecutionException, InterruptedException {
final Object obj = new Object();
Retryer<Object> retryer = Retryer.builder(() -> obj).create();
assertEquals(obj, retryer.execute());
assertEquals(obj, retryer.executeAsync().get());
}
@Test
void failedTest() {
assertThrows(RetryFailedException.class, () -> {
Retryer<Object> retryer = Retryer.builder(() -> {
throw new RuntimeException();
}).create();
retryer.execute();
});
}
@Test
void retryNumberTest() {
final int retryNumber = new Random().nextInt(9) + 1;
final AtomicInteger retryCounter = new AtomicInteger(-1);
Retryer<Object> retryer = Retryer.builder(() -> {
retryCounter.incrementAndGet();
throw new RuntimeException();
}).retryNumber(retryNumber).create();
try {
retryer.execute();
} catch (RetryFailedException e) {
e.printStackTrace();
}
assertEquals(retryNumber, retryCounter.get());
}
@Test
void checkerTest() {
Retryer<Object> retryer = Retryer.builder(() -> null)
.retryIfReturnNull()
.create();
assertThrows(RetryFailedException.class, retryer::execute);
}
}