Netty

Future

2018-03-08  本文已影响0人  毛小力
Future

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);
}
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");
    }
}
上一篇下一篇

猜你喜欢

热点阅读