总第509篇
2022年 第026篇
0 背景
1 为何需要并行加载
2 并行加载的实现方式
2.1 同步模型
2.2 NIO异步模型
2.3 为什么会选择CompletableFuture?
3 CompletableFuture使用与原理
3.1 CompletableFuture的背景和定义
3.2 CompletableFuture的使用
3.3 CompletableFuture原理
4 实践总结
4.1 线程阻塞问题
4.2 线程池须知
4.3 其他
5 异步化收益
6 参考文献
7 名词解释及备注
8 附录
ExecutorService executor = Executors.newFixedThreadPool(5);
ListeningExecutorService guavaExecutor = MoreExecutors.listeningDecorator(executor);ListenableFuture<String> future1 = guavaExecutor.submit(() -> {
//step 1
System.out.println("执行step 1");
return "step1 result";
});
ListenableFuture<String> future2 = guavaExecutor.submit(() -> {
//step 2
System.out.println("执行step 2");
return "step2 result";
});
ListenableFuture<List<String>> future1And2 = Futures.allAsList(future1, future2);
Futures.addCallback(future1And2, new FutureCallback<List<String>>() {
@Override
public void onSuccess(List<String> result) {
System.out.println(result);
ListenableFuture<String> future3 = guavaExecutor.submit(() -> {
System.out.println("执行step 3");
return "step3 result";
});
Futures.addCallback(future3, new FutureCallback<String>() {
@Override
public void onSuccess(String result) {
System.out.println(result);
}
@Override
public void onFailure(Throwable t) {
}
}, guavaExecutor);
}
@Override
public void onFailure(Throwable t) {
}}, guavaExecutor);
ExecutorService executor = Executors.newFixedThreadPool(5);
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
System.out.println("执行step 1");
return "step1 result";
}, executor);
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
System.out.println("执行step 2");
return "step2 result";
});
cf1.thenCombine(cf2, (result1, result2) -> {
System.out.println(result1 + " , " + result2);
System.out.println("执行step 3");
return "step3 result";
}).thenAccept(result3 -> System.out.println(result3));
ExecutorService executor = Executors.newFixedThreadPool(5);
//1、使用runAsync或supplyAsync发起异步调用
CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
return "result1";
}, executor);
//2、CompletableFuture.completedFuture()直接创建一个已完成状态的CompletableFuture
CompletableFuture<String> cf2 = CompletableFuture.completedFuture("result2");
//3、先初始化一个未完成的CompletableFuture,然后通过complete()、completeExceptionally(),完成该CompletableFuture
CompletableFuture<String> cf = new CompletableFuture<>();
cf.complete("success");
@FunctionalInterface
public interface ThriftAsyncCall {
void invoke() throws TException;
}
/**
* 该方法为美团内部rpc注册监听的封装,可以作为其他实现的参照
* OctoThriftCallback 为thrift回调方法
* ThriftAsyncCall 为自定义函数,用来表示一次thrift调用(定义如上)
*/
public static <T> CompletableFuture<T> toCompletableFuture(final OctoThriftCallback<?,T> callback , ThriftAsyncCall thriftCall) {
//新建一个未完成的CompletableFuture
CompletableFuture<T> resultFuture = new CompletableFuture<>();
//监听回调的完成,并且与CompletableFuture同步状态
callback.addObserver(new OctoObserver<T>() {
@Override
public void onSuccess(T t) {
resultFuture.complete(t);
}
@Override
public void onFailure(Throwable throwable) {
resultFuture.completeExceptionally(throwable);
}
});
if (thriftCall != null) {
try {
thriftCall.invoke();
} catch (TException e) {
resultFuture.completeExceptionally(e);
}
}
return resultFuture;
}
CompletableFuture<String> cf3 = cf1.thenApply(result1 -> {
//result1为CF1的结果
//......
return "result3";
});
CompletableFuture<String> cf5 = cf2.thenApply(result2 -> {
//result2为CF2的结果
//......
return "result5";
});
CompletableFuture<String> cf4 = cf1.thenCombine(cf2, (result1, result2) -> {
//result1和result2分别为cf1和cf2的结果
return "result4";
});
allOf
或anyOf
方法来实现,区别是当需要多个依赖全部完成时使用allOf
,当多个依赖中的任意一个完成即可时使用anyOf
,如下代码所示:CompletableFuture<Void> cf6 = CompletableFuture.allOf(cf3, cf4, cf5);
CompletableFuture<String> result = cf6.thenApply(v -> {
//这里的join并不会阻塞,因为传给thenApply的函数是在CF3、CF4、CF5全部完成时,才会执行 。
result3 = cf3.join();
result4 = cf4.join();
result5 = cf5.join();
//根据result3、result4、result5组装最终result;
return "result";
});
allOf
、anyOf
,区别在于allOf
观察者实现类为BiRelay,需要所有被依赖的CF完成后才会执行回调;而anyOf
观察者实现类为OrRelay,任意一个被依赖的CF完成后就会触发。二者的实现方式都是将多个被依赖的CF构建成一棵平衡二叉树,执行结果层层通知,直到根节点,触发回调监听。ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
System.out.println("supplyAsync 执行线程:" + Thread.currentThread().getName());
//业务操作
return "";
}, threadPool1);
//此时,如果future1中的业务操作已经执行完毕并返回,则该thenApply直接由当前main线程执行;否则,将会由执行以上业务操作的threadPool1中的线程执行。
future1.thenApply(value -> {
System.out.println("thenApply 执行线程:" + Thread.currentThread().getName());
return value + "1";
});
//使用ForkJoinPool中的共用线程池CommonPool
future1.thenApplyAsync(value -> {
//do something
return value + "1";
});
//使用指定线程池
future1.thenApplyAsync(value -> {
//do something
return value + "1";
}, threadPool1);
public Object doGet() {
ExecutorService threadPool1 = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100));
CompletableFuture cf1 = CompletableFuture.supplyAsync(() -> {
//do sth
return CompletableFuture.supplyAsync(() -> {
System.out.println("child");
return "child";
}, threadPool1).join();//子任务
}, threadPool1);
return cf1.join();
}
@Autowired
private WmOrderAdditionInfoThriftService wmOrderAdditionInfoThriftService;//内部接口
public CompletableFuture<Integer> getCancelTypeAsync(long orderId) {
CompletableFuture<WmOrderOpRemarkResult> remarkResultFuture = wmOrderAdditionInfoThriftService.findOrderCancelledRemarkByOrderIdAsync(orderId);//业务方法,内部会发起异步rpc调用
return remarkResultFuture
.exceptionally(err -> {//通过exceptionally 捕获异常,打印日志并返回默认值
log.error("WmOrderRemarkService.getCancelTypeAsync Exception orderId={}", orderId, err);
return 0;
});
}
@Autowired
private WmOrderAdditionInfoThriftService wmOrderAdditionInfoThriftService;//内部接口
public CompletableFuture<Integer> getCancelTypeAsync(long orderId) {
CompletableFuture<WmOrderOpRemarkResult> remarkResultFuture = wmOrderAdditionInfoThriftService.findOrderCancelledRemarkByOrderIdAsync(orderId);//业务方法,内部会发起异步rpc调用
return remarkResultFuture
.thenApply(result -> {//这里增加了一个回调方法thenApply,如果发生异常thenApply内部会通过new CompletionException(throwable) 对异常进行包装
//这里是一些业务操作
})
.exceptionally(err -> {//通过exceptionally 捕获异常,这里的err已经被thenApply包装过,因此需要通过Throwable.getCause()提取异常
log.error("WmOrderRemarkService.getCancelTypeAsync Exception orderId={}", orderId, ExceptionUtils.extractRealException(err));
return 0;
});
}
public class ExceptionUtils {
public static Throwable extractRealException(Throwable throwable) {
//这里判断异常类型是否为CompletionException、ExecutionException,如果是则进行提取,否则直接返回。
if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
if (throwable.getCause() != null) {
return throwable.getCause();
}
}
return throwable;
}
}
@FunctionalInterface
public interface ThriftAsyncCall {
void invoke() throws TException ;
}
/**
* CompletableFuture封装工具类
*/
@Slf4j
public class FutureUtils {
/**
* 该方法为美团内部rpc注册监听的封装,可以作为其他实现的参照
* OctoThriftCallback 为thrift回调方法
* ThriftAsyncCall 为自定义函数,用来表示一次thrift调用(定义如上)
*/
public static <T> CompletableFuture<T> toCompletableFuture(final OctoThriftCallback<?,T> callback , ThriftAsyncCall thriftCall) {
CompletableFuture<T> thriftResultFuture = new CompletableFuture<>();
callback.addObserver(new OctoObserver<T>() {
@Override
public void onSuccess(T t) {
thriftResultFuture.complete(t);
}
@Override
public void onFailure(Throwable throwable) {
thriftResultFuture.completeExceptionally(throwable);
}
});
if (thriftCall != null) {
try {
thriftCall.invoke();
} catch (TException e) {
thriftResultFuture.completeExceptionally(e);
}
}
return thriftResultFuture;
}
/**
* 设置CF状态为失败
*/
public static <T> CompletableFuture<T> failed(Throwable ex) {
CompletableFuture<T> completableFuture = new CompletableFuture<>();
completableFuture.completeExceptionally(ex);
return completableFuture;
}
/**
* 设置CF状态为成功
*/
public static <T> CompletableFuture<T> success(T result) {
CompletableFuture<T> completableFuture = new CompletableFuture<>();
completableFuture.complete(result);
return completableFuture;
}
/**
* 将List<CompletableFuture<T>> 转为 CompletableFuture<List<T>>
*/
public static <T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<T>> completableFutures) {
return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream()
.map(CompletableFuture::join)
.collect(Collectors.toList())
);
}
/**
* 将List<CompletableFuture<List<T>>> 转为 CompletableFuture<List<T>>
* 多用于分页查询的场景
*/
public static <T> CompletableFuture<List<T>> sequenceList(Collection<CompletableFuture<List<T>>> completableFutures) {
return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream()
.flatMap( listFuture -> listFuture.join().stream())
.collect(Collectors.toList())
);
}
/*
* 将List<CompletableFuture<Map<K, V>>> 转为 CompletableFuture<Map<K, V>>
* @Param mergeFunction 自定义key冲突时的merge策略
*/
public static <K, V> CompletableFuture<Map<K, V>> sequenceMap(
Collection<CompletableFuture<Map<K, V>>> completableFutures, BinaryOperator<V> mergeFunction) {
return CompletableFuture
.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream().map(CompletableFuture::join)
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.toMap(Entry::getKey, Entry::getValue, mergeFunction)));
}
/**
* 将List<CompletableFuture<T>> 转为 CompletableFuture<List<T>>,并过滤调null值
*/
public static <T> CompletableFuture<List<T>> sequenceNonNull(Collection<CompletableFuture<T>> completableFutures) {
return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream()
.map(CompletableFuture::join)
.filter(e -> e != null)
.collect(Collectors.toList())
);
}
/**
* 将List<CompletableFuture<List<T>>> 转为 CompletableFuture<List<T>>,并过滤调null值
* 多用于分页查询的场景
*/
public static <T> CompletableFuture<List<T>> sequenceListNonNull(Collection<CompletableFuture<List<T>>> completableFutures) {
return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream()
.flatMap( listFuture -> listFuture.join().stream().filter(e -> e != null))
.collect(Collectors.toList())
);
}
/**
* 将List<CompletableFuture<Map<K, V>>> 转为 CompletableFuture<Map<K, V>>
* @Param filterFunction 自定义过滤策略
*/
public static <T> CompletableFuture<List<T>> sequence(Collection<CompletableFuture<T>> completableFutures,
Predicate<? super T> filterFunction) {
return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream()
.map(CompletableFuture::join)
.filter(filterFunction)
.collect(Collectors.toList())
);
}
/**
* 将List<CompletableFuture<List<T>>> 转为 CompletableFuture<List<T>>
* @Param filterFunction 自定义过滤策略
*/
public static <T> CompletableFuture<List<T>> sequenceList(Collection<CompletableFuture<List<T>>> completableFutures,
Predicate<? super T> filterFunction) {
return CompletableFuture.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream()
.flatMap( listFuture -> listFuture.join().stream().filter(filterFunction))
.collect(Collectors.toList())
);
}
/**
* 将CompletableFuture<Map<K,V>>的list转为 CompletableFuture<Map<K,V>>。 多个map合并为一个map。 如果key冲突,采用新的value覆盖。
*/
public static <K, V> CompletableFuture<Map<K, V>> sequenceMap(
Collection<CompletableFuture<Map<K, V>>> completableFutures) {
return CompletableFuture
.allOf(completableFutures.toArray(new CompletableFuture<?>[0]))
.thenApply(v -> completableFutures.stream().map(CompletableFuture::join)
.flatMap(map -> map.entrySet().stream())
.collect(Collectors.toMap(Entry::getKey, Entry::getValue, (a, b) -> b)));
}}
public class ExceptionUtils {
/**
* 提取真正的异常
*/
public static Throwable extractRealException(Throwable throwable) {
if (throwable instanceof CompletionException || throwable instanceof ExecutionException) {
if (throwable.getCause() != null) {
return throwable.getCause();
}
}
return throwable;
}
}
@Slf4j
public abstract class AbstractLogAction<R> {
protected final String methodName;
protected final Object[] args;
public AbstractLogAction(String methodName, Object... args) {
this.methodName = methodName;
this.args = args;
}
protected void logResult(R result, Throwable throwable) {
if (throwable != null) {
boolean isBusinessError = throwable instanceof TBase || (throwable.getCause() != null && throwable
.getCause() instanceof TBase);
if (isBusinessError) {
logBusinessError(throwable);
} else if (throwable instanceof DegradeException || throwable instanceof DegradeRuntimeException) {//这里为内部rpc框架抛出的异常,使用时可以酌情修改
if (RhinoSwitch.getBoolean("isPrintDegradeLog", false)) {
log.error("{} degrade exception, param:{} , error:{}", methodName, args, throwable);
}
} else {
log.error("{} unknown error, param:{} , error:{}", methodName, args, ExceptionUtils.extractRealException(throwable));
}
} else {
if (isLogResult()) {
log.info("{} param:{} , result:{}", methodName, args, result);
} else {
log.info("{} param:{}", methodName, args);
}
}
}
private void logBusinessError(Throwable throwable) {
log.error("{} business error, param:{} , error:{}", methodName, args, throwable.toString(), ExceptionUtils.extractRealException(throwable));
}
private boolean isLogResult() {
//这里是动态配置开关,用于动态控制日志打印,开源动态配置中心可以使用nacos、apollo等,如果项目没有使用配置中心则可以删除
return RhinoSwitch.getBoolean(methodName + "_isLogResult", false);
}}
/**
* 发生异常时,根据是否为业务异常打印日志。
* 跟CompletableFuture.whenComplete配合使用,不改变completableFuture的结果(正常OR异常)
*/
@Slf4j
public class LogErrorAction<R> extends AbstractLogAction<R> implements BiConsumer<R, Throwable> {
public LogErrorAction(String methodName, Object... args) {
super(methodName, args);
}
@Override
public void accept(R result, Throwable throwable) {
logResult(result, throwable);
}
}
completableFuture
.whenComplete(
new LogErrorAction<>("orderService.getOrder", params));
/**
* 当发生异常时返回自定义的值
*/
public class DefaultValueHandle<R> extends AbstractLogAction<R> implements BiFunction<R, Throwable, R> {
private final R defaultValue;
/**
* 当返回值为空的时候是否替换为默认值
*/
private final boolean isNullToDefault;
/**
* @param methodName 方法名称
* @param defaultValue 当异常发生时自定义返回的默认值
* @param args 方法入参
*/
public DefaultValueHandle(String methodName, R defaultValue, Object... args) {
super(methodName, args);
this.defaultValue = defaultValue;
this.isNullToDefault = false;
}
/**
* @param isNullToDefault
* @param defaultValue 当异常发生时自定义返回的默认值
* @param methodName 方法名称
* @param args 方法入参
*/
public DefaultValueHandle(boolean isNullToDefault, R defaultValue, String methodName, Object... args) {
super(methodName, args);
this.defaultValue = defaultValue;
this.isNullToDefault = isNullToDefault;
}
@Override
public R apply(R result, Throwable throwable) {
logResult(result, throwable);
if (throwable != null) {
return defaultValue;
}
if (result == null && isNullToDefault) {
return defaultValue;
}
return result;
}
public static <R> DefaultValueHandle.DefaultValueHandleBuilder<R> builder() {
return new DefaultValueHandle.DefaultValueHandleBuilder<>();
}
public static class DefaultValueHandleBuilder<R> {
private boolean isNullToDefault;
private R defaultValue;
private String methodName;
private Object[] args;
DefaultValueHandleBuilder() {
}
public DefaultValueHandle.DefaultValueHandleBuilder<R> isNullToDefault(final boolean isNullToDefault) {
this.isNullToDefault = isNullToDefault;
return this;
}
public DefaultValueHandle.DefaultValueHandleBuilder<R> defaultValue(final R defaultValue) {
this.defaultValue = defaultValue;
return this;
}
public DefaultValueHandle.DefaultValueHandleBuilder<R> methodName(final String methodName) {
this.methodName = methodName;
return this;
}
public DefaultValueHandle.DefaultValueHandleBuilder<R> args(final Object... args) {
this.args = args;
return this;
}
public DefaultValueHandle<R> build() {
return new DefaultValueHandle<R>(this.isNullToDefault, this.defaultValue, this.methodName, this.args);
}
public String toString() {
return "DefaultValueHandle.DefaultValueHandleBuilder(isNullToDefault=" + this.isNullToDefault + ", defaultValue=" + this.defaultValue + ", methodName=" + this.methodName + ", args=" + Arrays.deepToString(this.args) + ")";
}
}
completableFuture.handle(new DefaultValueHandle<>("orderService.getOrder", Collections.emptyMap(), params));
美团外卖商家系统,既有日千万量级订单下的稳定性挑战,又具有B端特有的业务复杂性,同时也在商家生态、商家运营、智能硬件等方向创新与探索。通过在高可用、领域驱动设计、微服务等技术方向持续实践,积累了丰富的技术经验。欢迎加入美团外卖商家组技术团队,感兴趣的同学可以将简历发送至:pingxumeng@meituan.com。
阅读更多