Future

Future

• juc.Future：Java异步操作结果
• Future：Netty异步操作结果
• Promise：可设置结果的异步操作结果
• ProgressiveFuture：有进度的异步操作结果
• ChannelFuture：通道的异步操作结果

Java异步执行框架

将任务（Runnable或Callable）提交给执行器（Executor），执行器返回异步结果Future。

执行器随后安排线程执行任务，有3种完成状态：成功完成，失败完成，取消完成。

根据Future可以查询任务执行的状态和结果：

// 非阻塞查询状态：是否已完成（无论哪种完成状态）
boolean isDone();

// 取消执行
boolean cancel(boolean mayInterruptIfRunning);

// 非阻塞查询状态：是否已取消
boolean isCancelled();

// 阻塞查询结果
// 成功完成：返回结果V
// 失败完成：抛出失败原因ExecutionException
// 取消完成：抛出CancellationException
// 查询线程被中断：抛出InterruptedException
V get() throws InterruptedException, ExecutionException;

// 超时阻塞查询结果
// 阻塞超时：TimeoutException
V get(long timeout, TimeUnit unit) throws InterruptedException, 
    ExecutionException, TimeoutException;

执行器是如何将完成结果传递给Future的呢？

提交任务时，执行器实际返回的是一个FutureTask对象，FutureTask同时继承了Runnable和Future。

public class FutureTask<V> implements RunnableFuture<V>;
public interface RunnableFuture<V> extends Runnable, Future<V>;

执行器将执行的任务和返回的异步结果包装成同一个对象，FutureTask执行完任务后顺带就设置了状态和结果。

image.png

Future

异步通知结果的juc.Future。

Java异步框架中任务的执行是异步的，但是查询结果仍然是同步的。

Future增加了监听器机制，任务执行完成后，异步将结果通知给监听器。

public interface Future<V> extends java.util.concurrent.Future<V> {
    // 查询状态

    // 是否成功完成
    boolean isSuccess();
    // 失败完成原因
    Throwable cause();
    // 是否能取消
    boolean isCancellable();

    // 同步获取结果

    // 非阻塞获取结果，未完成返回null
    V getNow();

    // 阻塞等待操作完成，抛出失败完成异常，可响应中断
    Future<V> sync() throws InterruptedException;
    // 阻塞等待操作完成，抛出失败完成异常，不响应中断
    Future<V> syncUninterruptibly();
    
    // 阻塞等待操作完成，不抛失败完成异常，可响应中断
    Future<V> await() throws InterruptedException;
    // 同上，超时阻塞
    boolean await(long timeoutMillis) throws InterruptedException;

    // 阻塞等待操作完成，不抛失败完成异常，不响应中断
    Future<V> awaitUninterruptibly();
    // 同上，超时阻塞
    boolean awaitUninterruptibly(long timeoutMillis);

    // 异步通知结果

    // 添加监听器
    // 任务完成后isDone()，监听器将得到通知；如果添加时任务已经完成，则立即得到通知。
    Future<V> addListener(GenericFutureListener<? extends Future<? super V>> listener);
    Future<V> addListeners(GenericFutureListener<? extends Future<? super V>>... listeners);

    // 删除监听器
    Future<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener);
    Future<V> removeListeners(GenericFutureListener<? extends Future<? super V>>... listeners);
}

• GenericFutureListener

public interface GenericFutureListener<F extends Future<?>> extends EventListener {

    // Future完成后调用
    // future：调用此回调的Future
    void operationComplete(F future) throws Exception;
}

Promise

外部设置结果的Future。

juc.Future是通过实现类在内部设置结果，Promise提供接口在外部设置结果。

public interface Promise<V> extends Future<V> {

    // 设置成功完成结果并通知所有监听器
    // 如果已经成功或失败则抛出IllegalStateException
    Promise<V> setSuccess(V result);

    // 尝试设置成功完成结果并通知所有监听器
    // 如果已经成功或失败则返回false
    boolean trySuccess(V result);

    // 设置失败完成结果并通知所有监听器
    // 如果已经成功或失败则抛出IllegalStateException
    Promise<V> setFailure(Throwable cause);

    // 尝试设置失败完成结果并通知所有监听器
    // 如果已经成功或失败则返回false
    boolean tryFailure(Throwable cause);

    // 设置成不可取消
    // 如果已经取消则返回false
    boolean setUncancellable();
}

ProgressivePromise

有进度的Promosie

public interface ProgressivePromise<V> extends Promise<V>, ProgressiveFuture<V> {

    // 设置操作的进度，通知GenericProgressiveFutureListener监听器
    ProgressivePromise<V> setProgress(long progress, long total);
    
    // 如果已经完成或超出范围，则不作为，返回false
    boolean tryProgress(long progress, long total);
}

ChannelFuture

通道异步IO的结果。

Netty中所有IO操作都是异步，IO调用会立即返回，不管是否完成。返回值是一个ChannelFuture实例，通过它可以获取IO操作的状态和结果。

ChannelFuture对象在IO操作开始时创建，初始为未完成状态。依据IO操作结束的原因，ChannelFuture完成状态有三种：成功完成、失败完成、取消完成。

await()方法可以阻塞式等待IO操作完成，但更推荐使用非阻塞式addListener(GenericFutureListener)方法，IO操作完成时会通知监听器执行后续操作。

在ChannelHandler内不要调用await()。因为拦截器方法通常由IO线程调用，如果await()在拦截器方法内被调用，即由IO线程调用，则IO线程自己等待自己，形成死锁。

等待超时和IO超时是不一样的。IO超时是失败完成；等待超时是抛出TimeoutException。

public interface ChannelFuture extends Future<Void> {
    // 聚合通道对象
    Channel channel();
}

DefaultPromise

Netty异步执行的主要实现类。

• 设置结果
• 提供状态和结果的查询
• 阻塞和唤醒查询线程
• 异步通知监听器：普通、进度

public class DefaultPromise<V> extends AbstractFuture<V> implements Promise<V> {
    // 结果
    private volatile Object result;

    // 成功完成但结果为空时的占位对象
    private static final Object SUCCESS;
    // 失败完成设置的异常对象不能为空，无需占位
    // 取消完成的占位对象
    private static final CauseHolder CANCELLATION_CAUSE_HOLDER;
    // 设置不可取消的占位对象
    private static final Object UNCANCELLABL;

    // 利用Object对象的wait/notify机制实现查询线程的阻塞和唤醒

    // 监听器列表
    private Object listeners;
    // 负责通知监听器的线程
    private final EventExecutor executor;
    
    // 普通Promise通知监听器：调用监听器的operationComplete()方法
    notifyListener0(this, listener);
    void notifyListener0(Future future, GenericFutureListener l) {
        try {
            l.operationComplete(future);
        } catch (Throwable t) {
            logger.warn("...", t);
        }
    }

    // 进度Promise通知监听器：调用监听器的operationProgressed()方法
    notifyProgressiveListener0(this, listener, progress, total);
    void notifyProgressiveListener0(ProgressiveFuture future, 
        GenericProgressiveFutureListener l, long progress, long total) {
        try {
            l.operationProgressed(future, progress, total);
        } catch (Throwable t) {
            logger.warn("...", t);
        }
    }

}

DefaultChannelPromise

聚合Channel对象的Promise。

public class DefaultChannelPromise extends DefaultPromise<Void> 
    implements ChannelPromise {

    private final Channel channel;
    public DefaultChannelPromise(Channel channel) {
        this.channel = checkNotNull(channel, "channel");
    }
}