netty(九)初识Netty-Future & Promise
2021-11-10 本文已影响0人
我犟不过你
一、简介
在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)