文章目录
- 前言
- Future
- 主要方法
- Promise
- 主要方法
- Future 和 Promise 的区别
- 使用场景
- 代码示例
- Jdk中的Future
- Netty中的Future
- Netty中的Promise
- 同步处理任务失败 - await
- 异步处理任务失败
前言
在 Netty 中,Future 和 Promise 是用于处理异步操作结果的重要组件。
首先要说明 netty 中的 Future 与 jdk 中的 Future 同名,但是是两个接口,netty 的 Future 继承自 jdk 的 Future,而 Promise 又对 netty Future 进行了扩展
- jdk Future 只能同步等待任务结束(或成功、或失败)才能得到结果
- netty Future 可以同步等待任务结束得到结果,也可以异步方式得到结果,但都是要等任务结束
- netty Promise 不仅有 netty Future 的功能,而且脱离了任务独立存在,只作为两个线程间传递结果的容器
Future
Future 是一个表示异步操作结果的对象。它提供了一种机制来查询异步操作是否已经完成,并且可以获取操作的结果或者抛出的异常。Future 接口在 Java 标准库中也有实现(java.util.concurrent.Future),但 Netty 提供了自己的 Future 实现,更加符合其异步事件驱动的设计理念。
主要方法
boolean isDone():判断异步操作是否已经完成。
boolean isSuccess():判断异步操作是否成功完成。
boolean isCancelled():判断异步操作是否已被取消。
V get():阻塞等待异步操作完成,并返回结果或抛出异常。
V get(long timeout, TimeUnit unit):在指定时间内阻塞等待异步操作完成,并返回结果或抛出异常。
ChannelFuture future = channel.writeAndFlush(msg);
future.addListener((ChannelFutureListener) future1 -> {if (future1.isSuccess()) {System.out.println("Write and flush completed successfully.");} else {System.out.println("Write and flush failed.");future1.cause().printStackTrace();}
});
Promise
Promise 是 Netty 特有的一个接口,它扩展了 Future 接口,并提供了额外的方法来控制异步操作的状态。Promise 通常由 EventLoop 使用来完成异步操作,并通知 Future 的监听者。
主要方法
Promise< V >:这是一个泛型接口,表示异步操作的结果类型。
void setSuccess(V result):设置异步操作成功完成,并传入结果。
void setFailure(Throwable cause):设置异步操作失败,并传入异常。
void setUncancellable():设置异步操作不可取消。
void cancel(boolean mayInterruptIfRunning):取消异步操作。
ChannelFuture addListener(GenericFutureListener<? extends Future<? super V>> listener):添加一个监听器,当异步操作完成时调用。
ChannelPromise promise = channel.newPromise();
channel.writeAndFlush(msg, promise).addListener((ChannelFutureListener) future -> {if (future.isSuccess()) {System.out.println("Write and flush completed successfully.");} else {System.out.println("Write and flush failed.");future.cause().printStackTrace();}
});
Future 和 Promise 的区别
- Future 是一个只读的接口,它只能用来查询异步操作的状态和结果。
- Promise 是一个可写的接口,除了可以查询状态和结果外,还可以设置操作的结果或异常。
使用场景
- Future 通常用于消费异步操作的结果,例如在完成某个异步操作后执行某些逻辑。
- Promise 通常用于生产异步操作的结果,例如在某个异步操作完成后,通过 Promise 设置结果或异常。
代码示例
Jdk中的Future
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;@Slf4j
public class Test02JdkFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 创建线程池ExecutorService service = Executors.newFixedThreadPool(2);// 2. 提交任务Future<Integer> future = service.submit(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {log.debug("执行计算");//[DEBUG] [pool-1-thread-1] c.i.n.c.Test02JdkFuture - 执行计算Thread.sleep(1000);return 50;}});// 3. 主线程通过 future 来获取结果log.debug("等待结果");//[DEBUG] [main] c.i.n.c.Test02JdkFuture - 等待结果log.debug("结果是 {}", future.get());//[DEBUG] [main] c.i.n.c.Test02JdkFuture - 结果是 50}
}
Netty中的Future
import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;@Slf4j
public class Test03NettyFuture {public static void main(String[] args) throws ExecutionException, InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();EventLoop eventLoop = group.next();Future<Integer> future = eventLoop.submit(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {log.debug("执行计算");//[DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.Test03NettyFuture - 执行计算Thread.sleep(1000);return 70;}});//使用同步的方式(主线程中拿结果)
// log.debug("等待结果");//[DEBUG] [main] c.i.n.c.Test03NettyFuture - 等待结果
// log.debug("结果是 {}", future.get());//[DEBUG] [main] c.i.n.c.Test03NettyFuture - 结果是 70//使用异步的方式(执行线程中拿结果)future.addListener(new GenericFutureListener<Future<? super Integer>>(){@Overridepublic void operationComplete(Future<? super Integer> future) throws Exception {log.debug("接收结果:{}", future.getNow());//[DEBUG] [nioEventLoopGroup-2-1] c.i.n.c.Test03NettyFuture - 接收结果:70}});}
}
Netty中的Promise
import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ExecutionException;@Slf4j
public class Test04NettyPromise {public static void main(String[] args) throws ExecutionException, InterruptedException {// 1. 准备 EventLoop 对象EventLoop eventLoop = new NioEventLoopGroup().next();// 2. 可以主动创建 promise, promise是一个结果容器DefaultPromise<Integer> promise = new DefaultPromise<>(eventLoop);new Thread(() -> {// 3. 任意一个线程执行计算,计算完毕后向 promise 填充结果log.debug("开始计算...");//[DEBUG] [Thread-0] c.i.n.c.Test04NettyPromise - 开始计算...try {int i = 1 / 0;Thread.sleep(1000);promise.setSuccess(80);} catch (Exception e) {e.printStackTrace();promise.setFailure(e);//传递异常}}).start();// 4. 接收结果的线程log.debug("等待结果...");//[DEBUG] [main] c.i.n.c.Test04NettyPromise - 等待结果...//有异常则直接发出异常log.debug("结果是: {}", promise.get());//[DEBUG] [main] c.i.n.c.Test04NettyPromise - 结果是: 80}
}
同步处理任务失败 - await
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);eventExecutors.execute(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}RuntimeException e = new RuntimeException("error...");log.debug("set failure, {}", e.toString());promise.setFailure(e);
});log.debug("start...");
log.debug("{}", promise.getNow());
promise.await(); // 与 sync 和 get 区别在于,不会抛异常
log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
输出
12:18:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
12:18:53 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - null
12:18:54 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...
12:18:54 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - result java.lang.RuntimeException: error...
异步处理任务失败
DefaultEventLoop eventExecutors = new DefaultEventLoop();
DefaultPromise<Integer> promise = new DefaultPromise<>(eventExecutors);promise.addListener(future -> {log.debug("result {}", (promise.isSuccess() ? promise.getNow() : promise.cause()).toString());
});eventExecutors.execute(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}RuntimeException e = new RuntimeException("error...");log.debug("set failure, {}", e.toString());promise.setFailure(e);
});log.debug("start...");
输出
12:04:57 [DEBUG] [main] c.i.o.DefaultPromiseTest2 - start...
12:04:58 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - set failure, java.lang.RuntimeException: error...
12:04:58 [DEBUG] [defaultEventLoop-1-1] c.i.o.DefaultPromiseTest2 - result java.lang.RuntimeException: error...