zxcvbnmzsedr
4/4/2019 - 6:44 AM
重试工具类
package com.cnabc.base.common.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;
import java.util.List;
import java.util.concurrent.*;
/**
* RetryUtil重试执行工具类
* Created by Suo-Long
* Date：2017/4/21
* Time：14:22
*/
public class RetryUtil {
private static final Logger LOGGER = LoggerFactory.getLogger(RetryUtil.class);
private static final long MAX_SLEEP_MILLISECOND = 256 * 1000;
/**
* 重试次数工具方法.
*
* @param callable 实际逻辑
* @param retryTimes 最大重试次数（>1）
* @param sleepTimeInMilliSecond 运行失败后休眠对应时间再重试
* @param exponential 休眠时间是否指数递增
* @param <T> 返回值类型
* @return 经过重试的callable的执行结果
*/
public static <T> T executeWithRetry(Callable<T> callable,
int retryTimes,
long sleepTimeInMilliSecond,
boolean exponential) throws Exception {
Retry retry = new Retry();
return retry.doRetry(callable, retryTimes, sleepTimeInMilliSecond, exponential, null);
}
/**
* 重试次数工具方法.
*
* @param callable 实际逻辑
* @param retryTimes 最大重试次数（>1）
* @param sleepTimeInMilliSecond 运行失败后休眠对应时间再重试
* @param exponential 休眠时间是否指数递增
* @param <T> 返回值类型
* @param retryExceptionClasss 出现指定的异常类型时才进行重试
* @return 经过重试的callable的执行结果
*/
public static <T> T executeWithRetry(Callable<T> callable,
int retryTimes,
long sleepTimeInMilliSecond,
boolean exponential,
List<Class<?>> retryExceptionClasss) throws Exception {
Retry retry = new Retry();
return retry.doRetry(callable, retryTimes, sleepTimeInMilliSecond, exponential, retryExceptionClasss);
}
/**
* 在外部线程执行并且重试。每次执行需要在timeoutMs内执行完，不然视为失败。
* 执行异步操作的线程池从外部传入，线程池的共享粒度由外部控制。比如，HttpClientUtil共享一个线程池。
* <p/>
* 限制条件：仅仅能够在阻塞的时候interrupt线程
*
* @param callable 实际逻辑
* @param retryTimes 最大重试次数（>1）
* @param sleepTimeInMilliSecond 运行失败后休眠对应时间再重试
* @param exponential 休眠时间是否指数递增
* @param timeoutMs callable执行超时时间，毫秒
* @param executor 执行异步操作的线程池
* @param <T> 返回值类型
* @return 经过重试的callable的执行结果
*/
public static <T> T asyncExecuteWithRetry(Callable<T> callable,
int retryTimes,
long sleepTimeInMilliSecond,
boolean exponential,
long timeoutMs,
ThreadPoolExecutor executor) throws Exception {
Retry retry = new AsyncRetry(timeoutMs, executor);
return retry.doRetry(callable, retryTimes, sleepTimeInMilliSecond, exponential, null);
}
/**
* 创建异步执行的线程池。特性如下：
* core大小为0，初始状态下无线程，无初始消耗。
* max大小为5，最多五个线程。
* 60秒超时时间，闲置超过60秒线程会被回收。
* 使用SynchronousQueue，任务不会排队，必须要有可用线程才能提交成功，否则会RejectedExecutionException。
*
* @return 线程池
*/
public static ThreadPoolExecutor createThreadPoolExecutor() {
return new ThreadPoolExecutor(0, 5,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
private static class Retry {
public <T> T doRetry(Callable<T> callable, int retryTimes, long sleepTimeInMilliSecond, boolean exponential, List<Class<?>> retryExceptionClasss)
throws Exception {
if (null == callable) {
throw new IllegalArgumentException("系统编程错误, 入参callable不能为空 ! ");
}
if (retryTimes < 1) {
throw new IllegalArgumentException(String.format(
"系统编程错误, 入参retrytime[%d]不能小于1 !", retryTimes));
}
Exception saveException = null;
for (int i = 0; i < retryTimes; i++) {
System.out.println(i);
try {
return call(callable);
} catch (Exception e) {
saveException = e;
if (i == 0) {
LOGGER.error(String.format("Exception when calling callable, 异常Msg:%s", saveException.getMessage()), saveException);
}
if (!CollectionUtils.isEmpty(retryExceptionClasss)) {
boolean needRetry = false;
for (Class<?> eachExceptionClass : retryExceptionClasss) {
if (eachExceptionClass == e.getClass()) {
needRetry = true;
break;
}
}
if (!needRetry) {
throw saveException;
}
}
if (i + 1 < retryTimes && sleepTimeInMilliSecond > 0) {
long startTime = System.currentTimeMillis();
long timeToSleep;
if (exponential) {
timeToSleep = sleepTimeInMilliSecond * (long) Math.pow(2, i);
if(timeToSleep >= MAX_SLEEP_MILLISECOND) {
timeToSleep = MAX_SLEEP_MILLISECOND;
}
} else {
timeToSleep = sleepTimeInMilliSecond;
if(timeToSleep >= MAX_SLEEP_MILLISECOND) {
timeToSleep = MAX_SLEEP_MILLISECOND;
}
}
try {
Thread.sleep(timeToSleep);
} catch (InterruptedException ignored) {
}
long realTimeSleep = System.currentTimeMillis()-startTime;
LOGGER.error(String.format("Exception when calling callable, 即将尝试执行第%s次重试.本次重试计划等待[%s]ms,实际等待[%s]ms, 异常Msg:[%s]",
i+1, timeToSleep,realTimeSleep, e.getMessage()));
}
}
}
throw saveException;
}
protected <T> T call(Callable<T> callable) throws Exception {
return callable.call();
}
}
private static class AsyncRetry extends Retry {
private long timeoutMs;
private ThreadPoolExecutor executor;
public AsyncRetry(long timeoutMs, ThreadPoolExecutor executor) {
this.timeoutMs = timeoutMs;
this.executor = executor;
}
/**
* 使用传入的线程池异步执行任务，并且等待。
* <p/>
* future.get()方法，等待指定的毫秒数。如果任务在超时时间内结束，则正常返回。
* 如果抛异常（可能是执行超时、执行异常、被其他线程cancel或interrupt），都记录日志并且网上抛异常。
* 正常和非正常的情况都会判断任务是否结束，如果没有结束，则cancel任务。cancel参数为true，表示即使
* 任务正在执行，也会interrupt线程。
*
* @param callable
* @param <T>
* @return
* @throws Exception
*/
@Override
protected <T> T call(Callable<T> callable) throws Exception {
Future<T> future = executor.submit(callable);
try {
return future.get(timeoutMs, TimeUnit.MILLISECONDS);
} catch (Exception e) {
LOGGER.warn("Try once failed", e);
throw e;
} finally {
if (!future.isDone()) {
future.cancel(true);
LOGGER.warn("Try once task not done, cancel it, active count: " + executor.getActiveCount());
}
}
}
}
}