【杂谈】从实现角度看ChannelFuture

时间:2020-05-02 猫毛·波拿巴人气:3

JDK中的Future特性

在介绍Netty的ChannelFuture之前，我们先来看看JDK中的Future是如何实现的。总的来说就是任务提交的时候会使用装饰器模式，将任务包装成一个FutureTask。当执行器执行该Task的时候，不仅仅会执行用户提交的任务，还会执行装饰器添加的额外操作，例如在执行之前记录当前执行线程、执行完成后将任务结果保存在FutureTask对象内部等。

Thread runner => 装饰器添加的，在执行任务之前，会在对象内保存当前执行线程的引用，用于中断任务执行
Object outcome => 任务执行结果（返回值或异常对象），任务执行完成后会将结果set到此对象的outcome，后续可通过Future的get接口取出
Callable<V> callble => 用户提交的实际任务
WaitNode waiters => 用于保存等待线程，任务完成后会唤醒这些线程

详细请看本人过去整理的随笔：

揭开Future的神秘面纱——任务取消
揭开Future的神秘面纱——任务执行
揭开Future的神秘面纱——结果获取

Netty中的ChannelFuture

ChannelFuture是在Future基础上的完善，它支持添加监听器，在任务完成后自动执行相关操作。

这个其实就是观察者模式，个人认为就是在前面的C部分添加监听的方法调用，即执行线程执行完成后，如果发现该任务上有监听器，则该执行线程还会调用监听器接口。

话不多说，我们来看看它代码到底是怎么写的，跟Future的实现有何异同。相关实现关键内容在DefaultPromise类中，相对于前面的FutureTask。

1.对象属性

 private volatile Object result;
 private final EventExecutor executor;
 private Object listeners;
 private short waiters;
 private boolean notifyingListeners;

从对象属性中我们可以得知，该对象没有保存用户任务。实际上个人任务这是它跟Future最大的不同，Future所引用的本质上是一个任务，而ChannelFuture所引用的只有任务状态和任务结果。所以这个DefaultPromise对象不会被作为任务提交到执行器中。

2.任务完成时的线程唤醒与监听器触发

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;
    }

当任务执行完成，通过setValue将值传入DefaultPromise对象时，唤醒等待的线程，并触发监听器。

2.线程的等待与唤醒方式

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;
    }
private synchronized boolean checkNotifyWaiters() {
    if (waiters > 0) {
        notifyAll();
    }
    return listeners != null;
}

从上述代码可知，这里的线程等待与唤醒方式使用的内置的wait()和notify()方法，而FutureTask的等待队列是单独实现的。

注：这里尚不清楚这两种实现方式之间的优劣。

3.设置监听器

public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {
        checkNotNull(listener, "listener");

        synchronized (this) {
            addListener0(listener);
        }

        //添加监听器时，如果发现已经完成，则直接调用触发监听器
        if (isDone()) {
            notifyListeners();
        }

        return this;
    }

如果任务已经完成，则直接触发监听器。防止出现"调用setLisener的时候，任务已经完成，导致监听器不被触发"。

4.任务取消

public boolean cancel(boolean mayInterruptIfRunning) {
        if (RESULT_UPDATER.compareAndSet(this, null, CANCELLATION_CAUSE_HOLDER)) {
            if (checkNotifyWaiters()) {
                notifyListeners();
            }
            return true;
        }
        return false;
    }

从代码可以看出，它的实现与FutureTask有所不同，它并不会尝试调用执行线程的interrupt()方法来中断线程，只是将等待线程唤醒，并触发监听器。所以这个“取消”操作并不会影响代码的实际执行。

事实上“中断”也只是一种协作方式，它只是设置中断状态并将线程唤醒（如果该线程正处于挂起状态），如果用户代码中没有对中断状态进行判断，也没有使用wait()、sleep()等方法，中断操作也是不会实际“打断”代码执行的。

详细可看：【杂谈】线程中断——Interrupt

加载全部内容