vertx的Future设计
2024-03-19 19:12:05
异步痛点
1.回调地狱(CallBack hell) ; 解决方式 Promise 或 Future
2.执行异步后的结果如何回调currentThread ; 解决方式 Context 设计
3.如何处理依赖多异步的result进行逻辑 ; 解决方案 CompositeFuture
JDK的lib库 Callable和Future 问题
Callable任务可以返回结果，返回的结果可以由Future对象取出,但是调用Future.get()会阻塞当前线程
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L);
return report(s);
}
public V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {
if (unit == null)
throw new NullPointerException();
int s = state;
if (s <= COMPLETING &&
(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
throw new TimeoutException();
return report(s);
}
/**
* 等待完成，或者中断、超时
*
* @param timed 定义了超时则为true
* @param nanos 等待时间
*/
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
//轮询查看FutureTask的状态
for (;;) {
if (Thread.interrupted()) {//线程中断
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {//完成或取消状态
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING)//正在set result,Thread “运行状态”进入到“就绪状态”
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
else
LockSupport.park(this);
}
}
vertx的Future设计
1. Future通过事件注册(EventHandler)回调方式,来解决Thread阻塞问题.
UML图:
Futrue instances 采用FutureFactory生产，基于 SPI 机制:
public class FutureFactoryImpl implements FutureFactory {
private static final SucceededFuture EMPTY = new SucceededFuture<>(null);
/**
* 创建FutureImpl 实例
*/
public <T> Future<T> future() {
return new FutureImpl<>();
}
/**
* 返回一个常量 SucceededFuture 实例
* result本身为null, 可以减少内存开销
*/
public <T> Future<T> succeededFuture() {
@SuppressWarnings("unchecked")
Future<T> fut = EMPTY;
return fut;
}
/**
* 创建一个 SucceededFuture 实例
*/
public <T> Future<T> succeededFuture(T result) {
return new SucceededFuture<>(result);
}
/**
* 创建一个 FailedFuture 实例
*/
public <T> Future<T> failedFuture(Throwable t) {
return new FailedFuture<>(t);
}
/**
* 创建一个 FailedFuture 实例
*/
public <T> Future<T> failureFuture(String failureMessage) {
return new FailedFuture<>(failureMessage);
}
}
多异步的result如何组合(并行变串行)
使用的CompositeFuture，来处理多异步结果组合问题:采用计数器(Counters)的方法来解决 wait 问题
public class CompositeFutureImpl implements CompositeFuture, Handler<AsyncResult<CompositeFuture>> {
private final Future[] results;//定义数组
private int count;//计数器
private boolean completed;//是否完成
private Throwable cause;//错误原因
private Handler<AsyncResult<CompositeFuture>> handler;// 回调 eventHandler
public static CompositeFuture all(Future<?>... results) {
CompositeFutureImpl composite = new CompositeFutureImpl(results);//创建实例
int len = results.length; //获取 futures数组长度
for (int i = ; i < len; i++) {
results[i].setHandler(ar -> {
Handler<AsyncResult<CompositeFuture>> handler = null;
if (ar.succeeded()) {
synchronized (composite) { //添加内存屏障，防止并发问题
composite.count++;
if (!composite.isComplete() && composite.count == len) {//所有future成功
handler = composite.setCompleted(null);
}
}
} else {
synchronized (composite) {//添加内存屏障，防止并发问题
if (!composite.isComplete()) {//任何一个失败就失败
handler = composite.setCompleted(ar.cause());
}
}
}
if (handler != null) {//执行回调EventHandler
handler.handle(composite);
}
});
}
if (len == ) {//判断临界点
composite.setCompleted(null);
}
return composite;
}
}
public static CompositeFuture any(Future<?>... results) {
CompositeFutureImpl composite = new CompositeFutureImpl(results);
int len = results.length;
for (int i = ;i < len;i++) {
results[i].setHandler(ar -> {
Handler<AsyncResult<CompositeFuture>> handler = null;
if (ar.succeeded()) {
synchronized (composite) {
if (!composite.isComplete()) {//任何一个成功
handler = composite.setCompleted(null);
}
}
} else {
synchronized (composite) {
composite.count++;
if (!composite.isComplete() && composite.count == len) {//所有future失败
handler = composite.setCompleted(ar.cause());
}
}
}
if (handler != null) {//执行回调EventHandler
handler.handle(composite);
}
});
}
if (results.length == ) {//判断临界点
composite.setCompleted(null);
}
return composite;
}
private static CompositeFuture join(Function<CompositeFuture, Throwable> pred, Future<?>... results) {
CompositeFutureImpl composite = new CompositeFutureImpl(results);
int len = results.length;
for (int i = ; i < len; i++) {
results[i].setHandler(ar -> {
Handler<AsyncResult<CompositeFuture>> handler = null;
synchronized (composite) {
composite.count++;
if (!composite.isComplete() && composite.count == len) {//处理所有不管失败还是成功
// Take decision here
Throwable failure = pred.apply(composite);
handler = composite.setCompleted(failure);
}
}
if (handler != null) {
handler.handle(composite);
}
});
}
if (len == ) {//{//判断临界点
composite.setCompleted(null);
}
return composite;
}
/**
* 根据下标返回结果
* /
public <T> T resultAt(int index) {
return this.<T>future(index).result();
}
public interface CompositeFuture extends Future<CompositeFuture> {
/**
* 返回list Future.result()
*/
default <T> List<T> list() {
int size = size();
ArrayList<T> list = new ArrayList<>(size);
for (int index = ;index < size;index++) {
list.add(resultAt(index));
}
return list;
}
}
最新文章
- SQL联合查询（内联、左联、右联、全联）的语法（转）
- windows下使用TortoiseGit代替Git命令行操作
- 关于textarea标签在谷歌跟火狐可以拖动大小
- cygwin中vim的使用方法
- Max Sum of Max-K-sub-sequence hdu3415
- 学习如何看懂SQL Server执行计划（二）——函数计算篇
- 基于.net的通用内存缓存模型组件
- java8-新特性--（接口的默认方法与静态方法）
- 【你不一定知晓的】C#取消异步操作
- Android--Android Studio 打开ADM报错
- jQuery上传文件
- linux下文件描述符的查看及分析
- [Vim] Vim 常用基本操作
- ViewPage + RadioGroup + Fragment学习
- Linux Shell之表达式
- PHP打印重复的东西
- 日请求亿级的QQ会员AMS平台PHP7升级实践
- 笔记：javascript操作iframe内的DOM元素，及调用iframe内的方法
- tensorflow中使用tf.variable_scope和tf.get_variable的ValueError
- centos7.3安装python3.6.5