mirror of
https://gitee.com/dromara/hutool.git
synced 2025-04-24 18:04:54 +08:00
添加重试工具
This commit is contained in:
kongweiguang
parent
dd79ee4b62
commit
c317ef1e08
@ -0,0 +1,205 @@
|
||||
package org.dromara.hutool.core.thread;
|
||||
|
||||
import org.dromara.hutool.core.lang.Assert;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Arrays;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.function.BiPredicate;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import static java.util.Objects.nonNull;
|
||||
|
||||
/**
|
||||
* 重试任务类
|
||||
*
|
||||
* @author kongweiguang
|
||||
* @since 6.0.0
|
||||
*/
|
||||
public class RetryableTask<T> {
|
||||
/**
|
||||
* 执行结果
|
||||
*/
|
||||
private T result;
|
||||
/**
|
||||
* 执行法方法
|
||||
*/
|
||||
private final Supplier<T> sup;
|
||||
/**
|
||||
* 重试策略
|
||||
*/
|
||||
private final BiPredicate<T, Throwable> predicate;
|
||||
/**
|
||||
* 重试次数，默认3次
|
||||
*/
|
||||
private long maxAttempts = 3;
|
||||
/**
|
||||
* 重试间隔，默认1秒
|
||||
*/
|
||||
private Duration delay = Duration.ofSeconds(1);
|
||||
|
||||
/**
|
||||
* 构造方法，内部使用，调用请使用请用ofXXX
|
||||
*
|
||||
* @param sup 执行的方法
|
||||
* @param predicate 策略 {@link BiPredicate}
|
||||
*/
|
||||
private RetryableTask(Supplier<T> sup, BiPredicate<T, Throwable> predicate) {
|
||||
Assert.notNull(sup, "task parameter cannot be null");
|
||||
Assert.notNull(predicate, "predicate parameter cannot be null");
|
||||
|
||||
this.predicate = predicate;
|
||||
this.sup = sup;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 重试根据指定的异常，没有返回值
|
||||
*
|
||||
* @param <T> 返回值类型
|
||||
* @param run 执行的方法 {@link Runnable}
|
||||
* @param ths 指定异常 {@link Throwable}
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
@SafeVarargs
|
||||
public static <T> RetryableTask<T> retryForExceptions(Runnable run, Class<? extends Throwable>... ths) {
|
||||
return retryForExceptions(() -> {
|
||||
run.run();
|
||||
return null;
|
||||
}, ths);
|
||||
}
|
||||
|
||||
/**
|
||||
* 重试根据指定的策略，没有返回值
|
||||
*
|
||||
* @param <T> 返回值类型
|
||||
* @param run 执行的方法 {@link Runnable}
|
||||
* @param predicate 策略 {@link BiPredicate}
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
public static <T> RetryableTask<T> retryForPredicate(Runnable run, BiPredicate<T, Throwable> predicate) {
|
||||
return retryForPredicate(() -> {
|
||||
run.run();
|
||||
return null;
|
||||
}, predicate);
|
||||
}
|
||||
|
||||
/**
|
||||
* 重试根据指定的异常，有返回值
|
||||
*
|
||||
* @param <T> 返回值类型
|
||||
* @param sup 执行的方法 {@link Supplier}
|
||||
* @param ths 指定异常 {@link Throwable}
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
@SafeVarargs
|
||||
public static <T> RetryableTask<T> retryForExceptions(Supplier<T> sup, Class<? extends Throwable>... ths) {
|
||||
Assert.isTrue(ths.length != 0, "exs cannot be empty");
|
||||
|
||||
BiPredicate<T, Throwable> strategy = (t, e) -> {
|
||||
if (nonNull(e)) {
|
||||
return Arrays.stream(ths).anyMatch(ex -> ex.isAssignableFrom(e.getClass()));
|
||||
}
|
||||
return false;
|
||||
};
|
||||
|
||||
return new RetryableTask<>(sup, strategy);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 重试根据指定的策略，没有返回值
|
||||
*
|
||||
* @param <T> 返回值类型
|
||||
* @param sup 执行的方法 {@link Supplier}
|
||||
* @param predicate 策略 {@link BiPredicate}
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
public static <T> RetryableTask<T> retryForPredicate(Supplier<T> sup, BiPredicate<T, Throwable> predicate) {
|
||||
return new RetryableTask<>(sup, predicate);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 最大重试次数
|
||||
*
|
||||
* @param maxAttempts 次数
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
public RetryableTask<T> maxAttempts(long maxAttempts) {
|
||||
Assert.isTrue(this.maxAttempts > 0, "maxAttempts must be greater than 0");
|
||||
|
||||
this.maxAttempts = maxAttempts;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 重试间隔时间
|
||||
*
|
||||
* @param delay 间隔时间
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
public RetryableTask<T> delay(Duration delay) {
|
||||
Assert.notNull(this.delay, "delay parameter cannot be null");
|
||||
|
||||
this.delay = delay;
|
||||
return this;
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取结果
|
||||
*
|
||||
* @return 返回包装了结果的 {@link Optional}对象
|
||||
*/
|
||||
public Optional<T> get() {
|
||||
return Optional.ofNullable(this.result);
|
||||
}
|
||||
|
||||
/**
|
||||
* 异步执行重试方法
|
||||
*
|
||||
* @return 返回一个异步对象 {@link CompletableFuture}
|
||||
*/
|
||||
public CompletableFuture<RetryableTask<T>> asyncExecute() {
|
||||
return CompletableFuture.supplyAsync(this::doExecute, GlobalThreadPool.getExecutor());
|
||||
}
|
||||
|
||||
/**
|
||||
* 同步执行重试方法
|
||||
*
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
*/
|
||||
public RetryableTask<T> execute() {
|
||||
return doExecute();
|
||||
}
|
||||
|
||||
/**
|
||||
* 开始重试
|
||||
*
|
||||
* @return 当前对象 {@link RetryableTask}
|
||||
**/
|
||||
private RetryableTask<T> doExecute() {
|
||||
Throwable th = null;
|
||||
|
||||
while (--this.maxAttempts >= 0) {
|
||||
|
||||
try {
|
||||
this.result = this.sup.get();
|
||||
} catch (Throwable t) {
|
||||
th = t;
|
||||
} finally {
|
||||
//判断重试
|
||||
if (this.predicate.test(this.result, th)) {
|
||||
ThreadUtil.sleep(delay.toMillis());
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return this;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -0,0 +1,148 @@
|
||||
package org.dromara.hutool.core.util;
|
||||
|
||||
import org.dromara.hutool.core.array.ArrayUtil;
|
||||
import org.dromara.hutool.core.thread.GlobalThreadPool;
|
||||
import org.dromara.hutool.core.thread.RetryableTask;
|
||||
import org.dromara.hutool.core.thread.ThreadUtil;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.function.BiPredicate;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/**
|
||||
* 重试工具类
|
||||
* 自定义功能请使用{@link RetryableTask}类
|
||||
*
|
||||
* @author kongweiguang
|
||||
* @see RetryableTask
|
||||
* @since 6.0.0
|
||||
*/
|
||||
public class RetryUtil {
|
||||
|
||||
/**
|
||||
* 根据异常信息进行重试
|
||||
* 没有返回值，重试执行方法
|
||||
*
|
||||
* @param run 执行方法
|
||||
* @param maxAttempts 最大的重试次数
|
||||
* @param delay 重试间隔
|
||||
* @param recover 达到最大重试次数后执行的备用方法，入参是重试过程中的异常
|
||||
* @param exs 指定的异常类型需要重试
|
||||
*/
|
||||
@SafeVarargs
|
||||
public static void ofException(Runnable run, long maxAttempts, Duration delay, Runnable recover, Class<? extends Throwable>... exs) {
|
||||
if (ArrayUtil.isEmpty(exs)) {
|
||||
exs = ArrayUtil.append(exs, RuntimeException.class);
|
||||
}
|
||||
RetryableTask.retryForExceptions(run, exs)
|
||||
.maxAttempts(maxAttempts)
|
||||
.delay(delay)
|
||||
.execute()
|
||||
.get()
|
||||
.orElseGet(() -> {
|
||||
recover.run();
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据异常信息进行重试
|
||||
* 有返回值，重试执行方法
|
||||
*
|
||||
* @param sup 执行方法
|
||||
* @param maxAttempts 最大的重试次数
|
||||
* @param delay 重试间隔
|
||||
* @param recover 达到最大重试次数后执行的备用方法，入参是重试过程中的异常
|
||||
* @param exs 指定的异常类型需要重试
|
||||
* @param <T> 结果类型
|
||||
* @return 执行结果
|
||||
*/
|
||||
@SafeVarargs
|
||||
public static <T> T ofException(Supplier<T> sup, long maxAttempts, Duration delay, Supplier<T> recover, Class<? extends Throwable>... exs) {
|
||||
if (ArrayUtil.isEmpty(exs)) {
|
||||
exs = ArrayUtil.append(exs, RuntimeException.class);
|
||||
}
|
||||
return RetryableTask.retryForExceptions(sup, exs)
|
||||
.maxAttempts(maxAttempts)
|
||||
.delay(delay)
|
||||
.execute()
|
||||
.get()
|
||||
.orElseGet(recover);
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据自定义结果进行重试
|
||||
* 没有返回值，重试执行方法
|
||||
*
|
||||
* @param run 执行方法
|
||||
* @param maxAttempts 最大的重试次数
|
||||
* @param delay 重试间隔
|
||||
* @param recover 达到最大重试次数后执行的备用方法，入参是重试过程中的异常
|
||||
* @param predicate 自定义重试条件
|
||||
*/
|
||||
public static void ofPredicate(Runnable run, long maxAttempts, Duration delay, Supplier<Void> recover, BiPredicate<Void, Throwable> predicate) {
|
||||
RetryableTask.retryForPredicate(run, predicate)
|
||||
.delay(delay)
|
||||
.maxAttempts(maxAttempts)
|
||||
.execute()
|
||||
.get()
|
||||
.orElseGet(recover);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 根据异常信息进行重试
|
||||
* 有返回值，重试执行方法
|
||||
*
|
||||
* @param sup 执行方法
|
||||
* @param maxAttempts 最大的重试次数
|
||||
* @param delay 重试间隔
|
||||
* @param recover 达到最大重试次数后执行的备用方法，入参是重试过程中的异常
|
||||
* @param predicate 自定义重试条件
|
||||
* @param <T> 结果类型
|
||||
* @return 执行结果
|
||||
*/
|
||||
public static <T> T ofPredicate(Supplier<T> sup, long maxAttempts, Duration delay, Supplier<T> recover, BiPredicate<T, Throwable> predicate) {
|
||||
return RetryableTask.retryForPredicate(sup, predicate)
|
||||
.delay(delay)
|
||||
.maxAttempts(maxAttempts)
|
||||
.execute()
|
||||
.get()
|
||||
.orElseGet(recover);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 从不停止的执行方法
|
||||
*
|
||||
* @param run 执行方法
|
||||
* @param delay 间隔时间
|
||||
* @param isEx true：出现异常继续执行；false：则出现异常跳出执行。
|
||||
*/
|
||||
public static void ofNeverStop(Runnable run, Duration delay, boolean isEx) {
|
||||
while (true) {
|
||||
try {
|
||||
run.run();
|
||||
} catch (Throwable e) {
|
||||
if (!isEx) {
|
||||
break;
|
||||
}
|
||||
} finally {
|
||||
ThreadUtil.sleep(delay.toMillis());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 从不停止的执行方法，异步执行
|
||||
*
|
||||
* @param run 执行方法
|
||||
* @param delay 间隔时间
|
||||
* @param isEx true：出现异常继续执行；false：则出现异常跳出执行。
|
||||
*/
|
||||
public static void ofNeverStopAsync(Runnable run, Duration delay, boolean isEx) {
|
||||
CompletableFuture.runAsync(() -> ofNeverStop(run, delay, isEx), GlobalThreadPool.getExecutor());
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,91 @@
|
||||
package org.dromara.hutool.core.util;
|
||||
|
||||
import org.dromara.hutool.core.thread.RetryableTask;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
public class RetryUtilTest {
|
||||
|
||||
@Test
|
||||
void test() {
|
||||
//自定义根据异常重试
|
||||
CompletableFuture<RetryableTask<String>> task = RetryableTask.retryForExceptions(() -> {
|
||||
System.out.println("1231231");
|
||||
int a = 1 / 0;
|
||||
return "qqqq";
|
||||
}, ArithmeticException.class)
|
||||
.delay(Duration.ofSeconds(1))
|
||||
.maxAttempts(3)
|
||||
.asyncExecute();
|
||||
|
||||
Assertions.assertFalse(task.isDone());
|
||||
Assertions.assertEquals("兜底", task.join().get().orElseGet(() -> {
|
||||
return "兜底";
|
||||
}));
|
||||
Assertions.assertTrue(task.isDone());
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void noReturnTest() {
|
||||
//根据异常重试，没有返回值
|
||||
RetryUtil.ofException(
|
||||
() -> {
|
||||
System.out.println(123);
|
||||
int a = 1 / 0;
|
||||
},
|
||||
3,
|
||||
Duration.ofSeconds(1),
|
||||
() -> {
|
||||
System.out.println("兜底");
|
||||
},
|
||||
ArithmeticException.class
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void hasReturnTest() {
|
||||
//根据自定义策略重试
|
||||
String result = RetryUtil.ofPredicate(
|
||||
() -> {
|
||||
System.out.println(123);
|
||||
// int a = 1 / 0;
|
||||
return "ok";
|
||||
},
|
||||
5,
|
||||
Duration.ofSeconds(1),
|
||||
() -> {
|
||||
System.out.println("兜底");
|
||||
return "do";
|
||||
},
|
||||
(r, e) -> {
|
||||
System.out.println("r = " + r);
|
||||
System.out.println("e = " + e);
|
||||
return r.equals("ok");
|
||||
}
|
||||
);
|
||||
|
||||
Assertions.assertEquals("ok", result);
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void neverStop() {
|
||||
//异步一直执行
|
||||
RetryUtil.ofNeverStopAsync(() -> {
|
||||
System.out.println("async -->");
|
||||
}, Duration.ofSeconds(1), true);
|
||||
|
||||
System.out.println(" ================ ");
|
||||
//同步一直执行
|
||||
RetryUtil.ofNeverStop(() -> {
|
||||
System.out.println(123);
|
||||
}, Duration.ofSeconds(3), true);
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user