一、简介
在netty当中，我们需要进行异步处理的时候，经常会调用以下的两个方法：
Future & Promise
其实在我们使用JDK的时候，就知道有一个Future接口，用于异步时接收任务结果。
在Netty当中，基于JDK当中的Future接口，进行了扩展；后面又在Netty的Future基础之上，增加了Promise接口。
关于三者的关系请看以下的类图：
image.png
- JDK Future：只能同步等待任务结束（无论成功还是失败）才能得到结果。
- Netty Future：可以同步等待任务结束得到结果，也可以异步方式得到结果，前提是任务必须要结束。
- Netty Promise：不仅有 netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器。
二、扩展了哪些主要能力？
在这一章节，主要分析前面提到的netty新增的两个接口都新增了哪些功能。
| 功能/名称 | jdk Future | netty Future | Promise |
|---|---|---|---|
| cancel | 取消任务 | - | - |
| isCanceled | 任务是否取消 | - | - |
| isDone | 任务是否完成，不能区分成功失败 | - | - |
| get | 获取任务结果，阻塞等待 | - | - |
| getNow | - | 获取任务结果，非阻塞，还未产生结果时返回 null | - |
| await | - | 等待任务结束，如果任务失败，不会抛异常，而是通过 isSuccess 判断 | - |
| sync | - | 等待任务结束，如果任务失败，抛出异常 | - |
| isSuccess | - | 判断任务是否成功 | - |
| cause | - | 获取失败信息，非阻塞，如果没有失败，返回null | - |
| addLinstener | - | 添加回调，异步接收结果 | - |
| setSuccess | - | - | 设置成功结果 |
| setFailure | - | - | 设置失败结果 |
三、使用示例
下面针对Promise我们重点学习，看看针对不同场景下的使用。
例1 同步处理任务成功
public static void main(String[] args) throws ExecutionException, InterruptedException {
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() +", set success: " + 10);
promise.setSuccess(10);
});
System.out.println(Thread.currentThread().getName() +", start...");
//未产生结果，不阻塞，返回null
System.out.println(Thread.currentThread().getName() +", " + promise.getNow());
//阻塞等待结果
System.out.println(Thread.currentThread().getName() +", " + promise.get());
}
结果：
main, start...
main, null
defaultEventLoop-1-1, set success: 10
main, 10
例2 异步处理任务成功
public static void main(String[] args) throws ExecutionException, InterruptedException {
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() +", set success: " + 10);
promise.setSuccess(10);
});
System.out.println(Thread.currentThread().getName() +", start...");
//添加异步监听，当有结果时调用getNow获取结果
promise.addListener(future->{
System.out.println(Thread.currentThread().getName() +", " + future.getNow());
});
}
结果：
main, start...
defaultEventLoop-1-1, set success: 10
defaultEventLoop-1-1, 10
例3 同步处理任务失败 sync & get
sync或者get，区别是get会对异常信息再包一层
public static void main(String[] args) throws ExecutionException, InterruptedException {
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() +", set failure");
promise.setFailure(new RuntimeException());
});
System.out.println(Thread.currentThread().getName() +", start...");
//未产生结果，不阻塞，返回null
System.out.println(Thread.currentThread().getName() +", " + promise.getNow());
//阻塞等待结果
System.out.println(Thread.currentThread().getName() +", " + promise.get());
//System.out.println(Thread.currentThread().getName() +", " + promise.get());
}
结果：
main, start...
main, null
defaultEventLoop-1-1, set failure
Exception in thread "main" java.util.concurrent.ExecutionException: java.lang.RuntimeException
at io.netty.util.concurrent.DefaultPromise.get(DefaultPromise.java:349)
at com.cloud.bssp.netty.promise.Test3.main(Test3.java:34)
Caused by: java.lang.RuntimeException
at com.cloud.bssp.netty.promise.Test3.lambda$main$0(Test3.java:27)
at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
例4 同步处理任务失败 await
public static void main(String[] args) throws ExecutionException, InterruptedException {
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(()->{
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() +", set failure");
promise.setFailure(new RuntimeException());
});
System.out.println(Thread.currentThread().getName() +", start...");
//未产生结果，不阻塞，返回null
System.out.println(Thread.currentThread().getName() +", " + promise.getNow());
//阻塞等待结果
promise.await();
//isSuccess判断任务是否成功
System.out.println(Thread.currentThread().getName() +", " + promise.isSuccess());
}
结果：
main, start...
main, null
defaultEventLoop-1-1, set failure
main, false
例5 异步处理任务失败
public static void main(String[] args) {
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.execute(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + ", set failure");
promise.setFailure(new RuntimeException());
});
System.out.println(Thread.currentThread().getName() + ", start...");
//添加异步监听，当有结果时调用getNow获取结果
promise.addListener(future -> {
if (!promise.isSuccess()) {
//失败查看结果
System.out.println(Thread.currentThread().getName() + ", " + promise.cause());
}
System.out.println(Thread.currentThread().getName() + ", " + future.getNow());
});
}
结果：
main, start...
defaultEventLoop-1-1, set failure
defaultEventLoop-1-1, java.lang.RuntimeException
defaultEventLoop-1-1, null
例6 await 死锁检查
public static void main(String[] args) {
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);
eventExecutors.submit(()->{
System.out.println("1");
try {
promise.await();
// 注意不能仅捕获 InterruptedException 异常
// 否则 死锁检查抛出的 BlockingOperationException 会继续向上传播
// 而提交的任务会被包装为 PromiseTask，它的 run 方法中会 catch 所有异常然后设置为 Promise 的失败结果而不会抛出
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("2");
});
eventExecutors.submit(()->{
System.out.println("3");
try {
promise.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("4");
});
}
结果：
1
2
3
4
io.netty.util.concurrent.BlockingOperationException: DefaultPromise@6a77ffc2(incomplete)
at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:461)
at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:246)
at com.cloud.bssp.netty.promise.Test6.lambda$main$0(Test6.java:21)
at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:106)
at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
io.netty.util.concurrent.BlockingOperationException: DefaultPromise@6a77ffc2(incomplete)
at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:461)
at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:246)
at com.cloud.bssp.netty.promise.Test6.lambda$main$1(Test6.java:33)
at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:106)
at io.netty.channel.DefaultEventLoop.run(DefaultEventLoop.java:54)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
网友评论