一、什么是异步
异步编程一般涉及到两个线程：任务提交线程，任务执行线程。任务提交之后 任务提交线程不需要等待结果直接返回，由任务执行线程执行完任务之后回调任务执行线程
异步的好处：对于耗时比较大的操作，比如I/O, 网络的连接，可以不需要同步阻塞等待操作的完成，可以继续往下执行，只需要注册一个回调的函数。
异步编程的两个难点：
- 如何判断任务执行完成
- 如何拿到返回值
二、JDK中提供的异步模型
JDK中使用Future模式来实现异步编程
demo：
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2);
System.out.println("start");
Future<Integer> submit = executorService.submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("call 任务执行");
Thread.sleep(500);
return 99;
}
});
Integer value = null;
try {
value = submit.get();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(value);
System.out.println("end");
executorService.shutdown();
}
分析程序的运行
- 提交了一个带返回值的任务，并且返回了一个Future对象，直接就返回了
- 主函数调用了get方法，get方法是同步阻塞等待任务运行的结果
接下来通过源码看Future如何解决两个难点（如何判断任务执行完成、如何拿到返回值）
newFixedThreadPool返回的是ThreadPoolExecutor类型的线程池，继承自AbstractExecutorService，之后程序中调用的submit其实是复用的父类方法，也就是AbstractExecutorService中的sumit方法
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
在sumit中，把传入的Callable对象又包装成FutureTask对象并且返回这个对象，并且真正通过execute进行执行。这个execute是使用的ThreadPoolExecutor自己实现的execute方法。最终线程执行的结果或者是异常会被放入FutureTask中的outcome字段，主程序中通过FutureTask.get() 方法获得任务运行的结果。
记住是主线程调用get方法，get方法中有一个循环判断去判断值是否已经完成，当结果还没有的时候，主线程就会处于循环中而阻塞之后的代码。
Future 提供的功能是：用户线程需要主动轮询 Future 线程是否完成当前任务，如果不通过轮询是否完成而是同步等待获取则会阻塞直到执行完毕为止。所以从这里看，Future并不是真正的异步，因为它少了一个 回调 ，充其量只能算是一个同步非阻塞模式。在Java8 中新增了CompletableFuture， 实现了回调这个功能。
我的理解：Future是为了简化异步编程。当需要进行异步调用时，同步返回一个Future对象，任务执行者会向这个对象中设置结果，任务提交者需要从对象中拿到值，所以就是一个关于线程间通信的问题，而Future就是帮我们封装好了相关的方法。
三、netty的Future和Promise机制
- netty的Future继承自jdk的Future。最重要的是扩展了addListener方法，实现了异步这个功能
- Promise继承自Future，并且增加了主动设置结果的两个方法
Future
netty服务端 demo
NioEventLoopGroup boss = new NioEventLoopGroup();
NioEventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.channel(NioServerSocketChannel.class);
serverBootstrap.group(boss, worker);
serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new ProcotolFrameDecoder());
ch.pipeline().addLast(new LoggingHandler());
ch.pipeline().addLast(new MessageCodecSharable());
ch.pipeline().addLast(new RpcRequestMessageHandler());
}
});
// 同步等待连接的建立
ChannelFuture future = serverBootstrap.bind().sync();
// 同步等待channel关闭
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
boss.shutdownGracefully();
worker.shutdownGracefully();
}
用到了两次Future，但是都是使用了同步的方法，下面研究sync方法如何同步
serverBootstrap.bind()返回DefaultChannelPromise，继续执行sync()时，实际调用的是DefaultChannelPromise父类DefaultPromise中的的sync方法
// DefaultPromisel类
public Promise<V> sync() throws InterruptedException {
await();
rethrowIfFailed();
return this;
}
public Promise<V> await() throws InterruptedException {
if (isDone()) {
return this;
}
if (Thread.interrupted()) {
throw new InterruptedException(toString());
}
checkDeadLock();
synchronized (this) {
while (!isDone()) {
incWaiters();
try {
wait();
} finally {
decWaiters();
}
}
}
return this;
}
所以当执行sync时当前线程会被DefaultPromisel实现类这个对象锁锁住，那么什么时候 唤醒 呢？
最后返回的是PendingRegistrationPromise类对象，该类是DefaultPromise的孙子类。 regFuture是关于连接channel的future，并且向channel注册了一个listener，当触发了operationComplete事件之后调用这个listener，在这个listener中会置promise的状态，修改状态的代码在doBind0() 中， 最终会使用cas修改promise中的状态，如果修改成功，那么会调用notifyAll方法唤醒所有因为当前对象锁阻塞的线程，也就使得主线程跳出了sync();
(operationComplete事件触发调用listener这部分较为复杂，详细可以读第二篇参考文章)
Promise
demo
public static void main(String[] args) throws ExecutionException, InterruptedException {
EventLoop next = new NioEventLoopGroup().next();
// 传入一个线程池
DefaultPromise<Integer> promise = new DefaultPromise<>(next);
new Thread(() -> {
try {
System.out.println("开始计算");
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
promise.setSuccess(1);
}).start();
Integer ans = promise.get();
System.out.println(ans);
}
Promise可以理解为装结果的容器 和Future不同的两点
- DefaultPromise是可以主动创建的，Future都是submit提交任务返回的
- DefaultPromise可以写入结果，通过setSuccess()或者setFail()
着重研究setSuccess()和setFailure() 对于get阻塞方法，还是用到了DefaultPromise的await()方法。 这两个方法通过cas把set的值赋给DefaultPromise对象中的result字段，如果cas操作成功，就会唤醒所有的阻塞线程，并且调用listener。
private static final AtomicReferenceFieldUpdater<DefaultPromise, Object> RESULT_UPDATER =
AtomicReferenceFieldUpdater.newUpdater(DefaultPromise.class, Object.class, "result");
private boolean setValue0(Object objResult) {
if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
if (checkNotifyWaiters()) {
notifyListeners();
}
return true;
}
return false;
}
结论
- JDK中的Future配合Callable提交任务，因为缺少了Listener机制所以主线程必须轮询判断是否执行完了或者阻塞等待获取值，所以看起来不算是异步编程
- netty Future继承自Jdk的Future并进行了扩展，最主要的是加入了注册Listener机制
- Promise是为了解决Future不能主动创建，不能主动写入这两个问题。、
- Future和Promise底层都用到了DefaultPromise这个实现，主要需要理解其两个内容：1、await()阻塞和唤醒 2、Listener的注册和执行 （第二个我还不是很明白）
参考文章